综合实战项目

三个由浅入深的端到端项目，把系列全部知识串起来。每个项目给出：架构选型理由 → 关键代码骨架 → 踩坑点。

项目一：智能客服 Agent（Router + 工具 + 记忆）

需求：用户咨询订单、退款、商品问题；能记住用户偏好。

架构

用户消息 → 路由器（小模型，结构化输出）
            ├─ order     → 订单 agent（查单/物流工具）
            ├─ refund    → 退款 agent（查单 + 退款工具 + 金额审批闸）
            ├─ product   → 商品 agent（搜索 + 知识库 RAG）
            └─ chitchat  → 直接小模型回复（不走 agent loop）

共享层：SessionStore（会话记忆）+ UserFacts（长期画像注入）

为什么 Router 而不是单 agent：三类问题工具集完全不同、提示词风格不同；路由层还能把闲聊拦在 agent loop 外（省钱 90%）。

关键代码

async def handle(session_id: str, text: str) -> AsyncIterator[str]:
    messages = SessionStore(session_id).load()
    facts = await get_user_facts(session_id)     # "偏好简洁回复"

    route = await route_intent(text)             # 04 篇的结构化输出
    agent = AGENTS[route.target]                 # 各自的 system + tools

    messages.append({"role": "user", "content": text})
    async for delta in run_agent(agent, messages, user_profile=facts):
        yield delta                               # SSE 直通前端

    SessionStore(session_id).save(messages)

踩坑点

1. 退款金额 >5000 → 工具内直接返回「需人工」，别指望提示词拦
   （提示词是「请求」，工具校验才是「强制」）
2. 会话压缩丢关键信息 → 订单号、投诉状态单独结构化存储
3. 路由错误率 3% → 加「兜底分支」：低置信度时问用户一句再路由

项目二：数据分析 Agent（工具 + 结构化输出 + 安全）

需求：自然语言查数据：「上个月华南区销售额 top5 的商品」。

架构

用户问题 → 数据 agent（单一 agent，不需要多 agent）
  工具 1: list_tables / describe_table（让模型先摸清库表）
  工具 2: run_sql（只读连接 + SELECT 白名单 + LIMIT 强制）
  工具 3: chart_config（输出图表配置结构化数据）
→ 返回：结论文本 + 数据表格 + 图表配置（前端 ECharts 渲染）

关键代码

RUN_SQL_DESC = """在分析库执行只读 SQL 查询。规则：
1. 只允许 SELECT；禁止任何写操作
2. 必须带 LIMIT（默认 100，最大 1000）
3. 表结构先用 describe_table 了解，不要凭猜测写 SQL
4. 时间字段格式：created_at TIMESTAMP"""

async def run_sql(sql: str) -> str:
    if not sql.strip().upper().startswith("SELECT"):
        return "ERROR: 只允许 SELECT"
    if "LIMIT" not in sql.upper():
        sql += " LIMIT 100"                       # 强制兜底
    rows = await readonly_conn.fetch(sql)         # 只读账号连接
    return json.dumps([dict(r) for r in rows], default=str)[:20_000]

踩坑点

1. 模型猜列名 → 「先 describe 再写 SQL」写进系统提示，错误率大降
2. 大结果撑爆上下文 → 工具层截断 + 描述里写明「聚合后再取明细」
3. 安全底线：数据库用只读账号（工具校验是第二道，账号权限是第一道）
4. 图表：让模型输出 ECharts option 的 JSON（schema 约束），前端零逻辑

项目三：代码审查 Bot（Pipeline + CI 集成）

需求：PR 提交后自动审查，发现正确性/安全问题并留评论。

架构

PR webhook → 收集 diff（确定性代码，不用模型）
→ Stage 1 理解：agent 读改动文件与上下文（read/grep 工具）
→ Stage 2 审查：三个并行子代理（正确性 / 安全 / 测试覆盖）
→ Stage 3 裁决：汇总 agent 过滤误报、排序、生成评论（file:line 格式）
→ GitHub PR comment

为什么是 Pipeline + Fan-out 而不是单 agent：三个维度独立且并行（wall-clock = max 而非 sum）；汇总层压误报——单 agent 审查的误报率会淹没开发者。

关键代码

async def review_pr(pr) -> str:
    diff = await get_diff(pr)                     # 确定性步骤：纯代码
    changed = await get_changed_files(pr)

    # 三个维度并行（06 篇 fan-out）
    findings = await asyncio.gather(
        run_agent(SYSTEM_CORRECTNESS, prompt_for(diff, changed),
                  tools=[Read, Grep]),
        run_agent(SYSTEM_SECURITY, prompt_for(diff, changed),
                  tools=[Read, Grep]),
        run_agent(SYSTEM_TESTS, prompt_for(diff, changed),
                  tools=[Read, Grep]),
    )

    # 裁决层：过滤误报、去重、排序
    return await run_agent(
        "你是首席审查官。以下是三个审查员的发现，合并去重，"
        "剔除无法确认的问题，按严重度排序，引用 file:line。",
        "\n\n".join(findings), tools=[])

踩坑点

1. 「无法确认的问题」是评论噪音的主要来源
   → 审查提示词写死：只报告你能指出具体失败场景的问题
2. 大 PR 撑爆上下文 → 按 diff 分块审查再汇总
3. CI 时长敏感 → 子代理模型分级：正确性用旗舰，测试覆盖用小模型
4. 误报成本不对称 → 宁可漏报不要误报，开发者信任一旦失去就完了

三个项目的共性总结

环节	客服	数据分析	审查 Bot
编排模式	Router	单 agent	Pipeline + Fan-out
结构化输出	路由判断	图表配置	无（自然语言评论）
记忆	会话 + 画像	无（单任务）	无（单任务）
安全闸	退款金额审批	只读账号	只读工具
确定性部分	会话管理	SQL 执行	diff 收集/评论发布

提炼出的开发方法论：

1. 能用代码的不用模型（diff 收集、SQL 执行、路由后的业务逻辑）
2. 自主性取最低够用档（三个项目没有一个需要 supervisor 全自主）
3. 安全线放在工具层和基础设施层，不放在提示词层
4. 每个项目第一天就有：eval 集 + tracing + 预算闸

系列总结

篇目	一句话
全景与路线图	自主性阶梯：能低不高，agent 的灵活是拿可控换的
第一个 Agent	50 行 loop 是一切 agent 的骨架
工具开发实战	工具质量决定上限，description 就是 API 文档
结构化输出	schema 约束输出 + 校验失败回注重试
记忆与状态	四层记忆按需叠加，并发会话要加锁
多 Agent 编排	拆分是为了可控：Router→Pipeline→Supervisor
生产化工程	流式 + 限流 + tracing + eval + 预算三闸
综合实战项目（本篇）	能用代码的不用模型，安全放工具层

相关笔记：AI Agent Harness 系列 · LangChain/LangGraph · RAG 教程 · AI Agent 基础