Agent 这个词在过去两年被反复咀嚼但真正落到工程实现上很多人卡在同一个地方知道单轮调用能跑通一旦要处理多步骤任务、多角色分工整个系统就开始失控。我见过太多项目在 Demo 阶段惊艳上线后因为状态管理混乱、工具调用死循环、Agent 之间互相甩锅而崩盘。这篇内容不聊概念炒作只拆解 Agent 从单轮调用到多 Agent 协作的完整实现路径把每个阶段的核心机制、选型逻辑和踩坑经验摊开讲。无论你是刚接触 LangChain 的新手还是正在设计企业级 Agent 架构的工程师都能从中找到可复用的设计思路和避坑指南。1. 单轮调用Agent 的最小可用单元1.1 单轮调用的本质是带工具的 LLM 调用很多人把 Agent 想得太神秘其实最基础的 Agent 就是一个带工具调用能力的 LLM 请求。用户输入问题模型判断是否需要调用外部工具如果需要就生成工具调用参数执行工具后把结果塞回上下文再让模型生成最终回答。这个循环只跑一轮所以叫单轮调用。用 LangChain 实现一个最简 Agent 大概长这样from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气 # 实际项目中这里调用天气 API return f{city}今天晴气温 25 度 llm ChatOpenAI(modelgpt-4o, temperature0) prompt ChatPromptTemplate.from_messages([ (system, 你是一个助手可以调用工具回答问题。), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) agent create_openai_tools_agent(llm, [get_weather], prompt) executor AgentExecutor(agentagent, tools[get_weather], verboseTrue) result executor.invoke({input: 北京今天天气怎么样})这段代码的核心逻辑是模型看到用户问题后判断需要调用get_weather生成{city: 北京}这样的参数工具执行返回结果模型再基于结果生成自然语言回答。整个过程只经历一次思考-调用-回答的循环。1.2 为什么单轮调用撑不起真实业务单轮调用的局限非常明显。它只能处理一问一答一工具的简单场景一旦任务需要多步推理、多个工具串联、或者需要根据中间结果动态调整策略单轮调用就无能为力了。举个实际例子用户问帮我分析一下特斯拉最近三个月的股价走势并对比比亚迪的表现。这个问题需要第一步调用股票数据接口获取特斯拉数据第二步调用接口获取比亚迪数据第三步可能还需要调用计算工具做对比分析第四步生成图表或文字报告。单轮调用根本没法完成这种链式任务。更关键的是单轮调用没有记忆。每次请求都是独立的模型不知道上一轮发生了什么。这在需要多轮交互的场景下是致命的。注意很多教程把单轮调用包装成Agent 入门但实际上它只是一个带工具的函数调用。真正的 Agent 能力体现在多轮循环和状态管理上。1.3 从单轮到多轮ReAct 模式的引入解决单轮局限的第一个方案是 ReActReasoning Acting模式。它的核心思想是让模型在每一轮都输出思考和行动然后根据行动结果决定下一步。ReAct 的循环逻辑是模型接收用户问题输出 Thought思考下一步该做什么根据 Thought 输出 Action选择工具和参数执行 Action得到 Observation工具返回结果把 Observation 塞回上下文模型继续输出下一个 Thought重复直到模型认为可以给出 Final Answer这个循环让 Agent 具备了多步推理能力。用 LangChain 的create_react_agent可以快速实现from langchain.agents import create_react_agent, AgentExecutor from langchain import hub prompt hub.pull(hwchase17/react) agent create_react_agent(llm, tools, prompt) executor AgentExecutor(agentagent, toolstools, max_iterations10, verboseTrue)max_iterations这个参数非常关键。我踩过的坑是如果不设上限模型可能陷入死循环反复调用同一个工具却得不到有效结果。设成 10 到 15 是比较稳妥的范围具体取决于任务复杂度。ReAct 解决了多步推理的问题但它仍然是一个单体 Agent。所有工具、所有逻辑都塞在一个 Agent 里当任务复杂度继续上升时这个单体 Agent 会变得臃肿、难以维护、提示词越来越长、工具选择准确率下降。2. 多 Agent 协作分工带来的能力跃迁2.1 为什么要拆成多个 Agent单体 Agent 的问题在任务复杂度超过某个阈值后会集中爆发。我总结下来主要是三个第一提示词膨胀。当你有 20 个工具时系统提示词里要描述每个工具的功能、参数、使用场景提示词长度可能超过 3000 token。模型在这么长的上下文里选择正确工具的准确率会明显下降。第二职责混乱。一个 Agent 既要会查数据库又要会写代码还要会做数据分析它的人格是分裂的。模型在不同任务间切换时容易产生混淆。第三难以优化。当整个系统只有一个 Agent 时你没法针对某个环节单独调优。改了提示词可能修好了 A 场景却弄坏了 B 场景。多 Agent 协作的思路就是把这一个全能选手拆成多个专才每个 Agent 只负责一个明确的子任务通过编排层协调它们的工作。2.2 多 Agent 的三种典型拓扑结构在实际项目中多 Agent 的协作方式主要有三种拓扑第一种是流水线式Pipeline。Agent A 的输出是 Agent B 的输入B 的输出给 C像工厂流水线一样。这种结构适合步骤明确、顺序固定的任务比如数据采集 Agent → 数据清洗 Agent → 数据分析 Agent → 报告生成 Agent。第二种是主管式Supervisor。有一个主管 Agent 负责理解用户需求、拆解任务、分发给下面的 worker Agent最后汇总结果。这种结构灵活度高适合任务类型多样的场景。第三种是群聊式Group Chat。多个 Agent 在一个共享的对话空间里讨论每个 Agent 可以发言、可以调用工具通过多轮讨论达成共识。这种结构适合需要多视角分析的复杂决策场景。拓扑结构适用场景优点缺点流水线式步骤固定的流程化任务逻辑清晰、易调试灵活性差、无法处理分支主管式任务类型多样的通用场景灵活、易扩展主管 Agent 容易成为瓶颈群聊式需要多视角的决策场景视角全面、容错性好成本高、容易发散2.3 LangGraph 为什么成为多 Agent 编排的主流选择LangChain 早期的 Agent 实现是黑盒式的你给一个 AgentExecutor它内部怎么循环、怎么决策你很难干预。这在单 Agent 场景下还能接受但多 Agent 协作需要精确控制每个 Agent 的输入输出、状态流转、条件分支黑盒就不够用了。LangGraph 的出现解决了这个问题。它把 Agent 的工作流建模成一张图Graph节点Node是具体的处理逻辑可以是一个 Agent、一个工具调用、一段代码边Edge定义了节点之间的流转关系。你可以精确控制每个节点的输入是什么、输出是什么什么条件下走哪条边状态如何在节点之间传递和累积什么时候终止循环这种白盒式的编排让多 Agent 系统的可控性大幅提升。用 LangGraph 定义一个简单的主管式多 Agent 系统from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] next_agent: str def supervisor_node(state: AgentState): # 主管 Agent 决定下一步交给谁 # 实际项目中这里调用 LLM 做路由决策 last_message state[messages][-1] if 数据 in last_message.content: return {next_agent: data_agent} elif 报告 in last_message.content: return {next_agent: report_agent} return {next_agent: finish} def data_agent_node(state: AgentState): # 数据处理 Agent 的逻辑 return {messages: [(assistant, 数据已处理完成)]} def report_agent_node(state: AgentState): # 报告生成 Agent 的逻辑 return {messages: [(assistant, 报告已生成)]} workflow StateGraph(AgentState) workflow.add_node(supervisor, supervisor_node) workflow.add_node(data_agent, data_agent_node) workflow.add_node(report_agent, report_agent_node) workflow.set_entry_point(supervisor) workflow.add_conditional_edges( supervisor, lambda x: x[next_agent], { data_agent: data_agent, report_agent: report_agent, finish: END } ) workflow.add_edge(data_agent, supervisor) workflow.add_edge(report_agent, supervisor) app workflow.compile()这段代码定义了一个主管式的工作流主管节点根据消息内容决定路由到哪个 workerworker 处理完后回到主管节点主管再决定下一步直到任务完成。2.4 状态管理多 Agent 协作的命脉多 Agent 系统最容易出问题的地方就是状态管理。每个 Agent 需要知道当前任务进展到哪了之前有哪些输出下一步该做什么这些信息都通过状态来传递。LangGraph 的状态设计有几个关键点第一状态要可累加。用Annotated[list, operator.add]定义的消息列表每次节点返回新消息时会自动追加而不是覆盖。这个设计让对话历史自然累积。第二状态要精简。不要把整个上下文都塞进状态里只放必要的字段。状态越大每次节点间传递的开销越大调试也越困难。第三状态要可序列化。LangGraph 支持 checkpoint 机制可以把状态持久化到数据库实现断点续跑。这要求状态里的所有字段都是可序列化的不能放函数、连接对象这类东西。我踩过的一个坑是早期把数据库连接对象放进了状态里结果 checkpoint 序列化时直接报错。后来改成在节点内部按需创建连接用完即关问题才解决。3. 工具调用与中间件Agent 的能力边界3.1 工具设计的三个原则工具是 Agent 与外部世界交互的接口。工具设计得好不好直接决定了 Agent 的能力上限。我总结下来有三个原则原则一工具粒度要适中。太细的工具会让 Agent 需要调用很多次才能完成一个任务增加出错概率太粗的工具会让 Agent 失去灵活性。比如查询数据库这个工具就太粗应该拆成查询用户表查询订单表等具体操作。原则二工具描述要精确。工具的名称、描述、参数说明是模型选择工具的唯一依据。描述要写清楚这个工具做什么什么时候用参数是什么格式。我见过很多项目工具描述写得含糊导致模型频繁选错工具。原则三工具要有错误处理。工具执行失败时要返回明确的错误信息而不是抛异常。模型看到错误信息后可以决定重试、换工具或者放弃。如果直接抛异常整个 Agent 流程就中断了。tool def query_order(order_id: str) - str: 根据订单号查询订单详情。 Args: order_id: 订单号格式为 ORD 开头的 12 位字符串例如 ORD20240101001 Returns: 订单的详细信息包括商品、金额、状态等 try: # 实际查询逻辑 result db.query(order_id) if not result: return f未找到订单 {order_id}请确认订单号是否正确 return format_order(result) except Exception as e: return f查询订单时出错{str(e)}请稍后重试3.2 LangChain Agent 中间件机制LangChain 在较新版本中引入了中间件Middleware概念允许你在 Agent 执行的关键节点插入自定义逻辑。这在实际项目中非常有用常见的中间件类型包括日志中间件记录每次工具调用的输入输出用于调试和审计。生产环境必备出问题时能快速定位是哪一步出了错。限流中间件控制工具调用频率防止 Agent 在短时间内大量调用外部 API 导致超限。我一般会设置每分钟最多 30 次工具调用。权限中间件检查当前用户是否有权限调用某个工具。比如删除数据的工具只允许管理员调用。重试中间件工具调用失败时自动重试可以配置重试次数和退避策略。from langchain.agents.middleware import AgentMiddleware class LoggingMiddleware(AgentMiddleware): def before_tool_call(self, tool_name, tool_input): print(f[工具调用] {tool_name} 输入{tool_input}) def after_tool_call(self, tool_name, tool_output): print(f[工具返回] {tool_name} 输出{tool_output[:200]}) class RateLimitMiddleware(AgentMiddleware): def __init__(self, max_calls_per_minute30): self.max_calls max_calls_per_minute self.call_times [] def before_tool_call(self, tool_name, tool_input): now time.time() self.call_times [t for t in self.call_times if now - t 60] if len(self.call_times) self.max_calls: raise Exception(工具调用频率超限请稍后重试) self.call_times.append(now)3.3 工具调用的常见故障与排查工具调用环节是多 Agent 系统最容易出故障的地方。我整理了几个高频问题问题一模型不调用工具直接编造答案。这通常是因为工具描述不够清晰或者系统提示词没有强调必须使用工具获取信息。解决办法是在提示词里明确要求对于事实性问题必须调用工具验证不得凭记忆回答。问题二工具参数格式错误。模型生成的参数不符合工具定义的 schema。解决办法是在工具描述里给出明确的参数示例并在中间件里做参数校验格式不对时返回错误提示让模型重新生成。问题三工具调用死循环。模型反复调用同一个工具每次都得到相同结果却不肯停止。解决办法是设置max_iterations上限同时在中间件里检测重复调用如果连续三次调用同一工具且参数相同强制中断并返回错误。问题四工具返回结果过长。有些工具返回大量数据塞进上下文后导致 token 超限。解决办法是在工具内部做结果截断或摘要只返回关键信息。提示工具调用的调试建议开启 verbose 模式把每一轮的 Thought、Action、Observation 都打印出来。虽然日志会很长但排查问题时非常有用。4. 记忆机制让 Agent 记住该记住的4.1 短期记忆与长期记忆的分工Agent 的记忆分为短期和长期两类。短期记忆是当前会话的上下文包括用户输入、Agent 的思考过程、工具调用结果等。长期记忆是跨会话持久化的信息比如用户偏好、历史任务记录、领域知识等。短期记忆的实现相对简单用消息列表维护即可。关键是要控制长度不能让上下文无限增长。常见的策略有滑动窗口只保留最近 N 轮对话摘要压缩把早期对话压缩成摘要保留关键信息重要性筛选根据信息的重要性决定保留哪些长期记忆的实现就复杂得多。需要解决存什么怎么存怎么取三个问题。4.2 长期记忆的存储与检索策略存什么不是所有信息都值得长期保存。我一般只存三类信息用户明确表达的偏好如我喜欢简洁的回答、重要的任务结果如上次分析的结论是...、领域知识如产品文档、FAQ。怎么存可以用向量数据库存语义信息用关系数据库存结构化信息。LangChain 提供了VectorStoreRetrieverMemory等工具来简化这个过程。怎么取检索时要根据当前上下文选择最相关的记忆。简单的做法是用向量相似度检索复杂的做法是结合时间衰减、重要性评分等多因子排序。from langchain.memory import VectorStoreRetrieverMemory from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings embeddings OpenAIEmbeddings() vectorstore FAISS.from_texts([占位文本], embeddings) retriever vectorstore.as_retriever(search_kwargs{k: 3}) memory VectorStoreRetrieverMemory(retrieverretriever) memory.save_context( {input: 我喜欢用 Python 做数据分析}, {output: 好的我记住了} )4.3 记忆污染与安全防护记忆机制带来一个容易被忽视的风险记忆污染。如果 Agent 把错误信息、恶意指令写入了长期记忆后续所有会话都会受到影响。这在多用户场景下尤其危险。防护措施包括写入校验在写入长期记忆前用另一个 LLM 调用检查内容是否合理、是否包含恶意指令。隔离存储不同用户的记忆分开存储避免交叉污染。定期清理设置记忆的过期时间定期清理陈旧或低价值的记忆。审计日志记录所有记忆的写入和读取操作便于追溯问题。我见过一个案例某客服 Agent 把用户的一句玩笑话以后所有问题都回答不知道写入了长期记忆导致后续所有用户都收到了错误回答。这个坑的教训是长期记忆的写入必须经过严格校验不能什么内容都往里塞。5. 并发与性能Agent 扛住真实流量的关键5.1 Agent 系统的性能瓶颈在哪Agent 系统的性能瓶颈和传统 Web 服务完全不同。传统服务的瓶颈通常在数据库或网络 IO而 Agent 系统的瓶颈主要在三个方面第一LLM 调用延迟。每次 LLM 调用少则几百毫秒多则几秒甚至十几秒。一个多 Agent 任务可能涉及十几次 LLM 调用总延迟很容易超过 30 秒。第二工具调用延迟。外部 API 的响应时间不可控有些接口可能要几秒才返回。第三Token 消耗。多 Agent 系统的上下文通常很长每次调用都要传输大量 token既增加延迟又增加成本。5.2 并发处理的架构设计要让 Agent 系统扛住并发架构上需要做几件事异步化把所有 IO 操作LLM 调用、工具调用、数据库查询都改成异步。Python 里用asyncioLangChain 和 LangGraph 都支持异步接口。import asyncio from langchain_openai import ChatOpenAI async def process_task(task): llm ChatOpenAI(modelgpt-4o) result await llm.ainvoke(task) return result async def main(): tasks [process_task(f任务{i}) for i in range(10)] results await asyncio.gather(*tasks) return results任务队列把 Agent 任务放入队列由 worker 池消费。这样可以控制并发数避免瞬间大量请求打垮下游服务。常用的方案有 Celery、RQ、或者基于 Redis 的简单队列。流式输出对于长任务用流式输出让用户先看到部分结果而不是等全部完成。LangChain 的astream接口支持流式返回。缓存对于重复性高的查询缓存 LLM 响应或工具结果。比如今天天气怎么样这类问题同一城市短时间内可以复用结果。5.3 多 Agent 并行的收益与代价多 Agent 系统有一个天然优势可以并行执行。如果任务 A 和任务 B 没有依赖关系可以让两个 Agent 同时处理总耗时从 AB 变成 max(A, B)。但并行也有代价状态同步复杂多个 Agent 同时修改状态时需要处理冲突。LangGraph 的状态更新是原子的但跨节点的状态合并需要仔细设计。成本上升并行意味着同时消耗更多 token 和 API 调用配额。调试困难并行执行的日志交错在一起排查问题比串行困难得多。我的经验是只在任务确实独立且延迟敏感的场景下用并行其他情况优先用串行简单可靠。场景推荐策略理由多个独立数据源查询并行无依赖并行收益明显有依赖的步骤链串行后一步依赖前一步结果需要多视角分析并行 汇总各视角独立最后合并资源受限环境串行 队列控制并发避免过载6. 从理论到落地搭建多 Agent 系统的实操路径6.1 环境准备与依赖选型搭建多 Agent 系统的第一步是选型。核心依赖包括LLM 接入LangChain 的ChatOpenAI或其他模型接入层编排框架LangGraph推荐或 LangChain 的 AgentExecutor向量数据库FAISS本地、Chroma轻量、Milvus生产级状态存储Redis短期、PostgreSQL长期任务队列Celery 或 RQ安装依赖pip install langchain langchain-openai langgraph langchain-community pip install faiss-cpu redis celery6.2 一个可运行的多 Agent 项目骨架下面是一个最小可运行的多 Agent 项目结构project/ ├── agents/ │ ├── supervisor.py # 主管 Agent │ ├── researcher.py # 研究 Agent │ └── writer.py # 写作 Agent ├── tools/ │ ├── search.py # 搜索工具 │ └── database.py # 数据库工具 ├── graph.py # LangGraph 工作流定义 ├── state.py # 状态定义 └── main.py # 入口状态定义from typing import TypedDict, Annotated import operator class MainState(TypedDict): messages: Annotated[list, operator.add] research_result: str final_report: str next_step: str工作流定义from langgraph.graph import StateGraph, END from state import MainState from agents.supervisor import supervisor_node from agents.researcher import researcher_node from agents.writer import writer_node def route_supervisor(state: MainState): return state[next_step] workflow StateGraph(MainState) workflow.add_node(supervisor, supervisor_node) workflow.add_node(researcher, researcher_node) workflow.add_node(writer, writer_node) workflow.set_entry_point(supervisor) workflow.add_conditional_edges( supervisor, route_supervisor, { research: researcher, write: writer, end: END } ) workflow.add_edge(researcher, supervisor) workflow.add_edge(writer, supervisor) app workflow.compile()6.3 调试与可观测性建设多 Agent 系统的调试比单 Agent 复杂得多。我建议从项目第一天就建设可观测性结构化日志每次 LLM 调用、工具调用都记录结构化日志包含时间戳、Agent 名称、输入、输出、耗时、token 消耗。链路追踪用 trace_id 串联一次完整任务的所有调用方便查看完整执行链路。可视化LangGraph 支持把工作流导出为图可以直观看到节点和边的结构。调试时对照图看日志能快速定位问题。回放机制把每次任务的完整状态序列化保存出问题时可以回放重现。import logging import json from datetime import datetime logger logging.getLogger(agent) def log_llm_call(agent_name, input_text, output_text, duration, tokens): logger.info(json.dumps({ timestamp: datetime.now().isoformat(), agent: agent_name, input: input_text[:500], output: output_text[:500], duration_ms: duration, tokens: tokens }, ensure_asciiFalse))6.4 上线前的检查清单在把多 Agent 系统推向生产前我一般会过一遍这个清单所有工具都有错误处理和超时设置Agent 循环有最大迭代次数限制状态大小有上限不会无限增长有完整的日志和监控有降级方案LLM 不可用时返回兜底回答有成本控制token 消耗上限、调用频率限制有安全防护输入过滤、权限校验、记忆写入校验有压测数据并发能力、P99 延迟7. 多 Agent 协作的边界与未来演进7.1 什么时候不该用多 Agent多 Agent 不是银弹。我见过不少项目为了技术先进而强行上多 Agent结果复杂度飙升、效果反而下降。以下场景不建议用多 Agent任务简单且线性如果任务就是查数据-格式化-返回单 Agent 甚至直接函数调用就够了。延迟极度敏感多 Agent 意味着多次 LLM 调用延迟天然比单 Agent 高。如果业务要求 1 秒内响应多 Agent 基本不可行。成本极度受限多 Agent 的 token 消耗通常是单 Agent 的 3 到 10 倍。如果预算有限要慎重。团队缺乏经验多 Agent 系统的调试和维护门槛明显更高。如果团队没有单 Agent 的实战经验直接上多 Agent 容易失控。7.2 多 Agent 系统的演进方向从当前的技术趋势看多 Agent 系统正在往几个方向演进更智能的路由主管 Agent 的路由决策从简单的规则匹配进化到基于语义理解、历史效果学习的智能路由。动态 Agent 生成根据任务需要动态创建 Agent而不是预先定义好所有 Agent。这需要更强的元认知能力。Agent 间的协商机制多个 Agent 不只是简单的上下游关系而是能互相协商、辩论、达成共识。这在复杂决策场景下很有价值。记忆的共享与隔离多个 Agent 之间如何共享记忆、如何隔离敏感信息是一个持续演进的方向。安全与对齐随着 Agent 能力增强如何确保 Agent 的行为符合预期、不被恶意利用变得越来越重要。记忆防护、工具权限、行为审计都是关键环节。我在实际项目中的体会是多 Agent 系统的价值不在于用了多 Agent这个形式而在于它能否真正解决单 Agent 解决不了的问题。如果一个任务用单 Agent 加几个工具就能做好那就没必要上多 Agent。技术选型要服务于业务目标而不是反过来。最后分享一个实用技巧在搭建多 Agent 系统时先用单 Agent 跑通核心流程确认业务逻辑没问题后再逐步拆分出专门的 Agent。这样每一步都有可验证的基线出问题时也容易定位是拆分引入的还是原本就有的。一上来就设计复杂的多 Agent 拓扑往往会在调试阶段耗费大量时间。