Qi

English Title: Agent External Integration — RESTful APIs, OAuth 2.0 & Webhook Handling

Function Calling 让模型「知道该调什么工具」，但真正把 Agent 接到企业系统里，靠的是 HTTP API 集成：用 REST 拉取业务数据、用 OAuth 代表用户访问 SaaS、用 Webhook 接收异步事件。本文是系列第 11 篇，承接 Function Calling / Tool Use 的工具契约，向下衔接 Docker 与基础 DevOps 的部署与密钥注入。

0. 30 秒心智模型

1
2
3
4
5

用户意图 → LLM 选 Tool → API Wrapper（REST / OAuth）
                              ↓
                    外部系统（CRM / 工单 / 日历）
                              ↓
              Webhook 推送事件 → 验签 → 入队 → Agent 续跑

面试官与架构师常问的三条线：同步调用怎么稳、授权怎么续期、被动事件怎么可信。下面按此展开。

1. 为什么 Agent 必须做 API 集成

大模型本身没有你的客户名单、库存或审批流。Agent 的价值在于 在推理环中读写真实世界：

与 MCP 协议 的关系：MCP 标准化「发现工具 + 调用工具」的传输层；底层仍常是 REST。你可以把 Agent-friendly API Wrapper 同时暴露为 MCP Tool 与 LangChain @tool，业务 HTTP 逻辑只写一份。

工程原则： 模型只接触 窄接口、强类型、可审计 的 Wrapper，而不是把原始 OpenAPI 全文塞进 Prompt。

从 主流模型 API 调用实战 到本篇，差别在于：前者是 你主动请求 LLM，后者是 Agent 主动请求你的业务系统。两者都要管 timeout、重试与用量，但业务 API 往往还有 租户隔离、合规审计、写操作幂等 等额外约束——这些不应交给模型「临场发挥」，而应在 Wrapper 层写死策略。

2. RESTful API 调用模式

2.1 客户端选型：httpx 异步优先

Agent 服务多为 FastAPI / asyncio；httpx 同时支持 sync / async，连接池可复用，比逐请求 requests 更省延迟。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

import httpx
from typing import Any

class CRMClient:
    def __init__(self, base_url: str, api_key: str):
        self._client = httpx.AsyncClient(
            base_url=base_url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=httpx.Timeout(10.0, connect=5.0),
        )

    async def get_contact(self, contact_id: str) -> dict[str, Any]:
        r = await self._client.get(f"/v1/contacts/{contact_id}")
        r.raise_for_status()
        return r.json()

    async def aclose(self) -> None:
        await self._client.aclose()

2.2 重试与退避

对 429 / 502 / 503 与网络抖动应重试；对 4xx（除 429） 一般不重试，把错误转成 Tool Observation 让模型改参。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

import asyncio
import httpx

async def request_with_retry(
    client: httpx.AsyncClient,
    method: str,
    url: str,
    *,
    max_attempts: int = 4,
    **kwargs,
) -> httpx.Response:
    delay = 0.5
    for attempt in range(max_attempts):
        try:
            resp = await client.request(method, url, **kwargs)
            if resp.status_code in (429, 502, 503):
                retry_after = float(resp.headers.get("Retry-After", delay))
                await asyncio.sleep(retry_after)
                delay = min(delay * 2, 8.0)
                continue
            return resp
        except (httpx.TimeoutException, httpx.NetworkError):
            if attempt == max_attempts - 1:
                raise
            await asyncio.sleep(delay)
            delay *= 2
    raise RuntimeError("unreachable")

2.3 限流（Rate Limit）

Agent 可能在单轮对话中 连续多次 调同一 API。需在 Wrapper 层做令牌桶或分布式限流（Redis），避免打满厂商配额导致全站 429。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

import time
from collections import deque

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        self.rate, self.capacity = rate, capacity
        self.tokens = float(capacity)
        self.updated = time.monotonic()

    def acquire(self) -> None:
        now = time.monotonic()
        self.tokens = min(self.capacity, self.tokens + (now - self.updated) * self.rate)
        self.updated = now
        if self.tokens < 1:
            time.sleep((1 - self.tokens) / self.rate)
            self.tokens = 0
        else:
            self.tokens -= 1

面试要点： 区分 客户端重试 与 服务端幂等——POST 创建资源应带 Idempotency-Key 头，防止重试产生重复工单。

3. OAuth 2.0：Agent 工具如何拿令牌

SaaS（Google、GitHub、Salesforce）普遍要求 用户授权 后，后台用 refresh_token 换 access_token。Agent 不应把长期 refresh_token 放进 LLM 上下文，而应存在密钥库，由 Tool 运行时读取。

3.1 授权码流程（一次性）

1. 引导用户打开 authorize_url（scope 最小化）。
2. 回调接收 code，服务端 POST /token 换 access_token + refresh_token。
3. 将 refresh_token 加密存入 DB / Vault，绑定 user_id。

3.2 运行时刷新

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

import os
import time
import httpx

class OAuthTokenStore:
    def __init__(self):
        self._cache: dict[str, tuple[str, float]] = {}  # user_id -> (access, exp)

    async def get_access_token(self, user_id: str) -> str:
        access, exp = self._cache.get(user_id, ("", 0))
        if time.time() < exp - 60:
            return access
        return await self._refresh(user_id)

    async def _refresh(self, user_id: str) -> str:
        # 从 DB 读取 refresh_token（示例略）
        refresh_token = os.environ[f"REFRESH_{user_id}"]
        async with httpx.AsyncClient() as client:
            r = await client.post(
                "https://oauth2.googleapis.com/token",
                data={
                    "grant_type": "refresh_token",
                    "refresh_token": refresh_token,
                    "client_id": os.environ["OAUTH_CLIENT_ID"],
                    "client_secret": os.environ["OAUTH_CLIENT_SECRET"],
                },
            )
            r.raise_for_status()
            data = r.json()
        access = data["access_token"]
        self._cache[user_id] = (access, time.time() + data["expires_in"])
        return access

Agent 设计建议：

• Tool 参数只接受 业务 ID（如 calendar_id），令牌由 user_id 从 Session 解析。
• scope 按工具拆分：读日历只需 calendar.readonly，禁止默认申请 drive.full。
• 令牌刷新失败时返回明确 Observation：「授权已过期，请重新连接 Google 账号」。

4. Webhook：异步事件与验签

Webhook 是 服务器推、Agent 拉 的反面：外部系统在事件发生时 POST 你的 URL。典型用于：支付成功、PR 合并、工单状态变更。

4.1 处理流水线

1
2

POST /webhooks/github → 验签 → 解析 payload → 写入队列
        → Worker 消费 → 触发 Agent（新 thread 或续跑 checkpoint）

务必快速返回 2xx（如 202），重逻辑放队列；否则对方会重试，造成重复执行。

4.2 签名验证（GitHub 示例）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

import hmac
import hashlib
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()
WEBHOOK_SECRET = b"your-webhook-secret"  # 来自环境变量 / Secret Manager

@app.post("/webhooks/github")
async def github_webhook(request: Request):
    body = await request.body()
    sig = request.headers.get("X-Hub-Signature-256", "")
    expected = "sha256=" + hmac.new(WEBHOOK_SECRET, body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sig, expected):
        raise HTTPException(status_code=401, detail="invalid signature")

    event = request.headers.get("X-GitHub-Event")
    payload = await request.json()
    # await queue.publish({"event": event, "payload": payload})
    return {"ok": True}

4.3 幂等与去重

用 X-GitHub-Delivery 或业务 event_id 在 Redis 做 SET NX + TTL，防止重放。Agent 侧把「同一 PR 关闭」只处理一次，避免重复 @客户。

5. 设计 Agent 友好的 API Wrapper（作为 Tool）

好的 Tool 是 意图级 API，不是 OpenAPI 的机械映射。

1
2
3
4
5
6
7
8
9
10
11
12
13

from pydantic import BaseModel, Field
from langchain_core.tools import tool

class CreateTicketInput(BaseModel):
    title: str = Field(..., description="工单标题，50 字以内")
    priority: str = Field("normal", description="low | normal | high")

@tool(args_schema=CreateTicketInput)
async def create_support_ticket(title: str, priority: str = "normal") -> str:
    """当用户明确要求创建工单或投诉未解决时调用。成功返回单号。"""
    # client = get_crm_client_from_context()
    # ticket = await client.create_ticket(title=title, priority=priority)
    return "TICKET-2026-8842"  # 示例

与 Function Calling 衔接：描述写清 何时调用、必填字段、失败语义；参数用 Pydantic 约束，减少幻觉参数。

6. 安全：密钥与 Scope

1. 密钥不进 Prompt、不进 Git：本地用 .env，生产用 K8s Secret / Vault；CI 用 OIDC 而非长期 API Key。
2. 最小权限：REST 用只读 Key 做查询 Tool；写操作单独 Tool + 人工审批（HITL）。
3. 出站 SSRF 防护：禁止模型通过 Tool 指定任意 URL；Wrapper 白名单 base_url。
4. 审计：记录 user_id、tool_name、请求 ID、响应码；敏感字段脱敏后再写入 LangSmith Trace。
5. 多租户隔离：OAuth token、Webhook 路由按 tenant 分表，防止 A 客户事件触发 B 的 Agent。

部署层密钥注入、网络策略与镜像扫描见下一篇 Docker 与基础 DevOps。

7. 综合示例：FastAPI + Tool + Webhook

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# app/main.py — 最小骨架（示意）
import os
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI
from langchain_core.tools import tool

crm: httpx.AsyncClient | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global crm
    crm = httpx.AsyncClient(
        base_url="https://api.example.com",
        headers={"Authorization": f"Bearer {os.getenv('CRM_API_KEY')}"},
    )
    yield
    await crm.aclose()

app = FastAPI(lifespan=lifespan)

@tool
async def lookup_order(order_id: str) -> str:
    """查询物流状态。order_id 为订单号。"""
    assert crm is not None
    r = await crm.get(f"/orders/{order_id}")
    if r.status_code == 404:
        return "未找到订单，请核对单号。"
    r.raise_for_status()
    data = r.json()
    return f"订单 {order_id}：{data['status']}，预计 {data.get('eta', '未知')}"

# Agent 路由：POST /chat → Runner → lookup_order
# Webhook 路由：POST /webhooks/payment → 验签 → 若 paid 则 enqueue 续聊

生产环境应拆分为：API Gateway（鉴权、限流）、Agent Worker、Webhook Ingest 三个进程，避免 Webhook 流量拖垮对话接口。

若团队已采用 CrewAI / AutoGen 多 Agent 做角色分工，建议把 所有 HTTP 调用收敛到「工具专家」Agent 的 Tool 集，其它角色只通过消息传递业务结论，避免多个 Agent 各自持有一份 API Key，难以轮换与审计。

8. 常见陷阱与面试速记

Q：Agent 直接调 REST 和走 MCP 怎么选？
对外部生态、多客户端复用选 MCP；对单一后端、强定制逻辑，REST Wrapper + @tool 更简单。二者可共存。

Q：Webhook 如何驱动「长时 Agent」？
事件只负责 入队 + 唤醒；状态用 thread_id 与 Checkpoint 恢复，不在 Webhook 进程里跑完整 ReAct 循环。

Q：同步 REST 与 Streaming 混用？
对 LLM 用 SSE；对业务 API 仍是一次性 JSON。不要在 Tool 里对 REST 做 token 级 stream 解析——除非厂商明确支持 NDJSON 且你有背压控制，否则 Observation 难以在 ReAct 一轮内闭合。

9. 小结

API 集成是 Agent 的「手脚」：REST + httpx 负责同步读写，重试与限流 保证稳定性；OAuth 负责代表用户访问 SaaS，refresh 逻辑 必须远离模型上下文；Webhook + 验签 + 幂等 负责可信的异步触发。把 HTTP 细节封进 窄 Tool，模型只处理业务语义，才能同时满足安全、成本与可维护性。

完成本篇后，建议继续 Docker 与基础 DevOps，把 API Key、OAuth Client Secret 与 Webhook Secret 纳入镜像与编排的最佳实践。

系列导航

• 上一篇：Function Calling / Tool Use
• 下一篇：Docker 与基础 DevOps

English Title: Function Calling Deep Dive — Tool Schema Design, Parallel Calls & Error Handling

MCP 把工具暴露成标准协议之后，模型侧如何「选中工具、填好参数、消化结果」仍是 Agent 落地的核心。Function Calling（也称 Tool Use）不是让 LLM 直接执行代码，而是让模型输出结构化调用意图，由你的运行时真正执行并回传结果。本文从闭环流程、JSON Schema 设计、错误重试、并行调用、结果回灌到 OpenAI / Claude / Gemini 差异，给出可上线的 Python 示例，衔接系列中的 MCP 与 API 集成专题。

After MCP standardizes tool exposure, the model still must select tools, fill parameters, and consume results. Function Calling lets the LLM emit structured call intents while your runtime executes them. This article covers the full loop, schema design, retries, parallelism, and provider differences.

1. Function Calling 如何工作 | The Agent Loop

一次完整的工具调用闭环可以概括为四步：

1

Model → tool_call(s) → Execute → tool_result → Model → …

关键认知： 模型是「调度员」，不是「执行器」。它根据 tools 定义里的 description 与 parameters（JSON Schema）推断该调哪个函数；你注册的真实 Python/HTTP 函数才接触生产数据。多轮 Agent 就是在 messages 数组末尾不断追加 assistant（含 tool_calls）与 tool（含 result），直到模型不再请求工具、只返回最终文本。

典型消息序列如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

messages = [
    {"role": "system", "content": "你是助手，可用天气与搜索工具。"},
    {"role": "user", "content": "北京今天天气怎样？"},
    # 模型返回 assistant，带 tool_calls
    {
        "role": "assistant",
        "content": None,
        "tool_calls": [{
            "id": "call_abc",
            "type": "function",
            "function": {"name": "get_weather", "arguments": '{"city": "北京"}'},
        }],
    },
    # 你执行后回灌
    {
        "role": "tool",
        "tool_call_id": "call_abc",
        "content": '{"temp_c": 28, "condition": "晴"}',
    },
]
# 再次 chat.completions.create(messages=messages, tools=tools)

2. JSON Schema 参数设计 | Tool Parameter Design

tools[].function.parameters 遵循 JSON Schema 子集。设计质量直接决定模型能否一次填对参数。

推荐实践：

1. name — 动词 + 名词，如 search_documents、create_ticket，避免 do_stuff
2. description — 写清「何时用、何时不用、边界」；这是模型选工具的第一信号
3. 必填字段 — 用 required: ["query"]，减少漏填
4. 枚举约束 — 对固定选项用 enum，比自由字符串更稳
5. 控制粒度 — 宁可多个小工具，也不要一个「万能」工具塞满可选参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

tools = [{
    "type": "function",
    "function": {
        "name": "search_kb",
        "description": "在用户问题涉及产品文档、API 说明时检索知识库。不用于闲聊或实时新闻。",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "检索关键词，尽量保留用户原意",
                },
                "top_k": {
                    "type": "integer",
                    "description": "返回条数，默认 5",
                    "minimum": 1,
                    "maximum": 20,
                },
            },
            "required": ["query"],
            "additionalProperties": False,
        },
    },
}]

常见陷阱： arguments 在 API 里是字符串化的 JSON，必须先 json.loads 再校验；Schema 过于复杂（深层 oneOf）会降低填参成功率；字段名与业务代码不一致会导致静默失败——建议在执行前用 Pydantic 做二次校验。

3. 错误处理与重试 | Error Handling & Retries

工具层错误分三类，处理策略应不同：

不要把堆栈直接丢给模型——用简短、可行动的 JSON：

1
2
3
4
5
6
7
8

def run_tool(name: str, args: dict) -> str:
    try:
        result = TOOL_REGISTRY[name](**args)
        return json.dumps(result, ensure_ascii=False)
    except ValueError as e:
        return json.dumps({"error": "invalid_args", "message": str(e)})
    except Exception:
        return json.dumps({"error": "internal", "message": "工具暂时不可用，请稍后重试"})

重试层次：

• HTTP 层 — 对 LLM API 的 429/5xx 重试（与上一篇 API 指南一致）
• 工具层 — 幂等读操作可重试 2–3 次；写操作慎用自动重试
• Agent 层 — 同一 tool_call_id 只回灌一次结果；若模型重复请求相同调用，可在运行时做去重或缓存

若连续多轮工具失败，应设置 max_tool_rounds 上限，避免无限循环烧 Token。

4. 并行工具调用 | Parallel Tool Calls

现代模型（如 GPT-4o、Claude 3.5+）常在一次 assistant 消息中返回多个 tool_call，且彼此无依赖——例如同时查天气与搜新闻。你的执行器应：

1. 解析 message.tool_calls 列表
2. 并行执行（asyncio.gather 或线程池）
3. 按相同 tool_call_id 逐条回灌 role: tool 消息
4. 全部结果就绪后，再发起下一轮 LLM 请求

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

import asyncio
import json
from openai import OpenAI

client = OpenAI()

async def dispatch_tool(call):
    name = call.function.name
    args = json.loads(call.function.arguments)
    if name == "get_weather":
        return {"temp_c": 25}
    if name == "web_search":
        return {"items": ["..."]}
    raise ValueError(f"unknown tool: {name}")

async def handle_tool_calls(assistant_msg):
    tasks = [dispatch_tool(tc) for tc in assistant_msg.tool_calls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    tool_messages = []
    for call, res in zip(assistant_msg.tool_calls, results):
        if isinstance(res, Exception):
            content = json.dumps({"error": str(res)})
        else:
            content = json.dumps(res, ensure_ascii=False)
        tool_messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": content,
        })
    return tool_messages

注意： 有依赖关系的调用（先查用户 ID 再查订单）不应依赖模型并行——应在 Schema 层拆成顺序工具，或用编排层（LangGraph 等）显式控制。并行只适用于「彼此独立」的子任务。

5. 结果解析与回灌 | Parsing & Feeding Back

执行结果回灌时需遵守各厂商约定，否则下一轮请求会 400：

• OpenAI 兼容 — 每条 tool 消息必须带 tool_call_id，与 assistant 里 tool_calls[].id 一一对应；content 建议为字符串（JSON 文本即可）
• 顺序 — 先 append 带 tool_calls 的 assistant，再 append 所有 tool 消息，不要穿插 user
• 体积 — 大段检索结果应截断或摘要后再回灌，避免撑爆上下文；可只保留 title + snippet 前 N 条

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

def agent_turn(messages, tools):
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=tools,
    )
    msg = resp.choices[0].message
    messages.append(msg.model_dump(exclude_none=True))

    if not msg.tool_calls:
        return msg.content  # 最终答案

    for call in msg.tool_calls:
        args = json.loads(call.function.arguments)
        raw = TOOL_REGISTRY[call.function.name](**args)
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": json.dumps(raw, ensure_ascii=False),
        })
    return agent_turn(messages, tools)  # 递归直至无 tool_calls

解析技巧： 对模型返回的 arguments 做宽松解析（尾随逗号、单引号）可提升鲁棒性，但应在日志中记录原始字符串便于排错。若模型返回了未注册的工具名，回灌 {"error": "unknown_tool"} 比直接抛异常更能引导自修正。

6. 厂商差异 | OpenAI vs Claude vs Gemini

Claude 把工具结果放在 user 角色里，且需 tool_use_id 关联；Gemini 则在同一次 generateContent 的 parts 数组里交替 functionCall 与 functionResponse。若你做统一 Provider 抽象，建议在内部归一化为：

1
2
3
4
5
6
7
8
9
10

@dataclass
class ToolInvocation:
    id: str
    name: str
    arguments: dict

@dataclass  
class ToolResult:
    id: str
    content: str

上层 Agent 只处理 ToolInvocation / ToolResult，底层适配各 SDK 差异。DeepSeek、通义千问等 OpenAI 兼容端可直接复用 openai 客户端，仅改 base_url。

7. 完整 Python 示例 | Runnable Example

下面是一个最小可运行的「天气 + 计算」双工具 Agent（同步版，便于理解闭环）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

import json
from openai import OpenAI

client = OpenAI()

def get_weather(city: str) -> dict:
    return {"city": city, "temp_c": 26, "condition": "多云"}

def calc(expression: str) -> dict:
    # 生产环境请用安全表达式解析器，勿直接 eval
    return {"result": eval(expression, {"__builtins__": {}}, {})}

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市当前天气",
            "parameters": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "calc",
            "description": "计算数学表达式，如 (3+5)*2",
            "parameters": {
                "type": "object",
                "properties": {"expression": {"type": "string"}},
                "required": ["expression"],
            },
        },
    },
]

REGISTRY = {"get_weather": get_weather, "calc": calc}

def run_agent(user_input: str, max_rounds: int = 5) -> str:
    messages = [
        {"role": "system", "content": "你是助手，按需调用工具。"},
        {"role": "user", "content": user_input},
    ]
    for _ in range(max_rounds):
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=TOOLS,
        )
        msg = resp.choices[0].message
        messages.append(msg.model_dump(exclude_none=True))
        if not msg.tool_calls:
            return msg.content or ""
        for call in msg.tool_calls:
            fn = REGISTRY[call.function.name]
            args = json.loads(call.function.arguments)
            out = fn(**args)
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": json.dumps(out, ensure_ascii=False),
            })
    return "超过最大工具轮次"

if __name__ == "__main__":
    print(run_agent("上海天气如何？另外算一下 (12+8)*3"))

8. 实战要点 | Production Tips

1. 工具幂等 — 写操作带 idempotency_key，防止模型重试导致重复下单
2. 审计日志 — 记录每次 tool_call 的 name、args、latency、success，便于回放与合规
3. 人机确认 — 删数据、转账类工具在执行前插入 HITL 审批，不要全自动
4. 与 MCP 的关系 — MCP Server 暴露能力，Function Calling 是模型侧的「遥控器」；二者常组合：MCP 提供工具清单，LLM 通过 tools 数组选择调用（见上一篇 MCP 专题）
5. 测试 — 用固定 messages fixture 测 Schema 校验与错误回灌，而非只测最终自然语言

9. 总结 | Conclusion

Function Calling 的本质是结构化意图 + 运行时执行 + 结果回灌的循环。Schema 写得越清晰，并行与错误处理越规范，Agent 就越稳定。厂商 API 表面不同，但心智模型一致：把工具当函数签名暴露给模型，把执行权握在自己手里。掌握本文后，你已能搭建「能选工具、能并行、能容错」的 Tool Use 层；下一步是把工具背后的 REST/OAuth/Webhook 接到真实业务系统。

系列导航 Series Navigation：

• 上一篇：MCP 协议与 Server 开发
• 下一篇：API 集成（REST/OAuth/Webhook）

English Title: MCP in Practice — Connecting Agents to External Tools via Model Context Protocol

多 Agent 框架解决了「谁来做」，但 Agent 仍要对接数据库、工单系统、Git、Notion 等外部能力。过去每家 IDE 各自写插件、每家框架各自封装 Tool，集成成本重复且不可移植。Model Context Protocol（MCP） 由 Anthropic 提出并开源，用统一的 JSON-RPC 语义描述「能读什么、能调什么、能注入什么提示模板」，让 Host（宿主应用） 通过 Client 连接任意 Server，一次实现、多处复用。2026 年 Cursor、Claude Desktop、LangChain 等已原生或半原生支持 MCP，它正在成为 Agent 工具层的「USB-C」。

1. 什么是 MCP，为何成为 2026 事实标准

MCP 不是又一个 Agent 框架，而是 宿主与工具之间的协议层。它解决三类痛点：

与 OpenAI 的 Function Calling 相比：Function Calling 定义的是「模型在一次补全里如何声明调用」；MCP 定义的是「进程外 的能力如何被发现、鉴权、执行与回传」。二者互补——Host 常把 MCP Tool 映射为模型侧的 function，但 MCP 还额外标准化了资源读取与可复用 Prompt 模板。

2026 年 MCP 成为主流的原因很务实：供应链统一（社区已有 GitHub、Postgres、Slack 等 Server）、安全边界清晰（Server 独立进程、可限权）、厂商共建（Anthropic 规范 + 多 Host 实现）。当你要为团队内部系统开放给 Cursor/Claude 时，优先写 MCP Server 往往比为每个客户端各写一套插件更划算。

若你已在用 主流模型 API 的 tools 字段，可以把 MCP 理解为 把 Tool 实现从应用进程里抽出去：Host 只负责把 tools/list 映射进模型请求，真正执行发生在 Server 进程。这样换模型供应商时，业务集成层不必重写。

2. 架构：Host、Client、Server

1
2
3
4
5
6
7

┌─────────────┐     MCP      ┌─────────────┐     业务 API    ┌──────────────┐
│    Host     │ ◄──────────► │ MCP Client  │ ◄──────────────► │ MCP Server   │
│ Cursor/IDE  │  JSON-RPC   │ (内置/库)   │  stdio / HTTP   │ Git/DB/...   │
└─────────────┘             └─────────────┘                 └──────────────┘
       │
       ▼
   LLM（Claude/GPT 等）

• Host：面向用户的应用（Cursor、Claude Desktop、自定义 Agent 服务）。负责会话、模型调用、把 MCP 能力呈现给 LLM。
• Client：Host 内的协议实现，维护与 Server 的连接、能力发现（tools/list、resources/list）、调用转发。
• Server：暴露具体能力的最小单元，通常独立进程。通过 stdio（本地子进程）或 HTTP/SSE（远程服务）与 Client 通信。

一次典型交互：initialize 握手 → tools/list 发现工具 → 模型决定调用 → tools/call 执行 → 结果作为 tool 消息回到 Host。Resources 走 resources/read，不必经过模型的 function 通道，适合注入大段只读上下文。

传输选型：本地开发首选 stdio——Host 以子进程启动 Server，无网络暴露，调试简单。团队共享或 SaaS 化时用 Streamable HTTP / SSE，便于水平扩展与集中鉴权，但需额外处理连接保活与版本兼容。同一业务可同时提供两种 Transport，由部署环境选择。

3. 三大原语：Resources、Tools、Prompts

Resources 适合「让模型看见」：文档、配置、表结构。URI 与 MIME 类型由 Server 声明，Host 可按需拉取，避免把整个仓库塞进 system prompt。

Tools 适合「让模型做事」：每个 Tool 有 name、description、inputSchema（JSON Schema）。描述质量直接影响模型选工具的准确率——与系列 Prompt Engineering 中的工具边界写法一致。

Prompts 适合「标准化工作流」：把反复使用的评审、迁移、排障模板注册到 Server，团队共享同一套指令骨架，减少各项目复制粘贴 system prompt。

4. 动手写一个 MCP Server

4.1 Python（FastMCP）

1
2
3
4
5
6
7
8
9
10
11
12
13

# weather_server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("weather")

@mcp.tool()
def get_forecast(city: str) -> str:
    """返回指定城市的简要天气预报。"""
    # 实际项目里调用 OpenWeather 等 API
    return f"{city}: 晴，22°C，微风"

if __name__ == "__main__":
    mcp.run()  # 默认 stdio，供 Host 拉起子进程

在 Cursor / Claude Desktop 的配置中注册该命令（python weather_server.py），Host 启动时会 initialize 并列出 get_forecast。

4.2 TypeScript（官方 SDK）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({ name: "weather", version: "1.0.0" });

server.tool(
  "get_forecast",
  { city: z.string().describe("城市名，如 Shanghai") },
  async ({ city }) => ({
    content: [{ type: "text", text: `${city}: 晴，22°C` }],
  })
);

const transport = new StdioServerTransport();
await server.connect(transport);

工程建议：Tool 内只做 薄适配（参数校验 + 调用内部 REST/SDK），业务逻辑留在现有服务；Server 侧打结构化日志（tool_name、latency、error_code），便于与 Host 侧 trace 关联。

Resources 示例思路：为 Postgres Server 暴露 postgres://{db}/tables/{name}，返回 DDL 与采样行；模型写 SQL 前先 resources/read 对齐字段类型，再调用 run_readonly_query Tool，可显著降低幻觉列名。Prompts 则可注册 incident_triage(severity, service)，把 on-call 检查清单固化在 Server 而非各仓库的 .cursorrules 里。

5. 与 Claude / Cursor / LangChain 集成

Claude Desktop：在 claude_desktop_config.json 的 mcpServers 中声明 command 或 URL，重启后即可在对话里使用 Server 提供的 Tools/Resources。

Cursor：通过 MCP 设置添加 Server（stdio 或远程）。Agent 在规划任务时会 tools/list，再按需 tools/call；你可在规则里约束「先查 schema Resource 再写 SQL Tool」。

LangChain：使用 langchain-mcp-adapters 等包把 MCP Tool 转为 LangChain StructuredTool，接入 LCEL 或 LangGraph 节点。典型模式是图中一个 mcp_tools 节点负责绑定，与 LangChain / LangGraph 一文中的 bind_tools 编排衔接——MCP 负责 能力发现与进程隔离，LangGraph 负责 状态与重试。

集成时注意：不要把 MCP 当成数据库连接池。高 QPS 场景应在 Server 内做连接复用与超时；Host 侧对单次 tools/call 设置 deadline，避免模型反复重试打爆下游。

在 Cursor Agent 中，常见模式是「发现 → 调用 两步」：先 mcp_get_tools 拉 schema，再 mcp_call_tool 带精确参数，避免参数幻觉。Claude 侧则常把 MCP Tool 与内置联网搜索并存——在 system 或项目说明里写清 何时必须用内部 MCP、何时用公网，可减少模型误选工具。LangGraph 里可为 MCP 调用单独设 retry 与 fallback 边：Tool 超时则转人工节点，而不是让 LLM 无限重试同一 call。

6. 安全与治理

MCP 把能力拆到独立 Server，安全重点从「prompt 里别泄露密钥」升级为 供应链与权限：

1. 最小权限：Server 只暴露必要 Tool；读生产库用只读账号，写操作单独 Server 或二次确认。
2. 传输与身份：远程 Server 用 HTTPS + mTLS 或 OAuth；勿在仓库提交长期 Token；优先短期凭证与 Secret Manager。
3. 输入校验：所有 tools/call 参数按 JSON Schema 校验，防止 SQL 注入、路径遍历（../../etc/passwd）。
4. 人机在环：破坏性操作（删库、发版、转账）在 Host 层弹窗确认，不要完全交给模型自动 call。
5. 审计：记录 session_id、tool_name、参数摘要（脱敏）、调用方 Host 版本；便于 SOC2 与事故回溯。
6. 依赖供应链：只安装可信 MCP Server；stdio 模式等同 本地代码执行，务必审查源码与启动命令。

与 CrewAI / AutoGen 多 Agent 场景结合时：建议 一个 MCP Server 对应一个信任域（如「只读分析」与「写操作」分 Server），避免高权限 Tool 被探索性对话误触。

7. 总结

MCP 用 Host–Client–Server 分层和 Resources / Tools / Prompts 三类原语，把 Agent 工具集成从「每个 Host 写一遍」变成「每个系统写一次 Server」。落地路径清晰：先用 Python 或 TypeScript SDK 为内部 API 包一层薄 Server → 在 Cursor/Claude 验证 → 再接入 LangGraph 做编排与评测。下一篇将深入 Function Calling / Tool Use 闭环，讲清模型侧 tool_calls 与 MCP tools/call 如何配合。

系列导航 Series Navigation：

• 上一篇：CrewAI / AutoGen 多 Agent 协作
• 下一篇：Function Calling / Tool Use

English Title: Multi-Agent Frameworks — CrewAI Role-Playing vs AutoGen Conversation-Driven

当你已经会用单 Agent 完成「读文档 → 调工具 → 写答案」的闭环，下一步往往是把任务拆给多个专长不同的智能体。CrewAI 用角色与流程组织协作，AutoGen（现 AG2）用对话与消息传递驱动协作。二者都能做多 Agent，但心智模型、成本曲线和工程落点截然不同。本文帮你建立选型依据，并给出可运行的最小示例。

1. 何时需要多 Agent，何时单 Agent 足够

单 Agent 更合适的场景：

• 任务边界清晰，工具链固定（例如：查库 + 生成 SQL + 执行）
• 对话轮次可控，上下文在一两次工具调用内能收敛
• 团队希望最小依赖、最短上线路径

多 Agent 更值得投入的场景：

• 流程天然分阶段，且每阶段需要不同的系统提示与约束（调研 / 写作 / 审校）
• 需要对抗式或交叉验证（一个生成、一个挑错）
• 人类要在环中审批中间产物，再交给下一角色继续
• 单 Agent 的 prompt 已经臃肿，出现角色混淆、越权调用工具等问题

经验法则：若你只是把同一段 system prompt 复制三份并改名，多半还没赚到多 Agent 的收益；若各阶段的可观测输出、失败重试、人工卡点已经定义清楚，多 Agent 框架能显著降低编排代码的复杂度。

2. CrewAI：角色、目标与流程编排

CrewAI 的核心抽象是剧组（Crew）：每个 Agent 有明确的 role（职责）、goal（要达成的结果）、backstory（行为风格与专业背景）。Task 描述具体交付物，并绑定到执行者。Crew 把多个 Task 按 Process 串起来执行。

CrewAI 更贴近「岗位说明书 + 流水线」：你事先定义谁做什么、顺序如何，运行时较少出现「自由闲聊」。这对内容生产、竞品分析、报告生成等流程稳定的业务非常友好。

3. AutoGen / AG2：对话驱动的 GroupChat

AutoGen 将每个参与者建模为 ConversableAgent：既能调用 LLM，也能执行代码、调用函数。多 Agent 协作的典型模式是 GroupChat：所有消息进入共享频道，由 GroupChatManager（或新版中的 group chat 运行器）决定下一位发言者。

协作机制可以概括为：

1. Message passing — Agent A 的回复作为消息对象传给 B，可附带 tool_calls 与执行结果
2. Speaker selection — 轮询、auto（由 LLM 根据上下文选下一位）、或自定义函数
3. Nested chat — 子对话解决子问题，再把摘要回传主频道（控制上下文膨胀）

AG2（AutoGen 0.4+）在 API 上有所演进，但思想不变：用对话历史作为共享状态机，适合探索性任务、辩论式推理、需要多轮协商才能收敛的方案设计。代价是消息链更长，Token 与终止条件必须显式治理。

4. 对比一览

5. 代码示例

5.1 CrewAI：调研 → 撰稿 顺序流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="行业研究员",
    goal="收集 AI Agent 框架的 3 个对比维度与代表产品",
    backstory="你擅长结构化调研，只输出要点列表，不编造来源。",
    verbose=True,
)

writer = Agent(
    role="技术作者",
    goal="根据调研要点写一篇 800 字中文博客大纲",
    backstory="你面向开发者读者，语言简洁，小节清晰。",
    verbose=True,
)

research_task = Task(
    description="列出 CrewAI、AutoGen、OpenAI Agents SDK 的定位差异（各 3 条）",
    expected_output="Markdown 要点列表",
    agent=researcher,
)

write_task = Task(
    description="基于调研要点生成博客大纲（含 H2 标题）",
    expected_output="Markdown 大纲",
    agent=writer,
    context=[research_task],
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
)

result = crew.kickoff()
print(result)

5.2 AutoGen：双 Agent 群聊直至终止

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

import os
from autogen import ConversableAgent, GroupChat, GroupChatManager

llm_config = {"config_list": [{"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]}]}

planner = ConversableAgent(
    name="planner",
    system_message="你负责拆解任务，每次只提出下一步，不直接写长文。",
    llm_config=llm_config,
)

coder = ConversableAgent(
    name="coder",
    system_message="你根据 planner 的步骤写 Python 示例，代码需可运行。",
    llm_config=llm_config,
)

user = ConversableAgent(name="user", human_input_mode="NEVER")

group = GroupChat(agents=[user, planner, coder], messages=[], max_round=6)
manager = GroupChatManager(groupchat=group, llm_config=llm_config)

user.initiate_chat(
    manager,
    message="为「多 Agent 选型」写一段对比结论，并附一个最小 CrewAI 示例。",
)

生产环境请将 api_key 置于环境变量，并为 coder 配置沙箱执行；示例仅展示协作形态。

6. Token 成本与终止策略

多 Agent 的账单通常高于单 Agent，因为同一上下文会在多个角色间重复传递。

控费手段：

1. 限制轮次 — CrewAI 控制 Task 数量；AutoGen 设置 max_round / max_consecutive_auto_reply
2. 摘要中间态 — 长调研结果先压缩再交给 Writer，避免全文在多 Agent 间复制
3. 模型分级 — 调研/分类用 mini，终稿/审校用 flagship
4. 早停条件 — 检测 TERMINATE、任务完成 等关键词，或工具返回成功即结束
5. 可观测性 — 对每次 kickoff / 每轮 GroupChat 记录 prompt_tokens、completion_tokens

终止策略对照：

没有显式终止的 GroupChat，很容易在「互相客气」中烧掉数倍 Token——这是 AutoGen 新手最常踩的坑。

7. 如何选型

优先 CrewAI，若你：

• 已有清晰的 SOP（市场调研 → 大纲 → 正文 → 审校）
• 需要给非技术同事展示「岗位分工」图
• 希望默认顺序执行、减少对话跑偏

优先 AutoGen / AG2，若你：

• 问题本身需要多轮协商或辩论才能收敛
• 需要灵活的 UserProxy 人类审批
• 已有代码执行、函数调用密集的 Agent 生态，希望统一在消息层集成

仍可考虑单 Agent + 工作流引擎（如 LangGraph），当你要精细控制状态图、分支与持久化，而不想被「剧组」或「群聊」隐喻束缚时——系列前一篇 OpenAI Agents SDK 提供了另一种轻量编排路径。

8. 总结

CrewAI 用角色扮演 + 任务流水线降低「分工明确」类业务的编排成本；AutoGen 用共享对话释放「协商、迭代、人机共创」类场景的灵活性。二者不是替代关系，而是对不同协作形态的建模。落地时请先画清阶段交付物与终止条件，再选框架；否则多 Agent 只会把单 Agent 的混乱复制多份。

系列导航 Series Navigation：

• 上一篇：OpenAI Agents SDK
• 下一篇：MCP 协议与 Server 开发

系列第 07 篇：当 LangGraph 的图状态机显得过重时，OpenAI Agents SDK 用「Agent + Runner + Handoff + Guardrails」四条原语，把 2026 年多 Agent 编排压到可读的 Python 表面。

2025 年 OpenAI 将实验性的 Swarm 演进为 OpenAI Agents SDK（pip install openai-agents），定位为 轻量、生产就绪 的多 Agent 运行时：内置 Tracing、与 Responses API 深度集成，并支持 100+ 第三方 LLM。若你刚学完 LangChain / LangGraph 核心，本篇帮你建立第二套心智模型——何时用图，何时用 Handoff。

1. 定位：OpenAI Agents SDK vs LangGraph

1
2
3
4
5
6
7
8
9

用户输入 → Runner.run(triage_agent, query)
              ↓
         Input Guardrail（可选，首 Agent）
              ↓
         LLM + Tools / Handoff
              ↓
         Output Guardrail（可选，末 Agent）
              ↓
         final_output + Trace

选型建议（2026 主流实践）： 以 OpenAI 模型为主、团队希望 少写图、多写 Prompt 时，优先 Agents SDK；需要 强确定性状态机、HITL 中断、跨厂商图复用 时，LangGraph 仍是生产首选。二者可共存：LangGraph 节点内嵌 Runner.run 调用 OpenAI Agent 作为子任务。

从 Swarm 迁移的团队会明显感到 API 更「收口」：不再有零散 demo 级 helper，而是 Runner 统一调度 turn、tool、handoff。若你已在用 Assistants API，Agents SDK 可视为 Responses + 多 Agent 编排 的上层封装，减少自己拼 thread/run 状态的样板代码。

2. Agent 定义：instructions、tools、model

Agent 是可配置的 LLM 单元，最小集合为 name + instructions；生产环境通常再挂 tools、handoffs、guardrails 与 output_type（Pydantic 结构化输出）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

from agents import Agent, Runner, function_tool

@function_tool
def search_kb(query: str) -> str:
    """在内部知识库检索。"""
    return f"[mock] hits for: {query}"

support_agent = Agent(
    name="Support",
    instructions=(
        "你是客服 Agent。仅依据工具返回作答；"
        "无法确认时说明需要人工升级。"
    ),
    tools=[search_kb],
    model="gpt-4.1",  # 可省略，使用默认
)

与 LangChain 的差异： 工具用 @function_tool 装饰，docstring 即 schema 描述；无需单独 bind StructuredTool。output_type=MyModel 时，Runner 会驱动模型按 Pydantic 形状输出，适合工单分类、槽位抽取等 程序可读 场景。instructions 应写清 工具边界 与 拒绝策略，与系列 Prompt Engineering 中的 Constraints 段对齐。

执行入口统一为异步 Runner.run：

1
2
3
4
5
6
7

import asyncio

async def main():
    result = await Runner.run(support_agent, "如何重置 SSO？")
    print(result.final_output)

# asyncio.run(main())

多轮对话可传 result.to_input_list()、SDK Session，或 OpenAI 托管的 conversation_id——按「自控 vs 托管」选型，详见官方 Running agents 文档。

常见陷阱： instructions 过长却未拆 Handoff，导致单 Agent 上下文臃肿；工具 docstring 含糊，模型误选工具；output_type 与下游解析器字段不一致，引发静默截断。上线前用 10～20 条黄金用例跑 Runner.run，对照 Trace 检查 tool 选择与 handoff 目标是否符合预期。

3. Handoff：多 Agent 委托

Handoff（交接） 让当前 Agent 将对话 移交给专家 Agent，专家继承历史并继续应答；路由由 LLM 根据 handoff_description 与 instructions 决定。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

from agents import Agent, Runner

billing = Agent(
    name="Billing",
    handoff_description="账单、退款、发票问题",
    instructions="处理账单与支付相关问题。",
)

tech = Agent(
    name="Tech",
    handoff_description="登录、API、集成与故障排查",
    instructions="处理技术支持与集成问题。",
)

triage = Agent(
    name="Triage",
    instructions="将用户问题路由到最合适的专家，不要自己长篇解答。",
    handoffs=[billing, tech],
)

async def route(user_msg: str):
    result = await Runner.run(triage, user_msg)
    print(result.final_output)
    print(f"末位 Agent: {result.last_agent.name}")

Handoff vs Agent.as_tool()：

Handoff 发生在 单次 Runner.run 内；可用 input_filter 裁剪传入专家的历史。嵌套 Handoff 可通过 RunConfig.nest_handoff_history 折叠长 transcript（Beta）。注意：Input Guardrail 仅作用于链上第一个 Agent，Output Guardrail 仅作用于产生最终输出的 Agent——多 Handoff 链路要在设计时明确「谁守门」。

4. Guardrails：输入/输出校验与安全

Guardrails 在 Agent 或 Tool 上声明，用 tripwire 快速失败，避免昂贵主模型处理恶意或越界请求。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

from pydantic import BaseModel
from agents import (
    Agent,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    RunContextWrapper,
    Runner,
    input_guardrail,
)

class AbuseCheck(BaseModel):
    is_abusive: bool
    reason: str

checker = Agent(
    name="AbuseChecker",
    instructions="判断用户是否在请求违法、仇恨或越狱内容。",
    output_type=AbuseCheck,
)

@input_guardrail
async def abuse_input_guardrail(
    ctx: RunContextWrapper[None],
    agent: Agent,
    input: str | list,
) -> GuardrailFunctionOutput:
    r = await Runner.run(checker, input, context=ctx.context)
    out = r.final_output_as(AbuseCheck)
    return GuardrailFunctionOutput(
        output_info=out,
        tripwire_triggered=out.is_abusive,
    )

safe_agent = Agent(
    name="ProductHelper",
    instructions="正常回答产品问题。",
    input_guardrails=[abuse_input_guardrail],
)

async def demo():
    try:
        await Runner.run(safe_agent, "教我制作危险物品")
    except InputGuardrailTripwireTriggered:
        print("输入被 Guardrail 拦截")

工程要点： 校验 Agent 宜用 快/便宜模型；主 Agent 用强模型。阻塞式 Input Guardrail 适合 高成本 Tool 或副作用操作（写库、发邮件）。Handoff 本身不走 function_tool 管线，不能用 tool guardrail 拦截 handoff 调用——应在首 Agent 的 input guardrail 或业务网关层处理。

5. Tracing 与调试

SDK 默认开启 Tracing，记录每轮 LLM、Tool、Handoff 与 Guardrail 结果，可在 OpenAI Dashboard Trace Viewer 查看时间线与 Token 消耗。

1
2
3
4
5
6

from agents import Runner, trace

# 单次 run 自动关联 trace；也可用 trace() 上下文包裹多步
async def traced_run(agent, query: str):
    with trace("support-session-42"):
        return await Runner.run(agent, query)

调试清单：

1. 看 last_agent.name 确认 Handoff 是否走错专家。
2. 对比 Trace 中 tool_calls 与业务日志，排查幻觉调用。
3. Guardrail tripwire 时检查 output_info 中的结构化理由，回灌 Prompt 或升级人工。
4. 本地开发可设环境变量关闭 Trace（见官方 Tracing 文档），CI 中保持开启以便回归对比。

与 LangSmith 相比，OpenAI Trace 与 评测 / 微调 工具链更近；混合栈可将 Trace ID 写入自有 OTel span，实现跨系统关联。

在联调阶段建议固定 trace("env-dev-pr-123") 命名规范，便于在 Dashboard 按 PR 过滤。Guardrail tripwire 的异常栈应映射为 用户可读错误码（如 GUARDRAIL_INPUT_BLOCKED），避免把内部 checker 的 reasoning 原样暴露给终端用户。

6. 综合示例：分流 + 工具 + Guardrail

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

import asyncio
from pydantic import BaseModel
from agents import (
    Agent,
    GuardrailFunctionOutput,
    Runner,
    function_tool,
    input_guardrail,
    RunContextWrapper,
)

class Intent(BaseModel):
    off_topic: bool

intent_agent = Agent(
    name="Intent",
    instructions="判断是否与公司产品支持无关（闲聊、作业、政治）。",
    output_type=Intent,
)

@input_guardrail(run_in_parallel=False)
async def topic_guardrail(ctx, agent, input):
    r = await Runner.run(intent_agent, input, context=ctx.context)
    intent = r.final_output_as(Intent)
    return GuardrailFunctionOutput(
        output_info=intent,
        tripwire_triggered=intent.off_topic,
    )

@function_tool
def ticket_status(ticket_id: str) -> str:
    return f"Ticket {ticket_id}: in_progress"

resolver = Agent(
    name="Resolver",
    instructions="根据 ticket_status 回答进度，勿编造状态。",
    tools=[ticket_status],
    input_guardrails=[topic_guardrail],
)

triage = Agent(
    name="SupportTriage",
    instructions="支持类问题 handoff 给 Resolver。",
    handoffs=[resolver],
)

async def main():
    result = await Runner.run(triage, "工单 T-10086 现在什么状态？")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())

7. 生产考量

部署形态上，Agents SDK 适合 FastAPI / Celery Worker 内 asyncio 调用；高 QPS 场景在网关做鉴权与限流，Runner 层保持 无全局可变会话状态，Session 按 thread_id 隔离。与 Docker、Redis 队列的衔接见系列后续工程化篇章。

版本升级时关注 openai-agents-python Release Note：Handoff 嵌套、Sandbox Agent、MCP 托管工具等能力迭代较快，Pin 次要版本并在 staging 回放 Trace 回归，可降低生产行为漂移风险。

8. 小结与系列导航

OpenAI Agents SDK 用 Agent 定义能力边界、Handoff 实现专家路由、Guardrails 把安全与成本守门前移、Tracing 闭合调试闭环——在 2026 年与 LangGraph 并列为主流 Agent 框架之一。掌握「Handoff vs as_tool」「Guardrail 作用域」「Runner 会话模式」三条主线，即可在数天内搭起可观测的多 Agent 服务。

系列上一篇： LangChain / LangGraph 核心 —— 图状态机、Checkpoint 与确定性编排。

系列下一篇： CrewAI / AutoGen 多 Agent 协作 —— 角色化团队与对话式协作的另一条路径。

相关阅读：Agent 开发基础：Python 3.10+ 必备技能 · Prompt Engineering 系统性设计 · OpenAI Agents SDK 官方文档

English Title: LangChain & LangGraph Essentials for Agent Development — Interview Must-Knows

你已读过 LLM Agent 架构全景 与 LangGraph 生产实践，本文不再重复生态鸟瞰或部署细节，而是把 面试与上手 最常考的两块——LangChain 的 Runnable / LCEL / Agent 循环 与 LangGraph 的图运行时——压缩成可背诵、可写代码的知识清单。

前置知识建议：已完成 Embedding 与向量检索，理解 RAG 如何把检索结果注入 Prompt；模型调用见系列中的 API 实战文。下文默认使用 OpenAI 兼容的 ChatOpenAI，换 DeepSeek / Qwen 只需改构造参数。

0. 30 秒心智模型

1
2
3
4
5

用户输入 → LCEL 链（可选 RAG）→ Agent：LLM + bind_tools
                ↓ 多轮 tool call
         AgentExecutor（黑盒循环）  或  LangGraph（显式图 + Checkpoint）
                ↓
           最终 AIMessage / 结构化输出

面试官常顺着这条线追问：消息类型有哪些、谁执行工具、状态存在哪、如何防死循环。下面按模块拆开。

1. LCEL：链式组合的核心语法

LCEL（LangChain Expression Language） 把任意组件统一为 Runnable：invoke / batch / stream 接口一致，便于替换模型、加日志、做评测。

1
2
3
4
5
6
7
8
9
10
11
12

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是简洁的技术助手。"),
    ("human", "{question}"),
])
chain = prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()

print(chain.invoke({"question": "什么是 LCEL？"}))
# chain.stream(...) 同样可用

面试要点： LCEL 的价值是 组合性 + 可观测性（LangSmith 自动 trace 每个 Runnable），不是「又一种 DSL」。RunnableConfig 里的 callbacks、tags 用于链路追踪；configurable_fields 支持运行时换模型。

常考扩展：

• 并行与分支： RunnableParallel({"ctx": retriever, "q": RunnablePassthrough()}) 再 | prompt 是 RAG 标准写法；with_fallbacks([primary, backup]) 用于模型降级。
• 输入输出契约： 链的输入/输出类型在编译期可推断（get_input_schema），便于写单元测试与 JSON 校验。
• 与 Agent 的关系： Agent 内部仍是 Runnable；AgentExecutor 是对「agent Runnable + 工具执行循环」的包装，不是另一套 API。

手写 for 循环拼 prompt 也能跑，但失去统一 stream、批量评测与 Trace 切片，团队规模一大就难以维护——这是 LCEL 的工程理由，而非语法炫技。

2. Tool 定义与绑定（bind_tools）

Tool 是 Agent 与外部世界的契约：名称、描述、参数 Schema 直接影响模型是否 选对工具、填对参数。

1
2
3
4
5
6
7
8
9

from langchain_core.tools import tool

@tool
def search_docs(query: str, top_k: int = 3) -> str:
    """在内部知识库检索文档。query 为自然语言问题，top_k 为返回条数。"""
    return f"mock hits for: {query}"

tools = [search_docs]
llm_with_tools = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)

要点：

• 描述要写清 何时调用、输入含义、失败时返回什么，比函数名更重要。
• bind_tools 后模型输出 tool_calls；由 ToolNode 或自定义节点执行并写回 ToolMessage。
• 结构化工具可用 Pydantic BaseModel 或 @tool 自动生成 JSON Schema。
• 错误即 Observation： 工具抛错应捕获后返回可读字符串，让模型改参数重试，而不是让整个 Agent 崩溃。

1
2
3
4

from langchain_core.messages import ToolMessage

# ToolNode 执行后，消息序列为：
# HumanMessage → AIMessage(tool_calls=[...]) → ToolMessage(tool_call_id=...) → AIMessage(最终回答)

面试陷阱： 混淆 functions 旧 API 与 bind_tools / tool_calls 新 API；当前主流是 OpenAI 式 tool calling，Claude 走同一套 langchain-anthropic 适配层。

3. Agent Executor 与 ReAct 循环

经典 ReAct：Thought → Action（tool + args）→ Observation → 循环，直到模型不再发起 tool call 或达到 max_iterations。

1
2
3
4
5
6
7
8
9
10
11
12
13

from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你有 search_docs。无法回答时说明原因。"),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])
agent = create_tool_calling_agent(ChatOpenAI(model="gpt-4o-mini"), tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=5)

result = executor.invoke({"input": "LangGraph 和 AgentExecutor 区别？", "chat_history": []})

面试常问：

局限（必答）： 状态全在内存、难以精确插入人工节点、复杂分支要用 LangGraph。

消息类型速记表（必背）：

early_stopping_method="generate" 可在达到 max_iterations 时让模型强行总结，避免直接抛异常——生产可观测性要记录 停止原因（正常结束 / 超步 / 解析失败）。

4. LangGraph：State、Node、Edge、条件路由

LangGraph 把流程建模为 有向图，共享 State，节点返回 部分状态更新，框架负责 merge。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage

class State(TypedDict):
    messages: Annotated[list, add_messages]

def call_model(state: State):
    resp = ChatOpenAI(model="gpt-4o-mini").invoke(state["messages"])
    return {"messages": [resp]}

def should_continue(state: State) -> str:
    last = state["messages"][-1]
    if getattr(last, "tool_calls", None):
        return "tools"
    return END

graph = StateGraph(State)
graph.add_node("agent", call_model)
graph.add_node("tools", ToolNode(tools))  # langgraph.prebuilt
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")
app = graph.compile()

START / END 是哨兵节点；条件函数返回的字符串必须与 conditional_edges 第三参数字典的 键 一致，否则运行时报路由错误——面试手写代码时极易漏写映射表。

5. Checkpointing 与 Human-in-the-Loop（HITL）

Checkpointing 把每一步 State 持久化（内存 MemorySaver 或 Postgres PostgresSaver），支持：

• 进程崩溃后 从上次节点恢复
• 多轮对话 thread_id 隔离会话
• 时间旅行 调试（LangGraph Studio）

1
2
3
4
5
6
7

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "user-42"}}
app.invoke({"messages": [HumanMessage("查一下部署文档")]}, config)

HITL 常用 interrupt_before=["sensitive_tool"]：图在指定节点前暂停，人工审批后 app.invoke(None, config) 继续。面试要区分：HITL 是图级中断，不是 Prompt 里写「请人类确认」。

典型审批流：

1
2

agent 节点 →（interrupt_before tools）→ 等待人工 API 写入 state
         → 同一 thread_id 再次 invoke → tools 节点执行 → agent

checkpoint_id 与 thread_id 要纳入多租户设计：同一用户多设备、客服转接场景都依赖 thread 隔离。内存 MemorySaver 仅适合本地调试；生产用 Postgres / Redis 等后端，细节见 LangGraph 生产指南。

6. LangChain vs LangGraph：何时用哪个？

记忆口诀： LangChain 管 组件与链；LangGraph 管 有状态、可恢复的编排运行时。二者可共存：节点内仍用 LangChain 的 model、tool、retriever。

7. Runnable 综合示例（迷你 ReAct 图）

1
2
3
4
5
6
7

# 编译后的一次调用
out = app.invoke(
    {"messages": [HumanMessage("用 search_docs 查 LCEL")]},
    {"configurable": {"thread_id": "demo-1"}},
)
for m in out["messages"]:
    print(type(m).__name__, getattr(m, "content", "")[:80])

生产前检查清单：max 步数 / token 预算、tool 超时、checkpointer 后端、thread_id 与租户隔离、LangSmith LANGCHAIN_TRACING_V2=true。

与 Embedding 系列衔接： 检索链用 LCEL（retriever | format_docs | prompt | llm），Agent 链在检索结果之上再 bind_tools 做「查不到就调搜索 / 工单」类决策；向量库本身不是 LangGraph 的一部分，但常作为图中的独立 retrieve 节点，便于单独缓存与评测。

8. 面试 FAQ 速记

Q1：LCEL 和直接写 Python 函数拼 prompt 有什么区别？
统一 Runnable 接口，便于 stream、batch、组合与追踪；换模型只改链中一段。

Q2：bind_tools 之后谁执行工具？
模型只生成 tool_calls；执行器（AgentExecutor / ToolNode）负责调用并注入 ToolMessage。

Q3：LangGraph 的 State 为什么用 Annotated[list, add_messages]？
定义 reducer：新消息追加而非覆盖，避免多节点写同一字段时丢历史。

Q4：conditional_edges 和 AgentExecutor 内部路由有何不同？
前者 显式、可单测；后者黑盒在 executor 里，分支逻辑难定制。

Q5：Checkpoint 存的是什么？
每个 super-step 后的完整 State 快照 + 元数据，用于恢复与 HITL 续跑。

Q6：为什么生产 Agent 常从 AgentExecutor 迁到 LangGraph？
要 持久化、人工审批、精确循环控制、多 Agent 子图——这些在图里是一等公民。

Q7：和 CrewAI / AutoGen 比？
LangChain/LangGraph 偏 可编程编排与生态集成；CrewAI 偏角色剧本，AutoGen 偏对话式多 Agent。选型看团队是否要细粒度控制图与 Checkpoint。

Q8：stream 在图里怎么用？
app.stream(inputs, config) 按 节点完成 产出事件，适合 SSE 推前端；与 LLM token 级 stream 可嵌套在节点内部。

Q9：如何测试 Agent？
对 LCEL 链 mock LLM；对 LangGraph 测 条件路由函数 与单节点逻辑，再集成测 golden thread；避免只测最终字符串（易 flaky）。

常见踩坑：

9. 小结

掌握 LCEL 组合 → Tool 绑定 → ReAct 循环 → 图 State/Node/Edge → Checkpoint/HITL → 场景选型，足以应对大多数 Agent 框架面试题。实现时先用 LangChain 跑通工具链，再在 LangGraph 里把「是否继续调工具」「是否人工审批」画成显式边——这与 架构全景文 的 ReAct / 图状态机划分一致，而 生产指南 可继续深入 PostgresSaver、Studio 与监控。

下一篇将对比 OpenAI Agents SDK 的声明式 Agent 与 Handoff，帮助你在「LangChain 生态」与「官方 SDK」之间做技术选型。

系列导航

• 上一篇：Embedding 与向量检索
• 下一篇：OpenAI Agents SDK

English Title: Agent Memory with Embeddings & Vector Search — Chroma, Milvus & Qdrant

掌握大模型 API 调用之后，Agent 仍面临一个硬约束：上下文窗口有限，而业务记忆无限。对话历史、用户偏好、文档知识库、工具执行日志——若全部塞进 Prompt，成本与延迟会迅速失控。Embedding 将文本映射为稠密向量，再配合向量数据库做相似度检索，是构建 Agent 长期记忆 与 RAG 知识注入 的标准解法。它与 Prompt、Tool 调用并列，构成现代 Agent 栈的「数据面」。本文从原理到选型，再到可运行的 Python 流水线，帮你把「能对话」升级为「能记住、能查证」。

1. 为什么 Agent 需要向量记忆？

传统 Agent 只依赖滑动窗口内的 messages，会带来三类问题：

Agent 记忆可粗分为：短期（当前 thread 的 messages）、长期（跨会话的用户画像与摘要）、外部知识（PDF、Wiki、工单）。Embedding + 向量检索主要服务后两者；短期记忆仍建议配合 Redis 或数据库存原文，向量层负责「按意思找片段」。例如用户说「还是按上次那样配环境」，系统无需扫描全部历史，只需用当前意图检索「上次环境配置」相关块即可。这种语义索引比关键词匹配更抗表述变化，是 Agent 体验从「健忘」到「贴心」的关键跃迁。

2. 文本 Embedding 模型选型

Embedding 模型的任务是把语义相近的句子映射到向量空间中彼此靠近的位置。主流选择：

选型原则： 同一索引内必须使用同一模型；换模型需全量重嵌入。维度越高不一定越好——在召回率与存储/延迟之间权衡。中文 Agent 若走 API，可优先 text-embedding-3-small；若自建，BGE-M3 是常见默认。本地推理可用 sentence-transformers 加载 BGE，避免每次检索都走外网；注意 GPU 批处理能显著降低入库阶段的耗时。无论哪种模型，都应在离线集上做一次 MTEB 或自建问答对 的抽检，确认你的领域语料（工单、代码注释、产品手册）召回达标后再上线。

3. 相似度检索原理

向量检索的核心是比较查询向量 q 与库中向量 d 的相似度：

• 余弦相似度（Cosine）：衡量方向一致性，对向量长度不敏感，文本场景最常用
• 点积（Dot Product）：若向量已 L2 归一化，等价于余弦；未归一化时大范数向量会占优
• 欧氏距离（L2）：几何距离，部分库默认支持

百万级以上规模时，全量暴力扫描不可行，需 近似最近邻（ANN） 索引。HNSW（分层可导航小世界图）是工业界主流：构建时建多层图，查询时从顶层贪心下降，在 召回率 vs 延迟 间通过 ef_search、M 等参数调节。理解这一点有助于调参：召回偏低时先增大 ef，而非盲目加 chunk。另有 IVF、PQ 等索引适合超大规模与内存受限场景，但 Agent 记忆库往往在百万条以内，HNSW 通常足够。检索返回的是「相似」而非「相同」——务必在 Prompt 中要求模型仅依据检索片段回答，并在无相关结果时明确说「知识库中未找到」，降低胡编风险。

4. 向量数据库对比：Chroma vs Milvus vs Qdrant

务实建议： 学习与 PoC 用 Chroma；需要复杂 where 过滤（user_id、session_id）且要上生产，看 Qdrant；数据量与 SLA 要求极高、已有运维体系，选 Milvus。三者 Python SDK 心智模型相近：collection → upsert → query。

5. Agent 场景的 RAG 流水线

典型 RAG（Retrieval-Augmented Generation）在 Agent 中的位置：

1
2

文档/对话 → 分块(Chunk) → Embedding → 写入向量库
用户提问 → Query Embedding → Top-K 检索 → 拼入 Prompt → LLM 生成

与纯问答 RAG 不同，Agent 还需：写入时机（工具结果、用户确认的事实何时入库）、检索时机（Planner 决策前 vs 回答前）、引用格式（要求模型标注 [1][2] 便于审计）。记忆写入建议附带 metadata：user_id、source、timestamp、importance，便于过滤与过期清理。进阶做法是把检索封装为独立 Tool（如 search_memory(query)），由 LLM 决定何时查记忆，而不是每轮固定注入 Top-K——这在多跳任务中更省 token，也更接近人类「想起来再查」的行为。下一篇 LangChain / LangGraph 将把此类节点编排进状态图。

6. Python 示例：嵌入、存储、检索

以下用 Chroma 演示最小闭环（需 pip install chromadb openai）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

import chromadb
from openai import OpenAI

client = OpenAI()
chroma = chromadb.PersistentClient(path="./agent_memory")
collection = chroma.get_or_create_collection("memories")

def embed(texts: list[str]) -> list[list[float]]:
    resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
    )
    return [d.embedding for d in resp.data]

# 写入记忆
docs = [
    "用户偏好：接口文档用 OpenAPI 3.1",
    "上次部署失败原因：Redis 连接超时",
]
ids = ["mem-1", "mem-2"]
collection.add(
    ids=ids,
    documents=docs,
    embeddings=embed(docs),
    metadatas=[{"user_id": "u42"}, {"user_id": "u42"}],
)

# 检索
query = "部署出过什么问题？"
q_emb = embed([query])[0]
hits = collection.query(
    query_embeddings=[q_emb],
    n_results=2,
    where={"user_id": "u42"},
)
for doc, dist in zip(hits["documents"][0], hits["distances"][0]):
    print(doc, dist)

将 hits["documents"] 拼入 system 或 user message 即可驱动 Agent 回答。生产环境把 PersistentClient 换成 Qdrant/Milvus 对应客户端，接口模式不变。

7. 常见陷阱

另外：不要把密钥写进向量库；敏感内容入库前脱敏。评测时用固定「黄金问题集」测 Recall@K，而非凭感觉调 chunk。

8. 小结

Embedding 与向量检索是 Agent 记忆层 的基建：它不负责推理，却决定 Agent 能否在有限上下文中「想起」正确信息。建议路径：Chroma 本地跑通 RAG → 加上 metadata 过滤 → 按规模迁移 Qdrant/Milvus → 与 LangGraph 的 checkpointer 分工（状态机管流程，向量库管语义记忆）。监控指标建议关注：检索延迟 P99、Recall@5、注入 token 占比 与 「未找到仍作答」率，四者联动才能判断记忆系统是否真的在帮 Agent，而不是增加噪声。掌握本文后，即可进入框架层，把检索节点编排进多步 Agent。

系列导航 Series Navigation：

• 上一篇：主流模型 API 调用实战
• 下一篇：LangChain / LangGraph 核心

English Title: Mainstream LLM API Guide — OpenAI, Claude, DeepSeek & Qwen

掌握 Prompt Engineering 之后，下一步是把设计好的提示词真正「跑起来」。无论是构建对话机器人、文档问答，还是多步 Agent，底层都离不开对大模型 HTTP API 的熟练调用。本文聚焦 OpenAI、Claude、DeepSeek、通义千问四大主流服务的统一心智模型、计费与上下文管理、流式输出实现，以及各厂商 Python 调用示例，为后续 Embedding 检索与 Function Calling 专题打下基础。

After mastering Prompt Engineering, the next step is running your prompts in production. Whether you’re building chatbots, document Q&A, or multi-step agents, everything depends on fluent LLM HTTP API usage. This article covers a unified mental model, billing, context management, streaming, and Python examples for four major providers.

1. 统一心智模型 | Unified Mental Model

无论哪家厂商，一次 Chat Completion 调用的本质结构相同。把差异抽象掉之后，你只需要记住下面这张「通用蓝图」：

一次典型调用的生命周期是：组装 messages → 发送 HTTP 请求 → 解析 choices → 提取 content 或 tool_calls → 记录 usage。Agent 开发中，这个循环会被执行数十次，因此封装统一的 Provider 层是工程化的第一步。

关键洞察： DeepSeek 与通义千问均提供 OpenAI 兼容接口（Compatible Mode），只需替换 base_url 和 api_key，即可复用 openai 官方 SDK。Claude 使用独立的 Messages API，字段名略有不同（如 max_tokens 为必填），但语义完全对应。这意味着你的业务代码可以做到「一套抽象，多家后端」。

角色（role）的约定也趋于统一：system 设定行为边界，user 承载用户输入，assistant 是模型历史回复，tool 则用于回传工具执行结果——这是 Function Calling 闭环的基础。

2. Token 计费与成本优化 | Token Billing

所有主流 API 均按 Token 计费，而非按请求次数。计费公式为：

总费用 = 输入 tokens × 输入单价 + 输出 tokens × 输出单价

输入包含完整的 messages 历史（含 system prompt），输出则是模型生成的文本。同一对话轮次越多，输入 token 会线性增长——这是长对话 Agent 成本失控的主要原因。

五条实用优化策略：

1. 精简 System Prompt — 去掉冗余指令和重复示例，每多 500 token 系统提示，在千次调用后都是可观支出
2. 控制输出长度 — 设置合理的 max_tokens，并在 prompt 中明确要求简洁回答，避免模型「话痨」
3. 模型路由（Model Routing） — 分类、摘要等简单任务用轻量模型（gpt-4o-mini、deepseek-chat），复杂推理再上旗舰
4. Prompt Caching — OpenAI 与 Claude 均支持对重复前缀缓存，系统提示不变时可显著降低输入成本
5. 批量 API（Batch） — 非实时场景（如离线评估、数据标注）使用 Batch 接口，通常享 50% 折扣

响应体中的 usage 字段（prompt_tokens、completion_tokens）是成本监控的第一数据源。生产环境务必对每次调用打点，按 model、user、feature 维度聚合，才能做有效的 FinOps。

3. 上下文窗口与截断策略 | Context Window

每个模型都有上下文上限（Context Window），超出后 API 会直接报错。Agent 场景中，多轮对话 + 工具返回 + RAG 文档很容易触顶。

常见陷阱： 不同厂商对 token 的计算方式略有差异——中文通常 1–2 个汉字对应 1 token，英文约 4 字符 1 token。不要凭字符数估算，应使用各 SDK 提供的 token 计数工具（如 tiktoken）在发送前预检。另外，输入越长，首字延迟（TTFT）往往越高，需要在「信息完整」与「响应速度」之间权衡。

4. 流式响应（SSE）| Streaming Implementation

流式输出通过 Server-Sent Events（SSE） 逐块推送 delta，让用户在模型尚未生成完毕时就能看到文字逐字出现，显著降低感知延迟。对聊天类 Agent 而言，流式几乎是标配。

1
2
3
4
5
6
7
8
9
10
11
12
13

from openai import OpenAI

client = OpenAI()
stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "用三句话介绍 Agent"}],
    stream=True,
)

for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)

后端实现要点： 设置响应头 Content-Type: text/event-stream、Cache-Control: no-cache；若经过 Nginx 反向代理，需加 X-Accel-Buffering: no 禁用缓冲。Claude 流式使用 client.messages.stream()，事件类型为 content_block_delta，逻辑相同。

前端消费： 可用 fetch 配合 ReadableStream 逐行解析 data: {...} 行；注意处理连接中断与 [DONE] 结束标记，并在 UI 层做打字机效果与取消按钮。

5. Function Calling 预览 | Tool Use Preview

工具调用（Tool Use / Function Calling）是 Agent 与外部世界交互的核心机制。各厂商的实现已高度趋同：

• OpenAI / DeepSeek / Qwen — 请求中传 tools 数组，响应 choices[0].message.tool_calls
• Claude — 请求中传 tools，响应 content 块类型为 tool_use

模型不会直接执行你的函数。它只返回结构化 JSON：「调用哪个工具、传什么参数」。你的代码负责真正执行（查数据库、调 API），再把结果以 role: tool 的消息塞回 messages，发起下一轮请求——形成 LLM → Tool → LLM 的闭环。系列第 10 篇《Function Calling / Tool Use》将用完整示例拆解这一流程。

6. 厂商对比 | Provider Comparison

选型建议：国际化产品优先 OpenAI/Claude；成本敏感或国内部署选 DeepSeek/Qwen；开发阶段可用兼容接口快速切换，避免供应商锁定。

7. 各厂商 Python 调用示例 | Code Examples

7.1 OpenAI

1
2
3
4
5
6
7
8
9
10
11
12

from openai import OpenAI

client = OpenAI()  # 环境变量 OPENAI_API_KEY
resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "你是简洁的技术助手。"},
        {"role": "user", "content": "什么是 Token？"},
    ],
)
print(resp.choices[0].message.content)
print(resp.usage)  # 记录 token 消耗

7.2 Claude (Anthropic)

1
2
3
4
5
6
7
8
9
10
11

import anthropic

client = anthropic.Anthropic()  # ANTHROPIC_API_KEY
msg = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system="你是简洁的技术助手。",
    messages=[{"role": "user", "content": "什么是 Token？"}],
)
print(msg.content[0].text)
print(msg.usage)

7.3 DeepSeek（OpenAI 兼容）

1
2
3
4
5
6
7
8
9
10
11

from openai import OpenAI

client = OpenAI(
    api_key="sk-xxx",
    base_url="https://api.deepseek.com",
)
resp = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "什么是 Token？"}],
)
print(resp.choices[0].message.content)

7.4 通义千问（DashScope 兼容模式）

1
2
3
4
5
6
7
8
9
10
11

from openai import OpenAI

client = OpenAI(
    api_key="sk-xxx",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
resp = client.chat.completions.create(
    model="qwen-plus",
    messages=[{"role": "user", "content": "什么是 Token？"}],
)
print(resp.choices[0].message.content)

8. 实战要点 | Production Tips

1. API Key 走环境变量或密钥管理服务 — 绝不硬编码到 Git 仓库
2. 重试与指数退避 — 对 429（限流）和 5xx 使用 tenacity 等库自动重试
3. 合理超时 — 推理模型（如 deepseek-reasoner）耗时长，设置 60–120s timeout
4. 抽象 Provider 层 — 统一 chat(messages) -> str 接口，方便 A/B 测试与 fallback
5. 可观测性先行 — 记录 latency、token、model、error_code，接入 LangSmith 或自建日志

9. 总结 | Conclusion

四大 API 的调用范式已高度趋同：Messages 数组进，文本或 tool_calls 出。差异主要在定价、上下文长度、区域延迟与生态集成。Agent 开发者的务实策略是：用 OpenAI 兼容层统一 DeepSeek 与 Qwen，Claude 单独封装 Messages API，上层实现模型路由与成本监控。掌握本文内容后，你已具备构建「能对话、能流式、能记账」的 LLM 应用基础能力。

系列导航 Series Navigation：

• 上一篇：Prompt Engineering 系统性设计
• 下一篇：Embedding 与向量检索

English Title: Systematic Prompt Engineering for Agents — Beyond “Writing Prompts”

很多团队把 Prompt 当成「调文案」：多试几次、感觉对了就上线。这在单次聊天里或许够用，Agent 场景下这远远不够——你的 Prompt 同时服务 人类可读性 与 程序可解析性，还要在工具调用、多轮对话、RAG 注入下保持稳定。本文把 Prompt Engineering 当作 系统工程：从 System 设计、样例策略、推理链、结构化输出到版本治理，建立可复用的方法论。

1. System Prompt 设计：角色、约束与输出格式

把 System Prompt 当作 Agent 的运行时配置（Runtime Config），而不是开场白。推荐固定三段，顺序不要随意调换：

1
2
3
4

SYSTEM = """你是生产环境运维 Agent。
角色：根据告警与日志定位根因，并给出可执行的修复建议。
约束：仅使用已注册工具；禁止编造日志行；无法确认时返回 NEED_CLARIFICATION。
输出：先写 ## Analysis（Markdown），再写 ## Action（单行 JSON：{"tool": str, "args": dict}）。"""

工程经验： 约束段优先写 安全与合规（密钥、PII、越权工具），再写 质量（引用来源、标注不确定性）。输出格式要与下游代码契约一致——若解析器只认 JSON，就不要在 System 里允许「偶尔用自然语言总结」。多 Agent 系统中，每个子 Agent 的 System 应 窄而深，由 Orchestrator 负责全局目标，避免多个「全能 System」互相打架。上线前用 对抗用例 测一遍：空输入、超长输入、多语言混杂、伪造工具返回，确认 Agent 仍遵守格式与约束。

2. Few-shot：何时用、如何用

Few-shot 不是「多给几个例子就更聪明」，而是在 缩小输出分布——让模型对齐你期望的格式、语气与决策边界。

高质量 Few-shot 的特征：输入真实、输出可直接进业务库、覆盖失败模式（空值、歧义、多意图）。样例应放在 User/Assistant 轮次 中呈现，而非塞进 System——否则占用宝贵的「宪法」窗口，且难以单独迭代。动态 Few-shot（用 Embedding 检索历史优质对话）适合客服、运维等长尾场景，但要监控「检索到错误范例」导致的系统性偏差，并设置相似度阈值与人工抽检。定期 淘汰过时样例（产品改名、API 字段变更），否则模型会顽固复用过期格式。

3. Chain-of-Thought（CoT）与推理型 Agent

ReAct、Plan-and-Execute 等架构里，模型需要在 不确定环境 中多步决策。CoT 的核心是：把隐式推理外显化，便于调试、重试与人工审核。

1
2
3
4

请按以下步骤回答：
1. 列出已知条件与仍缺失的信息
2. 逐步推导（每步一行，标注依据：规则 / 工具结果 / 假设）
3. 给出最终结论（单独一行，前缀 FINAL:）

对数学、合规审查、故障根因分析尤其有效。生产上常见两种策略：（1）全量 CoT 写入日志，用户只见 FINAL；（2）模型原生思考通道（如 Extended Thinking）与主回答分离，减少 Token 浪费。注意：CoT 越长，越容易被 幻觉中间步骤 误导——关键结论仍应通过工具结果或规则引擎校验。

4. 结构化输出：JSON Mode 与 Schema 约束

Agent 的下游是代码。自然语言「看起来对」不等于 可执行。应在 Prompt 层与 API 层 双重约束。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# OpenAI — JSON Schema 严格模式（示意）
response = client.responses.create(
    model="gpt-4.1",
    input=[{"role": "user", "content": user_query}],
    text={
        "format": {
            "type": "json_schema",
            "name": "ticket_classify",
            "schema": {
                "type": "object",
                "properties": {
                    "category": {"type": "string", "enum": ["bug", "feature", "question"]},
                    "confidence": {"type": "number", "minimum": 0, "maximum": 1},
                },
                "required": ["category", "confidence"],
                "additionalProperties": False,
            },
            "strict": True,
        }
    },
)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

// Anthropic Claude — 用 tool_use 强制结构化输出（Node.js 示意）
import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();
const msg = await client.messages.create({
  model: "claude-sonnet-4-20250514",
  max_tokens: 1024,
  system: SYSTEM,
  tools: [{
    name: "submit_result",
    description: "提交结构化分析结果",
    input_schema: {
      type: "object",
      properties: {
        summary: { type: "string" },
        severity: { type: "integer", minimum: 1, maximum: 5 },
      },
      required: ["summary", "severity"],
    },
  }],
  tool_choice: { type: "tool", name: "submit_result" },
  messages: [{ role: "user", content: userQuery }],
});

失败处理： 解析失败时走固定重试 Prompt（「仅返回符合 Schema 的 JSON，不要解释」）；仍失败则降级为人工队列，切勿 JSON.parse 吞掉异常后静默继续。

5. Prompt 模板与版本管理

Prompt 不应散落在 if/else 字符串里。推荐 模板文件 + 变量注入 + 语义化版本号：

1
2
3
4
5
6
7
8

# prompts/ops_agent_v2.yaml
id: ops_agent
version: "2.1.0"
system: |
  {{ role_block }}
  {{ constraints_block }}
  当前环境：{{ env_name }}；允许工具：{{ tool_list }}
changelog: "2.1.0 收紧工具白名单；2.0.0 引入 CoT 输出段"

上线流程建议：评测集门禁（同一批黄金任务，对比通过率 / 平均 Token / 违规率）→ 灰度（5% 流量）→ 全量。日志中记录 prompt_id@version，与 LangSmith、OpenTelemetry 关联，出问题时才能回答「是模型变了还是 Prompt 变了」。团队内可维护 Prompt Registry：谁负责、适用场景、依赖的工具列表、最后一次评测日期——把 Prompt 当作与微服务同级的配置资产，而不是个人笔记本里的草稿。

6. 反模式与安全

牢记：System Prompt 是软约束。真正安全靠鉴权、沙箱、输出过滤与人工审批节点（Human-in-the-Loop），而不是在 Prompt 里写「请不要作恶」。对外暴露的 Agent 还应做 输出后处理：PII 脱敏、链接白名单、代码块静态扫描，形成「模型 + 规则」双保险。

7. 小结：在 Agent 栈中的位置

Prompt Engineering 连接 语言能力 与 工程契约：它决定 Tool 参数是否稳定、Planner 是否可解析、评估指标是否可复现。建议建立个人或团队的 Prompt 检查清单（角色是否单一、约束是否可测试、输出是否可解析、是否有版本号、是否过评测集），在每次迭代时勾选，避免凭直觉改一句就合并主分支。掌握本文六块能力后，进入模型 API、Embedding 与 RAG，才能把「会说话的模型」变成「可交付的 Agent 服务」。

系列导航

• 上一篇：TypeScript/Node.js 全栈 Agent 开发
• 下一篇：主流模型 API 调用实战

在 Agent 学习路线的第一层，Python 开发基础 负责数据清洗、脚本化实验与模型侧胶水；而 TypeScript + Node.js 则天然承接「Web 前端 + API + 流式对话」的全栈链路。若你的产品形态是对话界面、SaaS 控制台或需要快速迭代的 B 端工具，TS 往往是投入产出比更高的路径。本文聚焦如何用类型安全的 JS 生态构建可上线的 Agent 应用。

1. 为什么 Agent 开发离不开 TypeScript？

Agent 的核心难点不是调一次 Chat API，而是 工具 Schema、多轮状态、流式 UI 在前后端之间反复传递。模型输出的 Tool Call 本质是 JSON，字段多一个、少一个都会导致执行失败；会话里还要叠加 tool_calls、tool_results 与人工确认节点。TypeScript 的价值在于：

当工具从 3 个增长到 30 个时，没有类型的项目会在「模型幻觉 + 运行时解析失败」上付出成倍调试成本。此外，Discriminated Union 可精确建模「用户消息 / 助手消息 / 工具结果」等联合类型，配合 satisfies 能在重构时让编译器替你检查遗漏分支——这在多 Agent、多步骤编排里尤为省事。

2. 主流框架速览

LangChain.js 提供 createReactAgent、RunnableSequence 等与 Python 版概念对齐的 API，适合已有 LangGraph 经验、需要跨语言迁移的团队。Vercel AI SDK 把 streamText、generateObject 与 React Hook 打通，多模型通过 @ai-sdk/* 适配器切换，是 Next.js 场景的事实标准。Mastra 强调工作流、评估与 Tracing 在同一 TS 仓库内完成，适合从零搭建可观测的 Agent 平台。OpenClaw 则以本地 Gateway 守护进程为控制面，通过 WebSocket 连接 IM 通道与 Plugin，适合「个人助手常驻本机」而非纯 Web SaaS 的形态。

选型建议：产品 Web 对话优先 AI SDK；研究型编排优先 LangChain.js；需要 7×24 本机助手与多渠道 可评估 OpenClaw 的 Gateway 架构。三者并非互斥——例如在 Next.js 中用 AI SDK 做 UI 流，后台用 LangChain.js 跑 RAG 管道，是常见组合。

3. TypeScript 模式：Schema 即契约

工具定义应「单一数据源」：用 Zod 描述参数，再推导 TS 类型，避免手写两份 Schema。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

import { z } from "zod";
import { tool } from "ai";

const SearchArgs = z.object({
  query: z.string().min(1),
  limit: z.number().int().max(10).default(5),
});

type SearchArgs = z.infer<typeof SearchArgs>;

export const searchTool = tool({
  description: "搜索内部知识库",
  parameters: SearchArgs,
  execute: async ({ query, limit }) => {
    const hits = await kb.search(query, limit);
    return { items: hits };
  },
});

除 Zod 外，也可用 interface + 运行时校验 的折中：对外导出 interface SearchArgs，内部用 SearchArgsSchema.parse(raw) 兜底。LangChain.js 侧可用 StructuredTool + zodToJsonSchema 生成 OpenAI 兼容的 function schema；OpenClaw Plugin SDK 则常用 TypeBox 描述 parameters，与 Gateway 的 JSON Schema 校验对齐。原则不变：Schema 只维护一份，JSON Schema、TS 类型与文档都从它派生。

LangChain.js 绑定工具的最小示例如下，注意 schema 与 func 签名由 Zod 推断保持一致：

1
2
3
4
5
6
7
8
9
10
11
12
13

import { z } from "zod";
import { tool } from "@langchain/core/tools";

const getWeather = tool(
  async ({ city }) => {
    return await weatherApi.fetch(city);
  },
  {
    name: "get_weather",
    description: "查询城市天气",
    schema: z.object({ city: z.string() }),
  }
);

4. Node.js 异步：流式与 SSE

Agent 响应必须 边生成边推送，否则首字延迟会拖垮体验。用户感知到的「聪明」往往取决于首 token 是否在数百毫秒内出现，而不是最终答案有多长。Node 18+ 原生支持 ReadableStream，Fetch API 也可消费上游模型的 SSE；各框架在此基础上封装了 StreamingTextResponse 或 Data Stream 协议，把文本 delta、tool call 片段与完成事件编码成前端可解析的帧。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

// Next.js App Router 示例（Vercel AI SDK）
import { streamText } from "ai";
import { openai } from "@ai-sdk/openai";

export async function POST(req: Request) {
  const { messages } = await req.json();
  const result = streamText({
    model: openai("gpt-4o"),
    messages,
    tools: { search: searchTool },
    maxSteps: 5,
  });
  return result.toDataStreamResponse();
}

若不用框架封装，原生 Node 也可用 res.writeHead(200, { "Content-Type": "text/event-stream" }) 手写 SSE，按 data: ${JSON.stringify(chunk)}\n\n 推送 token 与 tool 事件。无论哪种方式，底层注意点一致：不要在 for await 里执行 CPU 密集计算阻塞事件循环；耗时工具应 await 完成后再写入流片段；生产环境设置 Cache-Control: no-cache、禁用缓冲（如 Nginx proxy_buffering off），必要时加心跳包，避免代理超时断连。客户端断开时，应监听 req.aborted 并取消上游 LLM 请求，节省 Token。

5. 全栈 Agent 架构

1
2
3
4
5
6
7
8
9

┌─────────────┐     SSE/DataStream     ┌──────────────────┐
│ React 客户端 │ ◄──────────────────► │ API Route / Hono │
│ useChat     │                      │ streamText + tools│
└─────────────┘                      └────────┬─────────┘
                                              │
                                     ┌────────▼─────────┐
                                     │ LLM Provider     │
                                     │ Vector DB / MCP  │
                                     └──────────────────┘

• 前端：useChat 管理消息列表、loading 与 tool call 卡片；可用 experimental_toolInvocations 展示「正在调用搜索…」等中间态。
• API 层：JWT 或 Session 鉴权、按用户限流、敏感工具（删库、发邮件）走二次确认或 RBAC 白名单。
• 数据层：threadId 映射 Redis 存最近 N 轮；长期记忆与 RAG 文档块写入向量库（系列第 05 篇《Embedding 与向量检索》展开）。

部署上，Next.js 可一键上 Vercel Edge；也可用 Hono + Node 或 Bun 获得更低冷启动。关键是把 模型密钥与 Tool 密钥 关在服务端，前端只拿会话 Token。若 Agent 需要调用企业内部 REST API，建议在 API 层做 Tool Gateway：统一 OAuth 刷新、审计日志与超时重试，避免把业务凭证直接交给 LLM 上下文。MCP（Model Context Protocol）正在成为连接外部工具的标准接口，系列第 09 篇将专门展开；在 TS 栈中可先以 HTTP MCP Server 暴露数据库或工单系统，再由 Agent 通过协议发现工具列表。

6. Python vs TypeScript：如何取舍？

实践上常见 混合架构：Python 跑离线 RAG 索引、微调与批评估，TS 服务暴露 HTTP/SSE 给产品——用 OpenAPI 或 tRPC 保持契约一致。团队若以前端为主、无重型 ML 管线，可全程 TS；若以 Notebook 探索为主，再逐步把稳定链路迁到 API 层。

7. 实战要点与常见陷阱

1. 工具粒度：一个 Tool 只做一件事，描述里写清输入示例与「何时不要调用」。
2. maxSteps 上限：streamText 的 maxSteps 防止 ReAct 死循环烧 Token。
3. 错误可观测：记录每次 tool 的 input/output 与 latency，便于回放（LangSmith / OpenTelemetry）。
4. 环境变量：OPENAI_API_KEY 等仅存服务端，切勿打进客户端 bundle。
5. 幻觉参数：对枚举类字段用 z.enum() 限制，减少模型编造非法状态。
6. 流式中断：用户点击「停止」时，前后端都要 abort 上游 fetch，避免幽灵计费与孤儿工具调用。

延伸阅读

• 上一篇：Python 3.10+ Agent 开发基础
• 下一篇：Prompt Engineering 系统性设计

掌握 TS 全栈栈后，建议继续学习系列中的 LangChain / LangGraph 核心 与 Function Calling，把类型安全的工具层接到更复杂的编排图上。下一篇将系统讲解 Prompt Engineering——无论 Python 还是 TypeScript，提示词设计都是 Agent 可靠性的底座。