AI Agent 运行全流程：从用户输入到答案返回的完整链路

做了一年多的 AI Agent 系统，最大的认知升级是：「调用 LLM API」和「构建 Agent 系统」是两件完全不同的事。
一句话 API 拿到结果很容易，但要做一个稳定、可观测、可降级的 Agent 系统，需要串起 7-8 个子系统，每个都有自己的坑。

这篇文章把一个 Agent 从用户敲下回车那一刻到最终答案出现在屏幕上那一刻的完整链路拆开讲。读完你应该能：

• 画出任意 Agent 系统的核心数据流
• 说出每个环节的关键决策点和常见 bug
• 独立设计一个生产级 Agent 系统

---

一、整体架构：一张图看懂

┌──────────────────────────────────────────────────────────────┐
│                         用户输入                              │
│                   "帮我看看上季度销售数据"                     │
└──────────────────────────┬───────────────────────────────────┘
                           │
                           ▼
┌──────────────────────────────────────────────────────────────┐
│  ① 输入层        │  Webhook / SSE / IM / API                  │
│  (Ingress)       │  + 鉴权 + 限流 + 参数校验                    │
└──────────────────────────┬───────────────────────────────────┘
                           │
                           ▼
┌──────────────────────────────────────────────────────────────┐
│  ② 预处理        │  - 注入 system prompt / Skills            │
│  (Preprocessing) │  - 检索 RAG 知识片段                       │
│                  │  - 加载长期记忆摘要                        │
└──────────────────────────┬───────────────────────────────────┘
                           │
                           ▼
┌──────────────────────────────────────────────────────────────┐
│  ③ Agent Loop   │   意图识别 → 计划拆解 → 选工具 → 执行       │
│  (核心循环)      │   ↓ 失败重试 / 换路径                        │
│                  │   反复循环直到任务完成或达到 max_iter       │
└──────────────────────────┬───────────────────────────────────┘
                           │
                           ▼
┌──────────────────────────────────────────────────────────────┐
│  ④ 工具执行层    │   MCP / Function Call / HTTP API            │
│  (Tools)         │   + 沙箱 + 超时 + 重试 + 权限校验            │
└──────────────────────────┬───────────────────────────────────┘
                           │
                           ▼
┌──────────────────────────────────────────────────────────────┐
│  ⑤ 后处理        │   - 结果解析 + 错误处理                     │
│  (Postprocess)   │   - 格式化 + 安全检查                       │
│                  │   - 敏感词过滤 / PII 脱敏                    │
└──────────────────────────┬───────────────────────────────────┘
                           │
                           ▼
┌──────────────────────────────────────────────────────────────┐
│  ⑥ 输出层        │   SSE 流式推送 / WebSocket / 一次性返回    │
│  (Egress)        │   + 存储对话 + 更新记忆 + 埋点              │
└──────────────────────────┬───────────────────────────────────┘
                           │
                           ▼
                      用户看到答案

核心循环就是：第 ② 步到第 ④ 步反复迭代，直到任务完成。

---

二、第①步：输入层（Ingress）

用户消息进来后的第一道关卡。这一层看起来简单，但其实最容易出 bug。

2.1 多渠道接入

渠道            协议                  注意点
─────────────────────────────────────────────────
Web Chat        WebSocket / SSE        长连接、鉴权、断线重连
Slack/飞书      Webhook + Signature    必须验签防伪造
API 调用       HTTPS POST            限流 + 配额
IDE 插件       本地 stdin/socket      离线场景

2.2 必做事项

async def ingress_handler(request: Request):
    # ① 鉴权（按渠道不同）
    user = await authenticate(request)

    # ② 限流（防滥用 + 成本控制）
    if not rate_limiter.allow(user.id):
        raise HTTPException(429, "Too Many Requests")

    # ③ 参数校验
    payload = IngressPayload(
        message=request.message.strip(),     # 去前后空格
        user_id=user.id,
        session_id=request.session_id,
        # ... 渠道元数据
    )
    return await agent_run(payload)

常见坑：

• ❌ 信任前端传来的 user_id（伪造身份）→ ✅ 必须从 session token 解析
• ❌ 不限流 → 一个人写个 for 循环能把月账单刷爆
• ❌ 不校验消息长度 → 单条 10MB 消息把 LLM 撑爆

---

三、第②步：预处理（Preprocessing）

把”裸消息”变成 Agent 能用的上下文。

3.1 上下文组装流水线

用户消息："看看上季度销售数据"
         │
         ▼
┌─────────────────────────────────────┐
│  Layer 1: System Prompt              │  ← Agent 角色、技能、约束
│  "你是一个数据分析助手..."           │
└─────────────────────────────────────┘
         +
┌─────────────────────────────────────┐
│  Layer 2: Skills（技能描述）         │  ← "用 sql_query 查数据库"
│  "## sql_query                       │
│   - 输入：表名 + 字段                │
│   - 输出：JSON 数据"                │
└─────────────────────────────────────┘
         +
┌─────────────────────────────────────┐
│  Layer 3: RAG 检索（短期记忆）        │  ← "用户之前查过..."
│  相关文档片段 × 3-5                  │
└─────────────────────────────────────┘
         +
┌─────────────────────────────────────┐
│  Layer 4: 用户长期记忆摘要           │  ← "用户偏好表格视图"
│  [memory_summary]                    │
└─────────────────────────────────────┘
         +
┌─────────────────────────────────────┐
│  Layer 5: 对话历史（最近 N 轮）       │  ← 最近 5-10 轮对话
└─────────────────────────────────────┘
         +
┌─────────────────────────────────────┐
│  Layer 6: 当前用户消息              │
└─────────────────────────────────────┘
         │
         ▼
   完整 context（送 LLM）

3.2 Token 预算控制

每层都要有 token 限制，避免上下文爆炸：

def build_context(user_msg: str, session_id: str) -> Context:
    return Context(
        system=truncate(SYSTEM_PROMPT, max_tokens=500),
        skills=truncate(SKILLS, max_tokens=800),
        rag_docs=top_k(query=user_msg, k=3, max_tokens=1500),
        memory=load_user_memory(user_id, max_tokens=300),
        history=truncate(load_history(session_id, last_n=10), max_tokens=2000),
        current=user_msg,
    )

经典 80/20 法则：

• System + Skills：固定 ~1300 tokens（5%）
• RAG + 记忆：~1800 tokens（10%）
• 历史：~2000 tokens（15%）
• 留给 LLM 输出的预算：~10000 tokens（70%）

---

四、第③步：Agent Loop（核心循环）

这是整个系统的大脑。Agent Loop 是一个 while 循环：

async def agent_loop(context: Context, max_iter: int = 25):
    for iteration in range(max_iter):
        # 调用 LLM，让它决定下一步
        response = await llm.chat(
            model="claude-sonnet-4-6",
            messages=context.messages,
            tools=context.tools,
            temperature=0,
        )

        # 情况 1: LLM 认为任务完成
        if response.finish_reason == "end_turn":
            return response.text

        # 情况 2: LLM 想调用工具
        if response.tool_calls:
            for call in response.tool_calls:
                result = await execute_tool(call, context)
                context.add_tool_result(call.id, result)
            # 继续循环，让 LLM 看结果后做下一步决策
            continue

        # 情况 3: LLM 输出格式异常
        log.warning(f"Unexpected: {response}")
        return response.text or "抱歉，我没理解。"

4.1 三大推理模式

Agent Loop 的本质是 让 LLM 推理。主流有 3 种：

A. ReAct（Reason + Act）

Thought: 需要先知道有哪些表
Action: list_tables()
Observation: [orders, products, users]

Thought: 找到 orders 表，查询 2025 Q3 数据
Action: sql_query("SELECT * FROM orders WHERE quarter='2025Q3'")
Observation: [{order_id: 1, amount: 1200}, ...]

Thought: 数据够了，整理成表格
Action: finish("2025 Q3 销售额是 ¥1,234,567...")

优点：每步都有推理，可解释性强
缺点：步骤多，token 消耗大

B. ReWOO（Reasoning WithOut Observation）

Plan:
- Step 1: list_tables()
- Step 2: sql_query("SELECT ...")
- Step 3: format_response()

Execute Step 1: [orders, products, users]
Execute Step 2: [{order_id: 1, amount: 1200}, ...]
Execute Step 3: 完成

优点：一次性规划完，省 token
缺点：plan 出错就全错，没法中途纠正

C. Plan-and-Execute（混合模式）

Plan: 先看表 → 查数据 → 总结（3 步）
Execute Step 1: [orders, products, users]
Observe: 看到 orders 表
Re-plan: 改用聚合查询
Execute Step 2: [...]
Final Answer: ...

优点：兼顾 ReAct 的灵活 + ReWOO 的效率
缺点：实现复杂度最高

4.2 工具调用格式

现代 LLM 用 OpenAI Function Calling 或 Anthropic Tool Use 格式：

TOOLS = [
    {
        "name": "sql_query",
        "description": "执行 SQL 查询并返回结果。表结构：orders(id, amount, date)",
        "input_schema": {
            "type": "object",
            "properties": {
                "sql": {"type": "string", "description": "SELECT 语句"},
                "limit": {"type": "integer", "default": 100}
            },
            "required": ["sql"]
        }
    },
    {
        "name": "send_email",
        "description": "发送邮件给指定收件人",
        "input_schema": {
            "type": "object",
            "properties": {
                "to": {"type": "string"},
                "subject": {"type": "string"},
                "body": {"type": "string"}
            },
            "required": ["to", "subject", "body"]
        }
    }
]

Tool schema 设计的 3 个黄金法则：

1. description 要具体（不是”查数据库”而是”查订单表”）
2. 限制返回行数（避免 100 万行的结果）
3. 标记副作用（写操作要二次确认）

---

五、第④步：工具执行层（Tools）

LLM 决定要调工具后，真正的代码执行在这一层。

5.1 工具执行流水线

工具调用请求
    │
    ▼
┌──────────────────┐
│ ① 权限校验        │  这个用户有权调这个工具吗？
└────────┬─────────┘
         ▼
┌──────────────────┐
│ ② 参数校验        │  schema 是否匹配？
└────────┬─────────┘
         ▼
┌──────────────────┐
│ ③ 超时控制        │  30 秒未返回就 kill
└────────┬─────────┘
         ▼
┌──────────────────┐
│ ④ 执行（沙箱）    │  隔离环境跑（Docker/Firecracker）
└────────┬─────────┘
         ▼
┌──────────────────┐
│ ⑤ 结果清洗        │  去掉敏感信息 + 截断过长输出
└────────┬─────────┘
         ▼
      返回 LLM

5.2 工具沙箱设计

为什么需要沙箱？

LLM 可能生成恶意 SQL、读敏感文件、执行 rm -rf /，必须隔离。

class ToolSandbox:
    async def execute(self, tool_call, user):
        # 网络隔离
        # CPU/内存限制
        # 文件系统只读（白名单路径）
        # 环境变量脱敏
        # 执行时间限制
        
        try:
            async with timeout(30):
                result = await asyncio.create_subprocess_exec(
                    tool_call.command,
                    env=sanitized_env,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                )
                stdout, stderr = await result.communicate()
                return SandboxResult(stdout=stdout[:10000], stderr=stderr)
        except TimeoutError:
            return SandboxResult(error="执行超时（30s）")

5.3 工具结果的处理

工具返回 10MB 数据怎么办？

def process_tool_result(result: str, max_tokens: int = 2000) -> str:
    # ① 截断（保留头尾 + 摘要中间）
    if token_count(result) > max_tokens:
        result = summarize(result, max_tokens)

    # ② 持久化到对象存储，返回引用 ID
    if token_count(result) > 10000:
        ref_id = save_to_object_storage(result)
        return f"结果太大，已存到 {ref_id}（用 get_result 工具查看）"

    return result

**这就是为什么大多数 Agent 系统都有第二个工具 “get_more_details”**——LLM 第一次拿到摘要，必要时再调工具拿详情。

---

六、第⑤步：后处理（Postprocessing）

工具结果 → LLM 最终答案 → 用户能看到的内容。

6.1 结果格式化

class OutputFormatter:
    def format(self, llm_response: str, format_type: str) -> str:
        if format_type == "markdown":
            return llm_response
        elif format_type == "json":
            return self.extract_json(llm_response)
        elif format_type == "table":
            return self.markdown_to_html_table(llm_response)

6.2 安全过滤

LLM 输出可能包含：

• 敏感信息（用户手机号、内部 API key）
• 不当内容（违规、政治敏感）
• 提示注入（用户诱导 LLM 输出恶意内容）

def safety_filter(text: str) -> str:
    # PII 脱敏
    text = mask_phone(text)
    text = mask_email(text)
    
    # 内容审核
    if contains_violation(text):
        return "抱歉，我无法回答这个问题。"
    
    # 反提示注入（检测 "忽略之前所有指令"）
    if detect_prompt_injection(text):
        log_security_event(text)
        return "已记录，请重新提问。"
    
    return text

---

七、第⑥步：输出层（Egress）

用户最终看到的体验就在这一层。

7.1 流式 vs 非流式

流式（SSE / WebSocket）：

用户问 → 1.5s 后开始看到第一个字 → 持续打字效果 → 6s 后完整

非流式（HTTP）：

用户问 → 6s 后一次性看到全部

99% 的场景应该用流式，体验差异巨大。

async def stream_response(request):
    async def event_generator():
        async for chunk in agent_run_streaming(...):
            yield f"data: {json.dumps(chunk)}\n\n"
        yield "data: [DONE]\n\n"

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

7.2 后台任务

有些任务不能流式（比如跑 5 分钟的 SQL 查询）：

用户提交 → 立即返回 task_id → 后台跑 → 完成后通知（WebSocket/SSE）

@app.post("/tasks")
async def create_task(request):
    task_id = generate_uuid()
    background_queue.add(agent_run, task_id, request)
    return {"task_id": task_id, "status": "pending"}

@app.get("/tasks/{task_id}/status")
async def get_status(task_id):
    return {"status": "running", "progress": "60%"}

7.3 持久化 + 记忆

每次对话结束都写库：

async def on_session_end(session_id, user_id):
    # ① 存储完整对话（用于回看）
    save_conversation(session_id, full_messages)
    
    # ② 提取关键事实（用于下次上下文）
    facts = await llm.extract_facts(messages)
    save_user_facts(user_id, facts)  # 比如 "用户偏好表格视图"
    
    # ③ 更新记忆摘要
    memory = await llm.summarize_history(user_id)
    save_user_memory(user_id, memory, max_tokens=300)

---

八、Context 压缩：必做的优化

随着对话进行，context 会越来越长。超过 32K tokens 后 LLM 开始变慢变贵，必须压缩。

8.1 三级压缩策略

┌────────────────────────────┐
│ L1: 滑动窗口                  │  ← 保留最近 N 轮
│   永远做                       │
└────────────────────────────┘
            ↓ 仍然超长
┌────────────────────────────┐
│ L2: 摘要压缩                  │  ← 把早期消息总结成 200 字
│   超过 50% 时做                │
└────────────────────────────┘
            ↓ 仍然超长
┌────────────────────────────┐
│ L3: Compaction（深度重组）    │  ← 重新生成完整对话
│   接近上限时做                 │
└────────────────────────────┘

8.2 实战代码

class ContextManager:
    def maybe_compress(self, messages: list[Message]) -> list[Message]:
        total_tokens = count_tokens(messages)
        limit = 30000

        if total_tokens < limit * 0.5:
            return messages  # 不需要压缩

        if total_tokens < limit * 0.8:
            # L1: 滑动窗口（保留最近 10 轮）
            return messages[-20:]

        if total_tokens < limit:
            # L2: 摘要早期消息
            early = messages[:-10]
            summary = llm.summarize(early, max_tokens=200)
            return [Message.system(f"之前的对话摘要：{summary}")] + messages[-10:]

        # L3: 完整 Compaction
        return llm.compact(messages, target_tokens=limit // 2)

Compaction 是 Claude Agent SDK 等成熟框架的核心技术——比简单滑动窗口效果好 30%+。

---

九、可观测性：让 Agent 系统可调试

没有可观测性 = 黑色盒子。生产 Agent 必须有以下埋点：

# 1. 全链路 Trace
@traceable(name="agent.session")
async def agent_run(session_id, query):
    # LangSmith / LangFuse / 自建 trace
    ...

# 2. 每步 LLM 调用
@traceable(name="llm.call")
async def llm_call(messages, tools):
    span = current_span()
    span.set_attribute("input.tokens", count_tokens(messages))
    span.set_attribute("model", model_name)
    span.set_attribute("cost_usd", calc_cost(...))
    ...

# 3. 每个工具调用
@traceable(name="tool.execute")
async def execute_tool(name, args):
    span = current_span()
    span.set_attribute("tool.name", name)
    span.set_attribute("tool.duration_ms", ...)
    span.set_attribute("tool.success", success)

LangSmith 是行业标准——花 10 分钟接入就能看到完整 trace。

---

十、容错降级：失败时的兜底

LLM 调用可能失败、工具可能超时、网络可能断。必须有降级链。

async def agent_with_fallback(query):
    # 第 1 层：主 Agent
    try:
        return await run_with_llm(query, "claude-sonnet-4-6")
    except (TimeoutError, RateLimitError, ModelError):
        log.warning("主 Agent 失败，降级")

    # 第 2 层：备用 LLM
    try:
        return await run_with_llm(query, "gpt-4o")
    except Exception:
        log.warning("备用 LLM 也失败")

    # 第 3 层：纯 RAG（不调工具）
    docs = await rag_search(query)
    return "抱歉，详细分析暂时不可用。以下是相关文档：\n" + format_docs(docs)

    # 第 4 层：兜底提示
    return "系统繁忙，请稍后再试。"

降级原则：用户体验不能断。即使 LLM 全挂了，也要给用户一个有价值的回复（RAG 结果 / 静态提示）。

---

十一、完整链路示例

以「帮我看看上季度销售数据」为例，看数据流转：

# 1. ingress: 接收
message = IngressPayload(
    message="帮我看看上季度销售数据",
    user_id="u_123",
    session_id="s_456",
    channel="web_chat",
)

# 2. preprocess: 组装上下文
context = Context(
    system="你是销售数据分析师...",
    skills=load_skills(["sql_query", "format_chart"]),
    rag_docs=search("Q3 销售"),
    memory=load_user_memory("u_123"),  # "用户偏好表格"
    history=load_history("s_456"),  # 最近 5 轮
    current=message.message,
)

# 3. agent_loop: 推理循环
for i in range(25):
    response = await llm.chat(context)
    if response.finish_reason == "end_turn":
        final_answer = response.text
        break
    for call in response.tool_calls:
        result = await tools.execute(call)
        context.add_tool_result(call.id, result)

# 4. postprocess: 清洗 + 格式化
final = safety_filter(final_answer)

# 5. egress: 流式返回给用户
yield {"event": "token", "data": token}
yield {"event": "done", "data": final}

用户视角：3 秒后看到第一个字，5 秒后看到完整答案 + 表格。

---

十二、关键指标：上线后看什么

指标	健康值	异常处理
任务完成率	> 90%	< 80% 检查 prompt
平均 Token/任务	< 5000	> 10000 检查上下文
P50 响应延迟	< 3s	> 10s 换小模型
P99 延迟	< 30s	> 60s 加超时熔断
成本/任务	< $0.05	> $0.2 启用 cache
用户重试率	< 10%	> 20% 检查答案质量
Tool 失败率	< 5%	> 15% 修工具

---

十三、常见误区（避坑指南）

❌ 误区 1：让 LLM 干所有事

"帮我写个爬虫" → LLM 自己爬

✅ 应该：LLM 写代码 → 工具执行代码 → LLM 解释结果

❌ 误区 2：把所有 Skills 都塞进 system prompt

把 100 个工具的 description 全塞进去 → 烧 10K tokens

✅ 应该：Skill 匹配（RAG 检索）→ 只注入相关的 3-5 个

❌ 误区 3：忽视工具结果的可信度

工具返回 HTML 页面 → 直接喂给 LLM → LLM 跟着执行里面的 JS

✅ 应该：工具结果当成不可信输入，做安全清洗后再喂 LLM

❌ 误区 4：Context 越大越好

把整个数据库 schema + 所有历史对话塞进去

✅ 应该：精准相关——80% 任务用 5K tokens 就够

❌ 误区 5：不做 Trace

"Agent 答错了，但不知道为什么"

✅ 必须：全链路 trace，每一步 LLM 输入输出 + 工具调用 + 耗时

---

十四、上线 Checklist

• 输入层鉴权 + 参数校验
• 限流 + 防刷
• 全链路埋点
• 工具调用沙箱
• 超时熔断
• 降级链（3 层）
• Context压缩
• PII 脱敏
• 内容审核
• 流式响应
• 持久化对话
• 监控告警

---

十五、推荐工具栈

组件	推荐	原因
Agent 框架	Claude Agent SDK / LangGraph	成熟 + 可观测
LLM	Claude Sonnet 4.6（主）+ GPT-4o（备）	推理强 + 降级
RAG	Qdrant / pgvector	性能 + 成本
可观测性	LangSmith / LangFuse	行业标准
工具沙箱	Docker / Firecracker	安全
消息队列	Redis Streams / RabbitMQ	异步任务
状态存储	PostgreSQL	持久化

---

十六、最后一句话

Agent 系统不是 LLM API 的简单封装，而是 多个子系统协同的分布式系统：

• 输入层决定谁能用
• 预处理决定LLM 看到什么
• Agent Loop 是大脑
• 工具层是手
• 输出层决定用户感受

每一个子系统都有自己的坑，但把它们串起来，就是 Agent 工程的核心能力。

希望这篇文章能让你少走一些弯路。下次面 Agent 架构设计题时，能画出这张图 + 说清楚每层选型理由。

---

作者：魏远标，贝斯平 AI 架构师，13 年 Java / AI 一线经验，最近 2 年专注 Agent 工程化。
博客：javai.tech
反馈：如果你也在做 Agent 系统，欢迎加微信 sherwin28 一起聊聊~

---

相关文章推荐：

• 《Claude Agent SDK 架构实践：从 Skill 到 MCP，构建多 Agent 协同系统》
• 《LLM 评估实战：Ragas + LangSmith 在政策 RAG 系统的落地》
• 《爬虫架构的三代演进：从 cheerio 到 Agent 再到 Crawl4AI》