企业官网建设流程全解析

1. 从"工具调用"到"自主行动"：为什么要聊 Agent 的工程实现

近一年我试过大量号称"Agent"的项目，说实话，大部分其实就是 LLM 套了一层工具调用的壳——你问一句它答一句，最多帮你查个天气、算个算术，离真正"干活"还有十万八千里。直到我接触了 AI Agent 的完整工程落地，才意识到一个关键问题：Agent 不是一个算法问题，而是一个系统工程问题。你不需要再追着热词跑，你需要一套能落地的框架。

这篇文章要聊的，就是我从实际开发中总结出来的 Agent 工程化方法论。很多资料都在讲"什么是 Agent"或者"Agent 能做什么"，但一旦你真正动手，就会发现大量问题浮出水面：状态怎么维护？循环怎么终止？模型上下文怎么管理？并发一上来怎么扛？这些都是文档里不会写、但工程上必须面对的事。

我把它拆成两个层面来讲：设计时的"七要素"，以及实现时的"七个决策点"。七要素帮你把 Agent 从概念描述变成可执行的设计蓝图；七个决策点则是你真正写代码时绕不开的分岔路口。这套东西适合刚接触 Agent 开发、想从零搭一套可用系统的工程师，也适合那些已经在用 LangChain、LangGraph、FastAPI 搭过 Demo、但总感觉"跑起来容易、扛不住真实场景"的朋友。它同样适用于想用 Rust 这类高并发语言重新实现 Agent 内核的开发者——因为核心的架构思路是语言无关的。

先解释一个容易混淆的概念：Agent 和传统程序的区别，不完全在于"能调用工具"——传统程序也能调用 API。真正的分水岭是决策权。传统程序里，调用哪个 API、什么时候调用、参数怎么填，都是代码写死的。Agent 则把这一系列操作的选择权交给了模型：模型根据用户的原始目标，自己规划步骤、选择工具、解析结果、决定下一步是继续还是收尾。这种由"执行者"变成"决策者"的转变，才是 Agent 的核心本质。

但你马上面临一个现实问题：模型是概率性的，代码是确定性的。把决策权交给一个"每次输出可能不一样的系统"，工程上怎么收敛？怎么保证不跑飞？怎么控制成本和时间？这就是后面要展开的"七要素"和"七个决策点"存在的意义。

2. 设计的起点：解构 Agent 的七要素

七要素是我从多个实际项目中抽象出来的设计骨架。它不是学术定义，而是"动手前必须想清楚的七个维度"。这七个维度分别是：目标定义、模型选型、工具集设计、记忆机制、上下文管理、状态管理、安全与边界。下面逐一拆开讲。

2.1 模型选型：不要只看"聪明不聪明"

模型选型是七要素里最容易被低估的一环。很多人第一反应是"选最强的模型"，但工程上是反过来——选最合适的。我来拆一拆不同模型在 Agent 场景里的真实差异。

首先是推理能力和工具调用格式的稳定性。Agent 的核心循环依赖模型输出结构化的工具调用指令，比如 function calling 的 JSON 格式。如果模型输出经常格式漂移，你后面解析逻辑会写得非常痛苦。实测下来，Claude 的 function calling 稳定性比多数开源模型好一个量级；GPT 系列对复杂工具参数的生成也比较可靠。开源模型里 Qwen 系列在工具调用上做得比较努力，但遇到复杂场景还是会有"想当然"式地捏造参数。

其次是延迟和成本。Agent 不是一轮问答，它是一个多轮循环。一次完整任务可能要调用模型 5~10 次甚至更多。如果单次延迟 3 秒，整个任务就是半分钟起步，用户的体感会非常差。所以很多线上 Agent 会把任务拆"轻"——简单子任务用小模型，复杂决策才换大模型。这就是所谓的"模型路由"策略，效果比单一模型要好得多，成本也能压到原来的三分之一左右。

还有一个经常被忽略的点：上下文窗口的消耗方式。同样 128K 上下文的模型，有的模型在长上下文下性能衰减严重。Agent 场景里，历史记录、工具返回结果、中间推理过程都会占用大量 token。选模型时不要只看"最长上下文"，要看"有效上下文"——就是上下文拉长之后，模型还能不能保持稳定推理。从我测试看，某些模型在上下文超过 60% 时，工具调用准确率明显下降，这在 Agent 场景里是致命的。

2.2 记忆机制：短期工作记忆和长期记忆要分开

记忆是 Agent 从"能用"到"好用"的分水岭。但记忆不是简单地"把历史对话丢给模型"，它至少分成两层。

短期工作记忆对应的是当前任务执行过程中的上下文，通常由程序管理（比如 LangGraph 里的 State，或者你自己定义的 Session 对象），它决定 Agent 当前步骤能"看到"什么。短期记忆的关键是控制长度——塞得越多，模型的注意力就越分散，还烧钱。常见的做法是"关键信息保留，过程信息压缩"。比如用户的最初目标、已经确认的事实必须原样保留；中间步骤的详细输出可以做摘要。

长期记忆则解决跨会话的问题。没有长期记忆的 Agent 每次对话都是"失忆"的，用户得反复交代背景，这体验极差。工程上用向量数据库（如 Chroma、Milvus、pgvector）存储历史关键信息，通过 embeddings 在对话开始时做检索召回。这里有个容易被坑的地方：不要把所有历史都灌进向量库，要按会话语义做分层——有的信息是一次性的，有的信息是长期偏好。比如"用户今天想订去上海的机票"是一次性意图；"用户习惯坐靠窗位置"是长期偏好。两者混在一起，检索质量会下降。

另外一个记忆的关键问题是"过期失效"。实际项目里，用户的偏好会变，昨天的决定今天可能推翻。所以长期记忆必须设计成"可更新、可失效"的，而不是只增不改。比较好的实践是给每条记忆加时间戳和置信度字段，召回时做加权，在 Agent 使用记忆时也可以附带提示"这是用户三个月前的偏好，仅供参考"，避免模型把旧信息当新指令执行。

2.3 工具集与状态管理：Agent 的"手"和"体"

工具集是 Agent 能执行动作的"手"。这里的工程问题不在于工具多不多，而在于工具的描述质量。模型是靠着工具的描述来决定调用哪个工具的，描述写得模糊，模型就瞎猜；描述写得精确，模型一选一个准。工具描述里必须包含：工具功能的一句话说明、关键参数的含义、适用的场景、不适合的场景。实践下来，给工具加"典型的调用示例"效果极佳，因为模型在 few-shot 下理解力提升幅度很明显。

状态管理经常被人忽略，但它其实是 Agent 工程中最容易出 bug的地方。Agent 是有状态的系统：它执行到一半，用户打断插话怎么办？工具返回失败，重试还是换路径？任务超时，怎么保留现场以便恢复？这些都需要状态管理来兜底。我建议用显式的状态机来管理 Agent 生命周期，而不是靠一堆布尔变量硬扛。状态至少包括：等待用户输入、正在推理、正在执行工具、任务完成、任务失败、需要人工介入。每个状态定义清楚可允许的转移路径，这个 Agent 才称得上"可控"。

从工程实现角度，状态管理可以依托 LangGraph 的 StateGraph，也可以自己用 Python 写一个状态机类，核心是把每一步的输入、输出、决策依据都记录成结构化的 State。Rust 实现则可以利用 enum 加 match 模式匹配来定义状态流转，类型系统带来的安全性在 Agent 这种多分支场景里极其舒服——编译器能帮你拦住"非法状态转移"。

2.4 安全边界与上下文管理：先想好怎么"死"

安全边界是七要素里最不该偷懒的一环。模型拥有决策权之后，它可能会尝试一些超出你预期的操作，比如调用一个危险的工具、删除生产环境的资源、访问没有权限的数据。工程上必须做四层防护：权限最小化——给 Agent 分配独立的 API Key，权限范围仅限任务必需的资源；操作确认——对"高风险动作"插入人工确认步骤，比如删除、支付、发送消息等操作必须二次确认；内容审核——对 Agent 的输入和输出跑敏感词过滤或模型审核；操作审计——全链路记录 Agent 的所有决策和动作，方便事后回溯定位。

这些防护看起来像是"多此一举"，但真实场景中，我见过 Agent 因为工具描述不清晰而调错了接口，结果给几百个用户群发了测试消息。别等到事故发生了才补防护。

上下文管理的关键是"预算"意识。每次请求的 token 数直接决定延迟和成本，而 Agent 天然会不断累积上下文。建议在设计系统时就做四级递进式压缩：裁剪——丢弃最老的低价值消息；摘要——用模型把中间讨论压缩成一段话；结构化提取——只保留关键实体和决策记录；终极方案——重置上下文，但要保留全局目标和已完成事项清单，保证 Agent 哪怕"失忆"了也知道自己为什么在这里、已经做到了哪里。

3. 实现的分岔口：七个决策点逐个拆解

当你把七要素想清楚，下一步就是真正写代码。但写代码的过程中，你会在七个地方反复纠结——我把它们定义为"七个决策点"。每个决策点都没有绝对正确的答案，只有适不适合你当前场景的答案。下面我把每个决策点的取舍依据、实战参数和踩坑经验一起讲清楚。

3.1 决策点一：用现成框架还是自研核心？

这是你动手写的第一个决策。现成框架（LangChain、LangGraph、AutoGen、CrewAI）的好处是生态成熟，工具链齐全，上手快。坏处是抽象层级高，一旦底层逻辑出问题，排查起来极其痛苦。我见过有人用 LangChain 写了一个看似复杂的 Agent，结果一次模型输出格式变化就导致整个链路崩溃，最后查了两天才发现是某个中间解析环节的容错没做好。

我的建议是：MVP 阶段用现成框架验证产品逻辑，正式系统里把核心循环换成自己的实现。LangGraph 的好处在于把 Agent 定义成了显式的图结构，节点和边一目了然，方便调试和状态控制。而 FastAPI 作为接入层，天然适合异步并发和流式输出。如果你评估下来决定自研，那么核心 Agent 循环——"规划→调用→观察→再规划"这个循环——用 Python 大概 200 行左右就能写得相对完整了。Rust 的话代码量会多一些，但运行时性能、并发能力和内存安全带来的收益非常显著。

我自己常用的做法是：用 LangGraph 做流程编排，但重写里面的模型调用和工具执行层，替换成自己的异步实现。这样既保留了图结构的清晰性，又摆脱了框架层可能带来的黑盒风险。

3.2 决策点二：Agent 循环如何设计？LLM 放在循环内还是循环外？

这个问题是我见过最多人踩坑的地方。Agent 的核心是循环：模型输出一个行动，程序执行它，把结果返回给模型，模型再看下一步。但循环里有个致命问题：模型输出永远是概率性的，你没法保证它每一步都完美。

所以实践里的关键设计是：把 LLM 调用封装成一个"可重试、可降级"的函数，并且把循环设计成有界循环——最多迭代 N 次（通常 N 取 5~10），超过次数直接终止并转人工。不要在循环里让模型无限自我发挥，那既烧钱又不稳定。每轮循环都记录行动、观察、思考摘要，方便后续做回溯和调试。

循环内外的另一个取舍是：要不要让模型在每一步都"重新读一遍"所有历史？当然不要。实践中我一般把历史分成"核心上下文"（始终保留）和"过程记录"（可裁剪或摘要）。核心上下文包含用户目标、已确认的关键事实、最近的观察结果。只有把这些维持精炼，模型的注意力才能集中在当前决策上。

我用一个实际数字来说明：同样是 10 步完成的任务，如果每一步都塞全部历史，总 token 消耗可能达到 8 万~12 万；如果维护精炼上下文，可以压到 2 万~3 万，成本差距接近四倍。

3.3 决策点三：并发怎么扛？同步阻塞还是异步事件驱动？

并发是 Agent 从 Demo 走向生产最痛苦的一道坎。Agent 的天然特性是长耗时（一次任务可能跑几十秒甚至几分钟）、多轮交互、频繁调用外部 API，这对传统"一个请求一个线程"的模型是致命的。我见过有人用 Django 同步写 Agent 接口，一旦用户同时发起 20 个任务，线程池直接被打满，后面所有请求排队到超时。

高并发 Agent 的核心思路是：不要让每个请求霸占一个工作线程，要把任务变成事件驱动的异步流水线。FastAPI 的 async/await 配合 httpx.AsyncClient，或者 Rust 的 Tokio 运行时，都是很好的载体。工具调用的网络请求是天然的 IO 等待点，在这些点上让出 CPU，并发能力会大幅提升（在我测试过的一个部署里，异步重构后吞吐从 5 请求/秒涨到了 80 请求/秒，QPS 提升 16 倍）。

并发往往还带出两个伴生问题：限流和队列。不是所有任务都需要即时处理，你可以把高耗时任务丢进消息队列（如 Redis Stream、RabbitMQ、或 AWS SQS），由 Worker 异步消费。交互型任务才走 WebSocket 实时通道，后台型任务走队列+Webhook 通知。这种分层设计，才是 Agent 并发问题的最终解。

3.4 决策点四：任务编排——单 Agent 还是多 Agent 协作？

很多热词文章都在鼓吹"Multi-Agent 是未来"，但我不建议一上来就搞多 Agent。多 Agent 的通信成本、上下文重复消耗、协调复杂度都会翻倍，稍有不慎整场对话就变成两个模型在互相"鸡同鸭讲"。

我的建议是：先用单 Agent + 多工具解决 80% 的问题。只有当你发现单一模型同时处理太多职责导致提示词爆炸、工具选择混乱时，才考虑拆分。拆分的粒度按"职责"拆，而不是按"流程步骤"拆。比如"数据分析 Agent"和"报告生成 Agent"是合理的分工；而"第一步 Agent"和"第二步 Agent"这种把流程硬切开的做法，会导致信息传递的大量损耗。

多 Agent 协作的工程实现上，最稳妥的模式是 supervisor（主管）模式：一个主管 Agent 负责任务分解、进度追踪、最终汇总；多个子 Agent 各司其职。这种模式在 LangGraph 里实现起来比较自然，每个子 Agent 是一个子图，主管负责调度它们。实测中，这种架构比"平级广播"式的多 Agent 协作稳定得多。

3.5 决策点五：模型推理时延怎么优化？缓存和路由策略

Agent 的响应速度决定了用户会不会留着用你的产品。这里有几个优化手段，按性价比从高到低排序。

第一招是语义缓存。把用户请求做 embedding，在向量库里找语义相近的历史请求，如果找到了且任务类型匹配，直接复用之前的答案。实测下来，对于重复度较高的客服场景，语义缓存可以把成本压掉四成以上、延迟压掉七成以上。需要注意的是加"置信度阈值"——两句话"语义相近"但实际意图不同，盲目复用缓存会闹笑话。

第二招是模型路由。简单意图直接走小模型（如 Qwen-Turbo）、复杂推理走大模型（如 Claude/GPT-4 级别）。路由可以用一个轻量模型分类器，也可以用规则判断（比如按关键字或用户输入的复杂度估算）。在很多实际项目里，这个策略能把平均成本降到原来的四分之一。

第三招是流式输出。Agent 的中间过程应该以 SSE（Server-Sent Events）流式推给前端，让用户先看到"正在思考、正在执行"，而不是干等一个完整的 JSON 回来。流式不只是体验优化，它还能尽早暴露 Agent 是不是卡住了——你可以设一个流式心跳超时，超时自动重启或降级。

3.6 决策点六：工具调用的失败处理——重试、降级还是换路？

工具调用是 Agent 最容易出错的地方。网络超时、参数错误、返回格式异常、权限不够，全都会让一次工具调用失败。如果你不处理失败，Agent 就会卡在原地或者干脆"胡言乱语"。

工程实践里我一般给每个工具调用设计三层容错：第一层，可重试错误（网络超时、5xx）——自动重试 2~3 次，指数退避，避免雪崩；第二层，参数类错误（4xx）——把错误信息反馈给模型，让模型自己修正参数后再试一次。这里有个技巧：把工具的"输入校验规则"和"常见错误示例"直接写进工具描述里，模型生成合法参数的成功率会大幅提升；第三层，不可恢复错误——记录错误，把问题抛回给用户或者标记为"需要人工介入"。

注意一个细节：工具返回的内容可能非常长。把完整返回直接塞回上下文，token 消耗会很夸张。我一般让工具的返回经过一次摘要——只保留关键字段、状态码、核心结果，大段文本做截断或向量化存储，需要时再按 ID 检索。这种"按需取用"的思路，能让长工具结果的成本从指数级降到线性级。

3.7 决策点七：可观测性与调试——Agent 黑盒怎么打开？

Agent 的可观测性比传统后端系统更难做。传统接口的输入输出是明确的、链路的每一步都是确定性的；Agent 的每一步是模型根据上下文"概率性"选择的，排查问题时你不仅要知道"发生了什么"，还要知道模型当时为什么这么选。

所以从第一天就要做好全链路日志。我在实际项目中设计了"思维链追踪"数据结构，每次模型决策都记录：输入上下文摘要、模型原始输出（包括思考过程）、选择的工具与参数、工具返回结果、下一步决策。这个数据在调试时极其有用。线上出现一次离谱行为时，回放这条链路，往往一眼就能看出是哪一步的上下文误导了模型。

另外一个实用的调试技巧是LangSmith 或自研追踪面板，你可以为每一次 Agent 运行生成一个 trace_id，把所有日志、token 用量、耗时都挂在这个 ID 下。用户一报问题，拿 trace_id 一查就知道问题出在哪个节点。没有这套机制，Agent 出问题你只能靠猜，效率极低。

4. 实操过程：用 FastAPI + LangGraph 从零搭一个可扛并发的 Agent

前面讲了理论和决策点，这一节我来完整走一遍实操：用 FastAPI + LangGraph + LangChain 搭建一个能流式输出、异步并发、带状态管理的 Agent 服务。这套方案是我实际项目里验证过的，可以直接当作参考骨架。

4.1 项目结构和依赖选型

首先明确技术栈：FastAPI 做 Web 接入层，LangGraph 做 Agent 状态图编排，LangChain 做工具封装和模型接口统一，Redis 做短期状态存储和消息队列，PostgreSQL 存长期数据和审计日志。这套组合的好处是每一层都有成熟生态，替换成本低。

依赖安装：

pip install fastapi uvicorn langgraph langchain langchain-openai redis sqlalchemy pgvector pydantic-settings

如果你的模型走 OpenAI 兼容协议（比如 Qwen、DeepSeek、Moonshot 等），直接用 langchain-openai 的 ChatOpenAI 配置 base_url 即可。

项目结构建议按模块拆分：

agent_service/ ├── main.py # FastAPI 入口 ├── agent/ │ ├── graph.py # LangGraph 状态图定义 │ ├── state.py # Agent 状态数据结构 │ ├── nodes.py # 各节点的执行逻辑 │ └── tools.py # 工具注册与封装 ├── llm/ │ ├── router.py # 模型路由 │ └── cache.py # 语义缓存 ├── api/ │ └── routes.py # HTTP/WebSocket 接口 └── store/ ├── redis_store.py # 短期状态存储 └── pg_store.py # 长期数据存储

这种结构把"Agent 核心逻辑"和"接入层"分开，后期如果要加多 Agent 或者迁移成 Rust 实现，能保留清晰的边界。

4.2 定义 Agent 状态机

LangGraph 的核心思想是"图"——你有若干个节点（Node），节点间通过边（Edge）连接，每条边可以带条件路由。我先定义状态结构：

# agent/state.py from typing import Any, Optional from pydantic import BaseModel, Field class ToolCallRecord(BaseModel): tool_name: str arguments: dict result_summary: str success: bool error: Optional[str] = None class AgentState(BaseModel): user_id: str = Field(..., description="用户标识") task_id: str = Field(..., description="任务标识") goal: str = Field(..., description="用户原始目标") step: int = Field(0, description="当前循环步数") history: list = Field(default_factory=list, description="对话历史") tool_records: list[ToolCallRecord] = Field(default_factory=list, description="工具调用记录") final_answer: Optional[str] = None status: str = Field(default="running", description="运行状态")

这里最关键的是 goal 字段——不管循环怎么变，用户最初的目标必须始终保留在状态里，给模型每一轮"锚定"用的。否则模型一旦在细节里绕晕了，就会忘掉用户想要什么。

下一步定义节点函数。LangGraph 的节点函数签名统一是(state) -> partial_state，你返回一个字典，会自动合并进全局状态。

# agent/nodes.py from .state import AgentState, ToolCallRecord from llm.router import get_model SYSTEM_PROMPT = """你是一个任务规划型 AI Agent。 用户的目标是：{goal} 当前进度：{step}/{max_steps} 步，以下是历史记录： {history_compact} 可使用的工具：{tool_schemas} 请严格按 JSON 格式回复，不要输出多余文字：如果是调用工具，回复 {{"type": "tool_call", "tool": "...", "args": {{...}}}}；如果是最终回答，回复 {{"type": "final", "answer": "..."}}。""" async def agent_reasoning_node(state: AgentState): """Agent 推理节点：决定下一步调用工具还是给出最终答案""" model = get_model(state) prompt = SYSTEM_PROMPT.format( goal=state.goal, step=state.step, max_steps=10, history_compact=summarize_history(state.history), tool_schemas=json.dumps(register_tool_schemas(), ensure_ascii=False)[:4000] ) resp = await model.ainvoke(prompt) parsed = parse_model_output(resp.content) if parsed["type"] == "tool_call": return {"step": state.step + 1, "history": state.history + [{"role": "assistant", "content": resp.content}]} return {"final_answer": parsed["answer"], "status": "completed"}

你可能会问：为什么不在这个节点里直接执行工具？因为 LangGraph 的图结构讲究"一个节点一个职责"。推理节点只管"决定做什么"，工具执行节点负责"去做"。分开的好处是可观测、可回放、可针对性地做重试。执行代码如下：

async def execute_tool_node(state: AgentState): """工具执行节点：执行推理节点输出的工具调用""" last_msg = state.history[-1] parsed = parse_model_output(last_msg["content"]) tool_name = parsed["tool"] args = parsed["args"] tool_func = get_tool(tool_name) try: result = await tool_func(**args) summary = summarize_result(result) record = ToolCallRecord(tool_name=tool_name, arguments=args, result_summary=summary, success=True) except Exception as e: error_msg = str(e)[:500] record = ToolCallRecord(tool_name=tool_name, arguments=args, result_summary=error_msg, success=False, error=error_msg) return { "tool_records": state.tool_records + [record], "history": state.history + [{"role": "tool", "tool_name": tool_name, "content": record.result_summary}] }

4.3 条件边：终止与重试的判定

LangGraph 里最有价值的机制是条件边（conditional edge）。我在图里定义了两个条件判断：一个是"是否该终止循环"——当状态里出现 final_answer，或者 step 超过 10，就走 END；另一个是"工具执行是否成功"——失败时返回推理节点，让它重新规划。

# agent/graph.py from langgraph.graph import StateGraph, END from agent.nodes import agent_reasoning_node, execute_tool_node def create_agent_graph(): graph = StateGraph(AgentState) graph.add_node("reasoning", agent_reasoning_node) graph.add_node("execute_tool", execute_tool_node) graph.add_edge("reasoning", "execute_tool") def route_after_execute(state): # 如果工具调用失败，回推理节点重试；否则继续推理 if state.tool_records and not state.tool_records[-1].success: return "reasoning" if state.final_answer or state.step >= 10: return END return "reasoning" graph.add_conditional_edges("execute_tool", route_after_execute) graph.set_entry_point("reasoning") return graph.compile()

这里有个细节值得注意：当工具失败回退到推理节点时，状态里已经积累了"工具返回的错误摘要"，模型看到错误后会有两个选择——修正参数重新调用，或者改走备用方案。你在提示词里应该明确指示："如果工具返回错误，优先检查参数，如果仍失败请告知用户并提供替代方案。"这能防止模型陷入"同一个工具无限重试"的死循环。

还有一个实践细节：设定最大步数 10，不是拍脑袋的数字。我统计过很多真实任务，大约 95% 的任务在 6 步以内能完成。超过 8 步时，上下文已经消耗很多，模型开始出现"绕圈"倾向，此时及时止损、转人工或让用户重新描述需求，反而是更好的体验。

4.4 FastAPI 异步接入：WebSocket 流式交互

Agent 是长耗时任务，HTTP 轮询让人着急，WebSocket 是更合适的交互通道。我写一个简单的 WebSocket 端点，通过流式中间结果推送，让前端实时看到 Agent 的"思考过程"。

# api/routes.py from fastapi import APIRouter, WebSocket from agent.graph import create_agent_graph from store.redis_store import save_task_state router = APIRouter() @router.websocket("/ws/agent") async def agent_endpoint(websocket: WebSocket): await websocket.accept() graph = create_agent_graph() async def stream_callback(event: dict): if event.get("type") == "node_start": await websocket.send_json({"event": "thinking", "node": event["node"]}) elif event.get("type") == "tool_call": await websocket.send_json({"event": "tool_call", "tool": event["tool"]}) first_msg = await websocket.receive_json() state = { "user_id": first_msg["user_id"], "task_id": first_msg["task_id"], "goal": first_msg["goal"], "history": [], } try: async for chunk in graph.astream(state, config={"callbacks": [MyCallbacks()]}): if chunk.get("final_answer"): await websocket.send_json({"event": "done", "answer": chunk["final_answer"]}) await save_task_state(state, status="completed") except Exception as e: await websocket.send_json({"event": "error", "message": str(e)}) finally: await websocket.close()

LangGraph 的astream天然支持异步流式遍历节点执行过程，你可以在回调里把每个节点的执行状态推给前端。这比"结果一次性返回"的体验好太多——用户能看到 Agent 正在"思考"、正在"调用工具"，而不是面对一个转圈圈的白屏。

4.5 并发扛压的底座：Redis 状态存储与队列削峰

如果只是单机部署，WebSocket 连接一多，内存里直接维护状态会被撑爆。我用 Redis 存储任务状态，每个 task_id 对应一份状态 JSON。这样即使进程重启，任务还能从 Redis 恢复。Redis 的过期时间设置为 30 分钟，超时任务自动清理。

# store/redis_store.py import json, aioredis redis = await aioredis.from_url("redis://localhost:6379") async def save_task_state(state: dict, ttl: int = 1800): key = f"task:{state['task_id']}" await redis.set(key, json.dumps(state.dict()), ex=ttl) async def get_task_state(task_id: str): raw = await redis.get(f"task:{task_id}") return json.loads(raw) if raw else None

但光有状态存储还不够，高并发下的瞬时尖峰怎么办？答案是队列削峰。实时交互型任务走 WebSocket 直接执行；后台非实时任务（比如定时生成日报、批量数据分析）走队列，由 Worker 异步消费。

# 队列生产者 async def enqueue_task(user_id: str, goal: str): task_id = uuid.uuid4().hex payload = {"task_id": task_id, "user_id": user_id, "goal": goal} await redis.rpush("agent_queue", json.dumps(payload)) return task_id

Worker 侧用独立的进程跑，异步从队列取任务并执行同一个 LangGraph 实例。这种"实时通道 + 异步队列"的双轨架构，基本能覆盖九成以上 Agent 产品的并发需求。实测下，单个 FastAPI 实例配合 Redis 队列和 4 个 Worker 进程，能稳定撑住每秒 100+ 的请求创建，任务吞吐取决于模型 API 本身的并发限制。

5. 常见问题与排查技巧实录

这一节我从真实踩坑经历出发，整理几个高频问题和排查思路。这些问题在官方文档里基本找不到答案，但实际开发中几乎必然遇到。

5.1 模型输出 JSON 格式不稳定怎么办？

Agent 的推理节点要求模型输出 JSON，但概率模型不可能保证 100% 格式正确。我经历过最离谱的情况：模型在 JSON 后面附带了一句"以上是我的分析"，导致 json.loads 直接抛异常。

解法分三层：第一层，提示词里加"只输出 JSON，不要任何解释"，并且用 few-shot 给出范例；第二层，写一个健壮的解析函数——先尝试整体 json.loads，失败则用正则提取最外层大括号内的内容，再尝试解析；第三层，终极兜底——解析失败时返回一个"格式化错误"信号给模型，让它重新输出。

def parse_model_output(content: str) -> dict: import json, re content = content.strip() try: return json.loads(content) except: pass # 尝试提取 JSON 大括号块 match = re.search(r'\{.*\}', content, re.DOTALL) if match: try: return json.loads(match.group()) except: pass # 兜底：返回一个特殊的错误指令 return {"type": "format_error", "message": "模型输出无法解析"}

不要小看这三层防御。线上一次格式化错误可能就让整个任务卡死，有了兜底逻辑，最多是多消耗一次模型调用成本，但绝不至于直接崩溃。

5.2 Agent 陷入循环出不来怎么处理？

Agent 绕圈是频率最高的问题之一。现象是：模型反复调用同一个工具，或者反复进行"推理→下结论→发现不对→再推理"，就是不给最终答案。原因通常是上下文中的关键约束信息丢失，或者工具结果不足以推进决策。

解决思路分三个方向：第一，在提示词里硬性规定"最多尝试 2 次同类工具，如果结果不理想，请调整策略或给出阶段性结论"；第二，在代码层做"重复检测"——如果推理节点连续 3 次输出的工具名一模一样且参数高度相似，直接判定循环，强制跳出并回复用户"该方案无法达成，是否需要换一种方式"；第三，兜底步数限制——也就是前面说的 max_steps=10，到点即停。

我这里给一个具体的重复检测实现：

def detect_loop(tool_records: list[ToolCallRecord]) -> bool: if len(tool_records) < 3: return False last3 = tool_records[-3:] if all(r.success for r in last3): return False names = [r.tool_name for r in last3] # 连续三次同一个工具 if len(set(names)) == 1: return True return False

这个函数放到条件边逻辑里，每当工具执行完，先检查是否触发循环检测。触发了就不再走reasoning，直接走一个"解释+人工求助"的终态节点。

5.3 上下文把模型"冲昏"了怎么压缩？

模型"犯糊涂"经常是因为上下文太满，关键信息被淹没。这时候需要做的是显式压缩，而不是简单截断。我在项目里用的是一套"三级压缩"方案。

第一级是历史消息裁剪。把所有历史里的工具返回做摘要，只保留"工具名 + 关键结果 + 状态码"，把工具原始大段输出替换掉；第二级是对话轮次摘要。如果历史超过了 10 轮，把前面的对话合并成一段 200 字以内的摘要；第三级是把"用户目标 + 已完成步骤"为核心上下文，始终放在最前面。

代码示意如下：

def compact_context(goal: str, history: list, max_rounds: int = 10): if len(history) <= max_rounds: return history head = history[:2] # 用户最初的请求务必保留 tail = history[-max_rounds:] middle = summarize(history[2:-max_rounds]) return head + [{"role": "summary", "content": middle}] + tail

你可能会担心摘要过程又要调用模型，增加延迟和成本。我的做法是：非关键步骤的摘要用规则生成（截取 + 模板拼装），只有关键转折点才调用小模型做语义摘要。实测下来，这种分级处理能把上下文从几万 token 压到几千，同时保持决策质量不掉。

5.4 工具返回内容太长导致 token 爆炸怎么办？

这个问题在第 3.6 节提过，这里展开一个真实案例。我做过一个网页内容分析 Agent，工具返回的是一整个网页的爬取文本，动辄几万字。直接塞回上下文，一轮任务轻松烧掉十几万 token。

我的解法是"先存后取"：工具执行时把完整文本存入对象存储（或 Redis），只把"文本长度 + 前 500 字预览 + 存储 ID"返回给模型。模型通过另一个工具retrieve_full_content(content_id)按需获取全文。多数情况下，模型只需要预览的内容就能做决策，全文读取是个低频动作。这个设计把平均单任务 token 消耗降到了原来的三分之一。

另一个补充策略是给每个工具加return_summary参数，让工具在执行时就返回一个结构化摘要而非原始全文。这需要在工具开发阶段就约定好返回格式，属于工具设计的一部分，别等上线了再补。

5.5 WebSocket 连接断开了怎么办？

Agent 任务可能跑 30 秒甚至更久，这期间用户可能刷新了页面、网络抖动，WebSocket 就断了。如果任务状态只存在内存里，连接一断任务就白跑了。

解法是前面设计的 Redis 状态存储：每个任务都有 task_id，WebSocket 断了没关系，任务在后台继续执行完，结果写入 Redis。用户重连时前端带上 task_id，后端从 Redis 拉状态，把已完成的结果发给用户。

我在前端协议里设计了三个事件：event: "pending"（任务在跑）、event: "done"（结果完成）、event: "need_reconnect"（提示用户重连后携带 task_id 继续查询）。这看起来只是个细节，但对真实用户体验的提升非常明显——用户不会因为刷新页面而丢失一个跑了半天的任务。

5.6 不同模型混用时的"脾气"适配

我在模型路由部分提到小模型和大模型混用，但不同模型的输出习惯差异很大。有些模型喜欢在 JSON 前加解释，有些模型在 tool_call 里给出的参数键名和工具定义的键名不完全一致，这些都会导致解析层报错。

解决思路是在模型适配层做"归一化"。我把所有模型输出先过一层normalize_model_output(raw, model_name)，内部做四件事：清理输出文本（去解释性文字）；统一工具名称大小写和别名；参数键名映射（比如把query映射到search_query）；缺省参数补全（比如默认 limit=5）。每个模型的适配规则写成一个独立的适配器，新增模型的时候只加适配器不改主逻辑。

用 Rust 实现时，同样的思路可以用 trait 定义ModelAdapter，每个厂商的适配器是一个独立 struct，编译期就保证适配器的行为一致性。这种语言无关的架构思路，让后端从 Python 迁移到 Rust 变得非常顺畅。

6. 从 Demo 到生产：我踩过的坑和最后几点建议

这篇文章能写到这里，是因为我在真实项目里把 Agent 从"技术演示"推进到了"线上可用"，中间踩过的坑远不止前面列的这些。最后再做几段经验分享，算是我个人最想让你记下的几条。

第一，永远别让 Agent 拥有"无限行动权"。无论你的安全边界设计得多周密，都要在循环上设死限——最大步数、最大 token、最大耗时。Agent 的"自主"永远是"有限自主"，失控的 Agent 造成的破坏可能远超你的想象。我现在的项目里，所有高风险操作默认走人工确认，即使这意味着用户体验会打折扣。这是用一次事故换来的教训。

第二，调试 Agent 最快的方式不是打日志，而是"回放"。给每一次运行加 trace_id，记录每一步的完整输入输出。模型出错的时候，你要看的是它"当时看到了什么"，而不是"它输出了什么"。LangSmith 这些工具能干这件事，自研也不复杂，关键在于从第一天就做，不要等上线后再补。没有全链路追踪的 Agent 系统，排查问题约等于大海捞针。

第三，架构上不要追求"一步到位"。刚起步的 Agent 项目，用单 Agent + 多工具 + 最大步数限制，完全够用。Multi-Agent 是把双刃剑，只在单 Agent 已经"不堪重负"时才考虑拆分。同理，不要一开始就上 Rust 重写核心——先用 Python 验证逻辑，等确实遇到性能瓶颈，再考虑把核心循环迁移到 Rust。我用 Python + FastAPI 搭的第一版就扛住了十万级用户量，性能瓶颈从来不在语言，而在模型调用和并发模型的设计上。

第四，成本控制要从第一天做起。Agent 是多轮调用，成本是普通 ChatGPT 提问的 5~10 倍。上线前必须设置"单任务成本上限"（比如 2 元/任务），超过直接熔断转人工。没有成本预算的 Agent 系统，火起来的第一天就是你亏损的开始。我在代码里给每次运行打上 token 统计，成本异常时自动告警，这个机制帮我避免过好几次"爆单"事故。

第五，也是最重要的一点——始终记得 Agent 的本质是"辅助人"，而不是"替代人"。真正好用的 Agent 产品，在面对不确定、高风险、高价值决策时，会坦诚地告诉用户"我拿不准，请确认"。这种"敢说不知道"的设计，反而会让用户更信任它。我在提示词里始终强调一句："如果你不确定或者发现方案不可行，请直接说明，不要硬编造一个答案。"模型遵循这个原则后，用户投诉反而减少了。

Agent 的工程实现是一个持续迭代的过程。七要素让你想清楚"为什么做"，七个决策点让你知道"怎么做"。但真正让 Agent 从"能用"变"好用"的，永远是你在真实场景中的一次次调试、一次次修复、一次次优化。希望这篇文章能帮你少踩几个坑，把更多时间花在真正有价值的事情上。