AgentSkillsCN

agentic-workflows

构建具备实时流式可视化、结构化输出与多智能体协作能力的生产级代理式 AI 系统。涵盖 Anthropic、OpenAI、vLLM 的 SDK,支持用于智能体互操作的 A2A 协议,集成 Pydantic 校验机制,利用 LangGraph 的检查点功能实现工作流的无缝恢复,搭配向量数据库内存(Pinecone/Chroma/FAISS),并配备防幻觉防护机制。适用于构建 AI 代理、多智能体系统、工具调用工作流,或需要将流式智能体推理结果实时呈现至 UI 的各类应用。

SKILL.md
--- frontmatter
name: agentic-workflows
description: Build production-grade agentic AI systems with real-time streaming visibility, structured outputs, and multi-agent collaboration. Covers Anthropic/OpenAI/vLLM SDKs, A2A protocol for agent interoperability, Pydantic validation, LangGraph checkpointing for workflow resumption, vector DB memory (Pinecone/Chroma/FAISS), and guardrails for anti-hallucination. Use when building AI agents, multi-agent systems, tool-calling workflows, or applications requiring streaming agent reasoning to UI.

Agentic Workflows Skill

Build intelligent, observable, and resilient AI agent systems.

Architecture Decision Flow

code
New Agent System Request
           │
           ▼
┌──────────────────────────┐
│ Single task or multi-step?│
│ Single → Simple LLM call │
│ Multi-step → Agent loop  │
└──────────────────────────┘
           │
           ▼
┌──────────────────────────┐
│ Need multiple specialists?│
│ Yes → Multi-agent (A2A)  │
│ No → Single agent        │
└──────────────────────────┘
           │
           ▼
┌──────────────────────────┐
│ Long-running/resumable?   │
│ Yes → LangGraph + checkpoint│
│ No → Simple agent loop   │
└──────────────────────────┘
           │
           ▼
┌──────────────────────────┐
│ Need memory across sessions?│
│ Yes → Vector DB          │
│ No → In-session state    │
└──────────────────────────┘

Provider Selection

ProviderBest ForStreamingTools
Anthropic ClaudeComplex reasoning, extended thinkingSSENative
OpenAI GPT-4General purpose, function callingSSENative
vLLMSelf-hosted, cost controlOpenAI-compatibleVia prompts

Quick Start Patterns

Anthropic Streaming with Tools

python
import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-5",
    max_tokens=4096,
    tools=[{"name": "search", "description": "Search the web", "input_schema": {...}}],
    messages=[{"role": "user", "content": "Research AI trends"}]
) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            if hasattr(event.delta, "text"):
                print(event.delta.text, end="", flush=True)
            elif hasattr(event.delta, "thinking"):
                print(f"[Thinking] {event.delta.thinking}")

Structured Output with Pydantic

python
import instructor
from pydantic import BaseModel

class Analysis(BaseModel):
    summary: str
    confidence: float
    sources: list[str]

client = instructor.from_provider("anthropic/claude-sonnet-4-5")
result = client.create(
    response_model=Analysis,
    messages=[{"role": "user", "content": "Analyze market trends"}],
    max_retries=3
)

Reference Documentation

TaskReference File
Anthropic/OpenAI/vLLM SDK patternsreferences/llm-sdks.md
Multi-agent with A2A protocolreferences/multi-agent.md
Streaming to UI (SSE/WebSocket)references/streaming.md
Pydantic structured outputsreferences/structured-outputs.md
Memory with vector DBsreferences/memory.md
Checkpointing & resumptionreferences/checkpointing.md
Guardrails & anti-hallucinationreferences/guardrails.md

When to Use Multi-Agent

ScenarioApproach
Different expertise neededMulti-agent with specialists
Verification requiredDebate pattern (critic agent)
Complex workflow orchestrationSupervisor + workers
Simple tool useSingle agent with tools
Independent subtasksParallel agents

Production Checklist

  • Structured outputs with Pydantic validation
  • Retry logic with exponential backoff
  • Streaming to UI for visibility
  • Checkpointing for long-running workflows
  • Guardrails for input/output validation
  • Memory persistence (vector DB or KV store)
  • Error handling with graceful degradation
  • Observability (logging, tracing)