最近在把一个基于 LangChain 的多智能体协同框架从实验 Demo 往可用方向推,过程比预想的曲折:单 Agent 跑得好好的任务,一旦拆成三四个角色,马上会出现上下文污染、任务死循环、工具调用幻觉。调整了路由策略、改了消息传递格式,又把每个子 Agent 的权限边界收紧之后,整个系统才算稳定下来。这篇东西不打算讲 LangChain 的基础 API,也不复制官方文档,核心是分享我设计多智能体框架时真正卡住过的地方,以及最后沉淀下来的可复用结构。适合已经在用 LLM 做应用、但还没踩过多智能体坑的开发者,也适合刚想从单 Agent 往协同架构转的朋友。
1. 单智能体走到头了:复杂任务的三个致命痛点
1.1 上下文窗口很快被无关信息撑爆
单 Agent 处理复杂任务时,最直接的问题是上下文窗口被“填充”。一个客服场景里,用户问订单状态,Agent 先查了订单接口,又顺便调了物流接口,再读了用户的历史投诉记录,中间还会夹着几段模型自己生成的推理文字。这些内容全堆在同一个上下文里,到任务结束时,最早的指令很可能已经被挤到窗口边缘,模型对它的注意力严重衰减,表现为“忘了最初的要求”。
做个实验就能感受到:让同一个 Agent 先做市场调研、再写报告、最后把报告翻译成另一门语言。前两步的中间产物都留在上下文里,第三步翻译时,模型反复丢失报告开头的人名和关键数字,翻译质量肉眼可见地下降。更远的问题是,调试这种单 Agent 流程时根本没法定位是哪一步把上下文污染了,因为所有信息都搅在一起。
1.2 工具调用出现身份混乱
当 Agent 同时持有太多工具时,它会对着错误的目标执行动作。比如一个智能体身上接了邮件发送、数据库查询、外部搜索三个工具,用户说“帮我把上周的销售数据整理成邮件发出去”。模型很可能拿数据库查询工具去搜邮件模板,或者用搜索工具查公司内部数据,然后一本正经地编一个结果。这不是模型不行,而是工具边界太宽,模型在单次决策中需要同时考虑“调用哪个工具”和“这个工具完成什么使命”,出错的概率会随着工具数量非线性上升。
1.3 角色切换撕裂了指令约束
真实的业务流程往往需要多个角色:一个负责信息收集,一个负责计算,一个负责最终表达。单 Agent 做法是给同一条 Prompt 里塞三套指令,让模型在同一个上下文里反复切换身份。实际跑下来,模型会角色混淆:计算任务还没完成,就开始用总结者的口吻输出结论。我也试过在 Prompt 里用分隔符强制区分角色,模型在一两轮内还能遵守,但只要对话超过 5 轮,角色边界必然漏洞百出。
这些痛点的根源,都是“所有信息、所有职责、所有状态放在同一个大脑里”。多智能体协同的思路,恰恰是把这些职责拆到独立 Agent 中,每个 Agent 只保留自己的工作记忆和专属工具,再通过一层调度系统沟通。
2. 协同框架的三个核心决策:角色、路由与消息格式
2.1 角色划分:宁多勿少?不,宁窄勿宽
设计多智能体框架的第一步,是定义每个 Agent 的角色、职责边界、可用工具。这个看起来很直接,实际上特别容易走极端:要么只拆两个 Agent,职责依旧过宽;要么拆得特别细,导致调度开销比任务本身还大。实践下来,我的原则是“每个 Agent 的职责窄到只需要一种工具和一种输出格式”。
我用这种思路拆了一个销售分析场景:
| Agent 名 | 职责边界 | 可用工具 | 输出格式 |
|---|---|---|---|
| DataAgent | 只处理数据库查询,返回原始表格 | SQLite 查询工具 | 结构化表格 |
| CalcAgent | 只做数值计算和统计,不接触原始库 | Python 计算工具 | 数值数组 |
| ReportAgent | 基于传入的表格和数值,写最终报告 | 无外部工具,只做文本生成 | Markdown 报告 |
这个划分的逻辑很简单:每个 Agent 的失败模式都是独立的,调试时看日志就能知道是查询错了、算错了还是写错了。另一个关键是,每个 Agent 的 Prompt 可以做得非常短,一两句话描述职责和输出格式就够了,不需要把整个业务流程塞进去。
2.2 路由策略:调度者要大胆地说“我不知道”
多智能体框架里,主控 Agent 的角色是“路由器”,它的任务是把用户的请求拆成子任务,分配给对应 Agent,然后汇总结果。这个路由 Agent 不需要很强,它甚至不需要调用任何工具,只需要一个清晰的决策逻辑。
我用的是两层路由:第一层是规则匹配,根据用户请求里的关键词直接指派 Agent。比如请求里出现“查”“Get”“数据”,默认先派 DataAgent。第二层是 LLM 判断,当规则匹配不明确时,让路由 Agent 用一次 LLM 调用判断该走哪个流程。实测下来,纯 LLM 路由在小规模 Agent 场景下非常不稳定,会反复随机派发任务,所以规则先行、LLM 补位是最稳的。
调度者还需要敢于输出“我不知道”。如果用户请求不在任何 Agent 的职责范围内,路由器应该直接拒绝,并给出可操作的指导,而不是硬把一个 Agent 拉出来干它不擅长的事。
2.3 消息格式:只传摘要,不传全量上下文
多智能体之间传递什么决定了整个系统的性能下限。我一开始图省事,直接把上游 Agent 的完整输出塞给下游 Agent,结果下游 Agent 的上下文迅速膨胀,最终输出质量快速劣化。后来改成统一消息结构,每个 Agent 的输入是一个结构化的消息对象,只包含四部分:任务描述、输入数据、必要上下文引用、输出格式要求。
这个结构里,最重要的是“必要上下文引用”。比如 ReportAgent 不需要知道用户原始请求有多长,只需要拿到 DataAgent 处理后的表格和 CalcAgent 算出的平均数和趋势值。这样每个 Agent 的上下文长度是可控的,模型可以全程把注意力集中在自己的任务上。
3. 用 LangChain 搭出可运行的多智能体骨架
3.1 环境准备和基本依赖
我用的是 Python 生态,核心依赖是 langchain、langchain-community,以及模型提供方的 SDK。这里不指定具体模型厂商,因为框架设计本身与模型无关。需要确认的版本要求:LangChain 0.1 以上的版本对create_agent和AgentExecutor的接口做了调整,老代码迁移需要注意。
pip install langchain langchain-community3.2 每个子 Agent 都是一个干净的执行单元
子 Agent 的核心是不跟其他 Agent 直接通信,它只接收一个TaskMessage对象,返回一个AgentResult对象。这样每个 Agent 都可以单独测试、单独替换,甚至可以换成其他类型的模型或工具。
下面是一个子 Agent 的实现骨架,用 LangChain 的create_agent简化封装:
from langchain.agents import create_agent from langchain.tools import tool from langchain_core.messages import SystemMessage from pydantic import BaseModel class TaskMessage(BaseModel): task: str input_data: dict = {} class AgentResult(BaseModel): content: str metadata: dict = {} @tool def query_database(sql: str) -> str: """执行 SQLite 查询并返回结果字符串""" import sqlite3 conn = sqlite3.connect("sales.db") cur = conn.cursor() cur.execute(sql) rows = cur.fetchall() conn.close() return str(rows) system_prompt = SystemMessage( content="你是数据分析员,只负责执行 SQL 查询。" "输入包含任务描述和 input_data,输出必须是查询结果字符串。" ) agent = create_agent( model="gpt-4o-mini", # 换成你的模型 tools=[query_database], system_prompt=system_prompt, verbose=True, )这段代码的关键是system_prompt反复强调“只做一件事”。我把所有子 Agent 的 Prompt 都设计成不超过两句话,目的就是压缩模型的决策空间,降低幻觉率。
3.3 调度器:让路由 Agent 指挥子 Agent 干活
调度器是一个单独的 Agent,它不持有工具,只负责解析用户请求、分发任务、收集结果。实际实现中,我用了一个简单的状态机结构,避免调度器自己陷入角色混乱。
class AgentScheduler: def __init__(self, agent_map: dict, router_prompt: str): self.agent_map = agent_map self.router = create_agent( model="gpt-4o-mini", tools=[], # 调度器不持有外部工具 system_prompt=SystemMessage(content=router_prompt), ) def dispatch(self, user_request: str) -> str: # 第一层:规则匹配 if "查询" in user_request or "数据" in user_request: target = "data_agent" else: # 第二层:LLM 路由 route_decision = self.router.invoke( {"messages": [("user", f"请判断以下请求应分配给哪个Agent:{user_request}")]} ) target = self._parse_decision(route_decision.content) if target not in self.agent_map: return "该请求不在可处理范围内,无法完成。" task_msg = TaskMessage(task=user_request, input_data={}) result = self.agent_map[target].invoke(task_msg.model_dump()) return result["content"]上面这段代码里,_parse_decision从路由输出的文本里提取 Agent 名。我用正则匹配预定义的 Agent 名,如果提取不到就返回默认拒绝。这个环节的容错非常重要,因为 LLM 路由输出经常带过多解释,直接拿它当 key 会炸。
3.4 协同循环:用 iteration 上限防死循环
多智能体框架最恐怖的问题是进入无限循环:Agent A 的结果触发 Agent B,Agent B 的结果又触发 A,两个人互相投喂直到超时或耗尽上下文。我在调度器里加了一个显式的max_rounds约束,每轮派发之后检查是否达到上限,达到就直接结束并返回已完成的部分结果。
max_rounds = 3 rounds = 0 while rounds < max_rounds: result = self.dispatch(sub_task_text) if self._is_final_result(result): return result sub_task_text = self._extract_next_subtask(result, user_request) rounds += 1 return "已达最大协作轮次,输出当前综合结果。"这个限制看起来粗暴,但非常有效。我踩过最真实的坑是:让一个 Agent 写文案,另一个 Agent 做校对,两者互相修改,最后 30 轮都没停,消耗了十几万 token。加了上限后,输出质量反而提高了,因为系统被迫在有限轮次内收敛。
4. 稳定性工程:重试、超时与故障隔离,一个都不能少
4.1 API 调用失败不完全靠“再来一次”
多智能体框架一大问题是 API 调用量陡增,限流和瞬时错误的概率大幅上升。我在子 Agent 返回层加了统一的异常处理 wrapper,所有 Agent 执行都从裸调用改成带重试的调用。
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type @retry( reraise=True, stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=30), retry=retry_if_exception_type((ConnectionError, TimeoutError)), ) def robust_invoke(agent_func, task_msg): return agent_func(task_msg)这里重试策略是指数退避:第一次失败等 2 秒,第二次 4 秒,第三次 8 秒左右,最多 3 次。太激进的固定间隔重试会把限流变成必现故障,我只在 ConnectionError 和 TimeoutError 这类瞬时错误上重试,模型返回格式错误不重试,而走另一条修复通道。
4.2 超时控制:倒排的排队问题
每个子 Agent 都可能卡住,尤其是模型推理过长或者工具调用挂起。我之前吃过一次大亏:报表 Agent 处理一个重查询,模型一直在思考阶段,整个协同链路被拖死了。后来给每个 Agent 调用设置了 30 秒超时,超时后立即返回一个降级结果,而不是让用户无限等待。
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout def invoke_with_timeout(agent_func, task_msg, timeout=30): with ThreadPoolExecutor(max_workers=1) as pool: future = pool.submit(agent_func, task_msg) try: return future.result(timeout=timeout) except FutureTimeout: future.cancel() return AgentResult(content="Agent执行超时,返回空结果。", metadata={"timeout": True})这种降级策略的意义在于,协同系统的整体可用性不能绑定到任何一个子 Agent 的单点表现上。
4.3 故障隔离:错误的 Agent 不能拖死全队
多智能体最可怕的情况不是某个 Agent 失败,而是失败的 Agent 把错误输出传给下游,下游以为那是合法数据,继续加工,最终输出一套漂亮但全错的报告。我在消息传递层加了一个简单的 schema 校验,下游 Agent 收到的 input_data 必须能通过 JSON schema 校验,不合格就直接丢弃这轮结果,并向调度器报错。
from jsonschema import validate def validate_agent_result(result: AgentResult, expected_schema: dict): try: validate(result.model_dump(), expected_schema) except Exception as e: return False, f"Schema校验失败: {e}" return True, ""这个校验把错误拦截在最前端,而不是让它沿着链条假传播。实际测试中,一次数据库查询返回了带 Invalid statement 的报错字符串,结果 ReportAgent 把这个报错当真数据写进了报告。有了 schema 校验后,这种故障能立刻被发现,整个系统可靠性上了一个台阶。
5. 调试与调优:从 Demo 到靠谱协同的九条经验
5.1 日志里必须能看到每一步的完整输入输出
多智能体系统一旦出错,最痛苦的是不知道哪个 Agent 的哪一步出了问题。我的做法是给每个 Agent 的 invoke 方法包一层日志装饰器,记录完整的输入任务、中间工具调用、原始输出和最终的 AgentResult。LangChain 自带的verbose=True会输出部分信息,但对多 Agent 调度,日志流是交织的,必须自己给每条日志加 Agent 名和任务 ID 前缀。
import logging logger = logging.getLogger("multi_agent") logging.basicConfig(level=logging.INFO) def log_task(agent_name: str, task_id: str): def decorator(func): def wrapper(task_msg): logger.info(f"[{agent_name}][{task_id}] 收到任务: {task_msg.task}") result = func(task_msg) logger.info(f"[{agent_name}][{task_id}] 输出: {result.content[:200]}") return result return wrapper return decorator这个日志帮我在调试时定位了至少一半的问题。有一次 ReportAgent 输出全是空行,查日志发现原因是 CalcAgent 返回了空数组,而 ReportAgent 的提示词里没有针对空数组的处理说明。
5.2 上下文引用要面向“生成摘要”,而不是“复制全文”
我反复提到的一个调优原则是:每个 Agent 的输入只包含它完成任务所需的最小上下文。具体做法是在调度器里增加一个summarize阶段,用一次廉价的模型调用或者是简单的截断函数,把上游输出变成一段摘要。这个操作极大地降低了 token 消耗,也让模型注意力更集中。
例如 DataAgent 返回了 1000 行查询结果,真正给到 CalcAgent 的不是这 1000 行,而是这 1000 行经过预聚合后的几个关键指标。我用 Pandas 先做groupby和mean,再做to_string()传入 Agent,这样 CalcAgent 输入从几千 token 降到几十 token,准确率还提高了。
5.3 工具选择要留给 Agent 一个“无工具”选项
我见过很多框架强行让 Agent 每轮都调用工具,导致模型在没必要时也调用工具,然后返回方棍。在create_agent里,如果工具列表非空,LangChain 会让模型能调用工具,但不强制。我在子 Agent 的 Prompt 里明确写上“如果能根据输入数据直接计算,不要调用任何工具”,然后再设置工具的return_direct参数,让工具调用后直接返回结果而非交给模型重述。
5.4 防止 Agent 循环依赖的“祖父约束”
除了轮次上限,我用一个更精细的手段:在派发任务时给每个任务带一个trace_parent字段,调度器维护一个全局调用链,如果在调用链中发现同一个 Agent 即将第二次被调用,就强制改为直接返回上一轮结果。这样可以处理“A 上报给 B,B 又让 A 补充”这种递归情况,而不必等到轮次耗尽。
5.5 用一套固定测试集做回归,而不是每次靠嘴说“感觉变好了”
多智能体调优最忌随机漫步。我整理了一个包含 20 个任务的小测试集,每个任务有明确的预期输出类型和可接受的答案范围,每次改动路由逻辑或 Prompt 后,就跑一遍这个测试集,记录完整率、工具调用正确率、最终结果可接受率。这个做法让调优有了可衡量的基准,不再依赖个人手感。
5.6 模型选择:不同角色的模型能力要求可以不同
不是所有子 Agent 都需要用旗舰模型。在调度器里,我给每个 Agent 单独配置了模型名。路由 Agent 用最便宜的最小模型即可,因为它的任务只是判断派发方向;数据分析 Agent 如果加上工具调用,需要稍微强一点的模型;ReportAgent 是最终输出质量的关键,可以给最强模型。整个系统的 token 成本立即下降了一大截,同时最终报告质量没有明显变化。
6. 边界感:什么时候别用多智能体
6.1 拆分的成本要小于带来的收益
多智能体必然带来更多 API 调用、更高的延迟、更复杂的调试链路。如果你的任务本身没有明显的职责边界,比如“帮我写一首诗”,强行拆成“诗人 Agent”“评论家 Agent”“修改 Agent”,得到的“诗”通常还没有单 Agent 直接写好。因为每拆一层,都会增加信息损耗。我用一个简单判断标准:如果任务的中间步骤不需要不同的工具和不同的记忆,就别拆。
6.2 上下文小的时候,单 Agent 完全够用
用户请求只是“查询今天的天气”这种小任务,给单 Agent 一个天气工具就够了。调用一套多智能体框架反而会引入路由 API 的调用、任务序列化和结果反序列化,纯粹是画蛇添足。我把这些轻量任务直接加到调度器的“直通”分支:匹配到简单关键词,不派发子 Agent,直接用内置工具完成,效率和稳定性都更好。
6.3 多智能体不是非黑即白:混合架构才是最常用的终态
实际生产中我倾向的不是“所有地方都多智能体”,而是一个混合架构:顶层是一个主控 Agent,它手里同时握着简单工具和子 Agent 列表。面对请求时,主控先判断任务是“简单工具能直接完成的”还是“需要多个专业 Agent 协作的”。前者走工具调用,后者走任务路由,这样既保住了简单场景的响应速度,又在中高风险场景中获得协同能力。
以我自己的实践为例,最终框架里三个子 Agent 并没有时刻全部参与,很多时候只调用其中一个,这时它退化为一个普通 Agent。只有确实需要跨角色协作的时候,协同调度逻辑才真正发挥价值。这种边界的把控,比多智能体本身更重要,也更容易被忽略。
6.4 未来扩展的切入点
在现在框架的基础上,我还在考虑两个扩展点:一是让子 Agent 拥有自己的记忆分层,各自维护短期任务记忆独立上下文,长期记忆由聚合 Agent 统一管理;二是引入自反思机制,当一个 Agent 的输出校验失败时,不是简单报错,而是把失败原因和修复建议回传给上游 Agent,让它自我修正一次。这些方向都还在实验期,但框架的模块化结构让我可以比较平滑地测试和替换,不需要大规模重写。