生产化工程

哓番茄大约 4 分钟

生产化工程

demo 到生产之间隔着：流式 API 服务、并发与限流、可观测、评估回归、成本护栏。本篇把这些「最后三公里」一次讲完。

一、把 agent 包成流式 API 服务

agent 的标准交付形态：SSE 流式接口。FastAPI 实现：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import json

app = FastAPI()

class ChatRequest(BaseModel):
    session_id: str
    message: str

@app.post("/chat")
async def chat(req: ChatRequest):
    async def event_stream():
        # 恢复会话（05 篇的 SessionStore）
        messages = SessionStore(req.session_id).load()
        messages.append({"role": "user", "content": req.message})

        # 事件 1：文本增量
        async for delta in agent_stream(messages):
            yield f"data: {json.dumps({'type': 'text', 'delta': delta})}\n\n"

        # 事件 2：工具调用（前端可渲染「正在查询订单…」）
        # 事件 3：结束（带 usage，前端可显示 token 消耗）
        yield f"data: {json.dumps({'type': 'done', 'usage': last_usage})}\n\n"

        SessionStore(req.session_id).save(messages)

    return StreamingResponse(event_stream(), media_type="text/event-stream")

前端消费（EventSource 或 fetch 流）：

const resp = await fetch("/chat", { method: "POST", body: JSON.stringify(req) });
const reader = resp.body!.getReader();
// 逐 chunk 解析 SSE data: 行，按 type 分发渲染

SSE vs WebSocket：
  单向推送（LLM 流式天然单向）→ SSE 足够，且自带 HTTP 语义
  双向（语音、协作）→ WebSocket
  → 网络细节见[实时通信篇](../../computer/net/04-realtime-communication.md)

二、并发与限流

agent 是长连接 + 重计算（一个会话跑几十秒、烧几十万 token），网关层必须设防：

# 1. 全局并发上限（保护上游 LLM 配额与自身内存）
sem = asyncio.Semaphore(20)

@app.post("/chat")
async def chat(req: ChatRequest):
    if sem.locked():
        return JSONResponse({"error": "busy"}, status_code=503)
    async with sem:
        ...

# 2. 用户级限流（令牌桶：每人每分钟 N 次）
# 3. 会话级排队（05 篇：同一 session 串行，防上下文竞争）
# 4. 任务级预算（06 篇：轮数/token/wall-clock 三闸）

降级链路设计：

上游 LLM 超时/限流 → 重试（指数退避）→ 备用模型 → 队列化（稍后处理）
全部失败 → 模板兜底（「转人工」永远可用）

三、可观测：tracing 是刚需

多轮工具调用的 agent，出问题时没有 trace 就是一头雾水。最小可行观测：把每次 LLM 调用与工具执行记录成 span 树。

import time, uuid, contextvars

current_trace = contextvars.ContextVar("trace", default=None)

def trace_event(kind: str, data: dict):
    t = current_trace.get()
    if t:
        t["spans"].append({
            "ts": time.time(), "kind": kind,   # llm_call / tool_use / tool_result
            "data": data,
        })

async def traced_agent(task: str):
    trace = {"id": str(uuid.uuid4()), "task": task, "spans": []}
    token = current_trace.set(trace)
    try:
        return await run_agent(task), trace
    finally:
        current_trace.reset(token)
        # 落库：每条 trace = 一次完整调用的「飞行记录仪」
        await db.insert_trace(trace)

开箱方案：Langfuse / LangSmith（自建选 Langfuse，开源），SDK 埋点几行接入，白送调用链可视化、token 统计、人工标注、prompt 版本管理。

排查示范——「客服 agent 为什么答错」：

打开 trace → 看工具调用链：
  llm: 用户问退款 → tool_use: refund_policy(query="退货")
  tool_result: 返回的是「退货」政策而非「退款」政策 ← 问题在这
  llm: 基于错误文档回答 → 用户收到错误答案
结论：搜索工具的 description 没区分退款/退货 → 改描述 → eval 回归

四、评估回归：改任何东西都要跑

改动前先问：怎么证明没变坏？
  改了系统提示 / 工具描述 / 换模型 → 全都可能让行为退化

eval 集最小形态（一个 JSON 文件起步）：
[
  {"task": "查订单 A1024 状态", "assert_tool": "query_order", "assert_contains": "已发货"},
  {"task": "用户要退款超过 5000", "assert_contains": "人工"},
  {"task": "闲聊打招呼", "assert_no_tool": true}
]

跑法：CI 里对 eval 集逐条执行 agent，断言通过率
阈值：低于基线 → 阻断合并

评估方法（精确断言 / LLM-as-judge / 人工抽查）与安全（prompt injection 纵深防御）在 Harness 评估安全篇有完整展开——生产化必读。

五、成本护栏

class Budget:
    """三层闸：轮数 / token / 时间"""
    def __init__(self, max_turns=20, max_tokens=200_000, max_seconds=300):
        self.max_turns, self.max_tokens, self.max_seconds = max_turns, max_tokens, max_seconds
        self.used_tokens = 0

    def check(self, turn: int, elapsed: float) -> str | None:
        if turn >= self.max_turns: return "max_turns"
        if self.used_tokens >= self.max_tokens: return "max_tokens"
        if elapsed >= self.max_seconds: return "timeout"
        return None

# agent loop 每轮开头：
reason = budget.check(turn, elapsed)
if reason:
    # 优雅停止：让模型总结进度，而不是硬截断
    final = await call_llm(system=SYSTEM, messages=messages + [
        {"role": "user", "content": f"预算用尽（{reason}），总结目前进展与未完成事项。"}
    ])
    return final

省钱杠杆按 ROI 排序：prompt caching（前缀稳定）> 工具输出截断 > 小模型分级 > 压缩触发。多轮 agent 会话里 caching 命中率常到 80%+，费用直接砍一个数量级。

六、上线检查清单

□ 流式：SSE 接口 + 前端增量渲染 + 工具调用事件
□ 限流：全局并发 / 用户级 RPS / 会话级排队
□ 降级：重试 → 备用模型 → 兜底话术
□ 观测：tracing 落库（或 Langfuse）+ token 费用打点
□ 评估：eval 集 + CI 阈值阻断
□ 预算：轮数/token/时间三闸 + 优雅停止
□ 安全：工具白名单、写操作审批、prompt injection 防护
□ 数据：会话保留策略、脱敏、用户删除权（合规）
□ 灰度：新 prompt/模型先 5% 流量，观测指标再全量

本篇小结

交付形态：SSE 流式服务，事件分类型（text/tool/done）推给前端
限流三件套：全局并发、用户 RPS、会话串行；降级链路必备
tracing 是排查 agent 的唯一手段：最小自建或直接上 Langfuse
eval 回归 + 成本三闸 + 安全清单 = 敢上生产的三个签名

下一篇：综合实战项目——用三个端到端项目串起全部知识。