去年年中我接到一个让我头疼了大半个月的需求团队攒了一大批零散的调研素材希望系统能在十分钟内把它变成一份多维度的分析报告——要包含市场判断、技术路线评估和潜在风险提示。一开始我用了最直觉的方案把全部材料塞进一个超长Prompt让同一个大模型一口气输出所有分析。结果非常不稳定回答越到后面越糊甚至出现同一个数字前面说增长、后面说下降的情况。折腾了两周之后我转向LangGraph用Multi-Agent架构重新搭了整个流程。这篇文章就是这套分析平台从0到1的完整实战记录会覆盖为什么放弃单Agent、LangGraph的核心机制、整体架构设计、核心代码实现以及我当时踩过的五个坑和后续的扩展思路。内容适合准备做多Agent编排、正在评估LangGraph、或者遇到过单Agent任务一多就失灵这类问题的朋友。不管你是刚入门还是已经写过一些LangGraph示例这套实战路径应该都能给你一些可复用的参考。1. 为什么放弃一个Prompt跑到底单Agent的三大痛点1.1 上下文污染长对话里的事实漂移最开始的问题不是模型推理能力不够而是输入的上下文太乱。一份调研素材里往往既有市场规模数据、又有技术架构细节还有合规和风险提示。把这些全部塞进同一个Prompt模型的注意力会被摊薄分析市场部分的时候还好写到技术部分时已经开始丢信息等写到风险部分时经常出现编造数据和结论前后打架。我当时的调试方式很笨每次输出不对就去改Prompt措辞要求它记住前面的结论。但问题在于这个记住根本没法被强制。后来我统计了一下将近20%的回答里存在事实漂移——同一个指标前面说增长后面说下降。这不是模型不行而是任务结构本身不合理。打个比方让一个什么都会的资深顾问同时负责市场、技术、风控三份报告还要求他一次性写完。他当然能做但大概率每个部分只能做到七成深度而且写到后面会记不清自己前面说了什么。Multi-Agent的思路很朴素把三份报告拆给三个专职顾问各管各的最后汇总的人只做整合。1.2 工具调用的串行瓶颈第二个痛点是效率。如果分析流程还需要调用外部工具——比如查行业数据库、读取接口返回的文档——单Agent模式下这些调用天然是串行的。一个Agent要先把材料读一遍判断该调哪个工具等工具返回再接着写分析。整个过程像一个人既当项目经理、又当分析师、又当资料员所有事情排队做。我当时需要同时拉取市场数据和解析技术文档单Agent一次完整运行大概要跑七八分钟而且经常在最后一步超时重来。拆成三个Agent之后市场和技术两个维度可以各自行动整体耗时反而压缩到原来的一半左右。这也让我意识到多Agent不只是能力分工很多时候是并行度带来的收益。1.3 角色混在一个Prompt里导致的指令膨胀单Agent方案还有一个维护问题。随着需求细化我的Prompt越来越长十几条角色设定、三四段输出格式说明、若干条领域知识再加一堆禁止编造数据之类的红线。每次改一个维度的要求整个Prompt都要跟着调整而且重跑结果经常漂移——因为模型对不同指令的注意力权重并不稳定。等到要同时支持市场分析、技术分析、风险评估三个角色时我意识到Prompt已经彻底失控了。与其继续堆指令不如把职责拆开每个Agent只维护自己那一小块Prompt。于是LangGraph进入视野它天然支持把Agent当作图里的节点来编排每个节点职责单一状态流转全靠显式的State比在Prompt里反复提醒模型要可靠得多。2. LangGraph的核心机制把Agent当图来编排2.1 StateGraph、Node、Edge三个基本概念LangGraph的整体思路是把一次完整任务当成一个有向图每个节点是一段处理逻辑比如一个Agent调用或一个数据处理函数。节点之间用边连接数据靠一个全局State对象在各节点之间传递。State是一个典型的Python字典用TypedDict定义字段。每个节点函数接收当前State处理后返回一个字典里面只有它要更新的字段。LangGraph会自动把这些返回值合并进全局State交给下一个节点。边上可以加条件逻辑根据State里的某个字段决定下一步走到哪个节点也可以直接连到END表示收尾。我这里有一个特别小的示例当初就是跑通它才真正理解这套抽象from langgraph.graph import StateGraph, END from typing import TypedDict class DemoState(TypedDict): message: str def first(state: DemoState) - dict: print(first:, state[message]) return {message: hello} def second(state: DemoState) - dict: print(second:, state[message]) return {} graph StateGraph(DemoState) graph.add_node(first, first) graph.add_node(second, second) graph.add_edge(first, second) graph.add_edge(second, END) app graph.compile()调用app.invoke({message: start})就能看到first和second顺序执行。这个示例虽然简单但已经包含了LangGraph的全部核心概念State定义全局数据节点处理业务边控制流转方向。2.2 为什么选LangGraph而不是手写状态机可能有人会说不就是几个函数按顺序跑吗自己写个循环加判断不就行了我当时确实先尝试了手写版本。等到需要处理三个Agent谁先跑、谁跑完了、要不要重试、超时了怎么办这些问题时手写代码的维护成本很快就上来了。手写方案的问题集中在三处一是状态更新散落在各个分支里调试的时候要翻很多行代码才能弄清楚某一步改了什么字段二是条件判断一旦复杂循环出口很容易漏掉出现死循环自己还不知道三是如果要支持断点续跑、记录执行轨迹又得自己造轮子。LangGraph把这几件事收敛到了图结构里。State天然是一份全局快照每次节点运行前后都能看到完整状态条件边把下一步去哪显式画在图上而不是藏在代码分支里编译后的应用还支持设置最大递归深度防止异常循环。对于我这种需要快速迭代、频繁调整编排逻辑的场景这套抽象比手写状态机省心得多。2.3 环境准备与工程目录规划我这边的环境很朴素Python 3.11虚拟环境安装langgraph依赖包再加上大模型SDK。工程目录按职责分成了四块schemas/存放State定义和各个Agent的输出结构。agents/每个Agent一个文件负责各自的Prompt和调用逻辑。graph/编排逻辑组装StateGraph。utils/通用工具比如JSON清洗、字段校验。这个分法不是什么标准答案但很有效先想清楚State里有哪些字段再写Agent先画好节点和边再去填函数内部逻辑。目录结构最大的价值是让整张图一目了然——看到graph文件夹就知道流程的主干看到agents文件夹就知道换哪个Agent时只需要改哪个文件。3. 分析平台整体架构Supervisor 三个分析Agent3.1 场景设定从原始文本到多维度分析报告平台接收的输入很简单一段或几段调研素材文本。期望输出是一份Markdown格式的分析报告包含三个固定章节市场分析、技术分析、风险评估最后来一段综合结论指出当前最值得关注的方向和最需要警惕的风险点。我采用的设计模式是Supervisor主管模式一个Supervisor节点充当调度者它不直接做分析而是根据当前State判断还缺哪个维度的结果依次调度对应的分析Agent所有分析结果齐了之后进入一个汇总节点由它生成最终报告。这种模式的好处是分析Agent之间彼此隔离、没有依赖。市场Agent不需要关心技术Agent写了什么反过来也一样。隔离带来的稳定性提升很明显任何单个Agent的输出异常都可以单独修复不会因为一个环节出错导致整条链路全崩。3.2 状态定义数据在Agent之间如何流转State里需要承载几类信息原始素材、各分析Agent的结果、状态标记、最终报告。我定义了下面这样一份数据结构后面会展开代码细节source_material是唯一不变的核心输入market_result、tech_result、risk_result是三个Agent各自的独立输出字段彼此不交叉final_report是汇总节点生成的结果iteration_count用于记录循环次数防止路由逻辑出问题时死循环。为什么要把每个Agent的结果拆成独立字段而不是用一个analysis_results列表因为LangGraph默认的State更新是合并字典语义如果所有Agent都去写同一个列表字段很容易互相覆盖或者需要额外的合并逻辑。拆成独立字段是最省心的做法也让路由和调试时的状态一眼可见。3.3 路由逻辑谁来决定下一步派谁上场Supervisor节点的核心逻辑实际上非常直白。每个分析Agent完成后会把对应的结果字段更新进State。Supervisor在每次入口处检查这三个字段如果market_result为空就路由到市场Agent如果market_result已有但tech_result为空就路由到技术Agent如果tech_result也有了但risk_result还空就路由到风险Agent如果三个字段都非空就进入汇总节点汇总完成结束。这套硬编码规则的优点是可预期、可调试。我不建议在一开始就让Supervisor用大模型做意图路由因为LLM的判断会有随机性一旦选错Agent后面整条链路都会被带偏。规则能覆盖90%的常见路径剩下的边界情况可以后续再叠加模型决策。4. 核心代码实战从State定义到Agent编排4.1 定义State数据结构State是整个平台的数据总线字段设计直接决定后续代码好不好写。我最终用到如下结构from typing import TypedDict class AnalysisState(TypedDict): source_material: str market_result: str tech_result: str risk_result: str final_report: str iteration_count: intsource_material存放原始素材三个result字段存三个Agent的分析输出final_report存汇总结果iteration_count是安全衡量器用来限制循环次数。这里有一个实际项目里的教训一开始我没有iteration_count字段结果有一次路由判断写错条件图里循环了四十多次才因为超时中断。加入这个字段之后每次路由前都给它加一超过阈值就强制进汇总节点相当于给图加上一道熔断保护。4.2 用工厂函数实现三个分析Agent三个分析Agent的业务逻辑基本一致读取source_material按各自的角色Prompt生成分析再返回结构化结果。用工厂函数生成节点避免复制粘贴三份代码def create_analysis_node(result_field: str, agent_prompt: str): def node(state: AnalysisState) - dict: analysis llm.invoke( agent_prompt.format(source_materialstate[source_material]) ) return {result_field: analysis.content} return node实际项目中我还会让Agent调用一个自定义的输出工具让模型以JSON的形式返回结果而不是直接吐正文。这样能极大减少输出里混着解释性文字的问题。工具定义类似这样class AnalysisOutput(BaseModel): summary: str Field(description核心结论不超过300字) key_points: list[str] Field(description三个关键要点) confidence: float Field(description信心评分0到1之间)三个Agent的Prompt各自独立市场Agent强调市场规模、竞争格局、增长趋势技术Agent强调架构设计、技术选型、实现风险风险Agent强调合规、供应链、资金链等方面的隐患。每一个Prompt都不需要知道另外两个Agent的存在这从根本上消除了Prompt互相干扰的问题。4.3 实现Supervisor路由Supervisor节点本身不调用LLM只做路由判断def supervisor_route(state: AnalysisState) - str: state[iteration_count] state.get(iteration_count, 0) 1 if state[iteration_count] 4: return summarize if not state.get(market_result): return market if not state.get(tech_result): return tech if not state.get(risk_result): return risk return summarize这段逻辑很简单但有一个细节值得强调判断结果是否为空时要小心空字符串和无内容的返回值。我遇到过Agent返回了本部分没有发现值得关注的市场信息这样的文本它非空但本质上没有完成分析任务。后来统一约定Agent的输出必须包含summary字段的JSON如果summary为空或confidence过低Supervisor可以选择重新调度该Agent。这个校验逻辑虽然不算复杂但对最终报告质量影响很大。4.4 组合成图并invoke有了节点和路由逻辑组合成图就是最后一步from langgraph.graph import StateGraph, END graph StateGraph(AnalysisState) graph.add_node(supervisor, supervisor_node) graph.add_node(market_agent, create_analysis_node(market_result, MARKET_PROMPT)) graph.add_node(tech_agent, create_analysis_node(tech_result, TECH_PROMPT)) graph.add_node(risk_agent, create_analysis_node(risk_result, RISK_PROMPT)) graph.add_node(summarizer, summarizer_node) graph.set_entry_point(supervisor) graph.add_conditional_edges( supervisor, supervisor_route, { market: market_agent, tech: tech_agent, risk: risk_agent, summarize: summarizer, } ) graph.add_edge(market_agent, supervisor) graph.add_edge(tech_agent, supervisor) graph.add_edge(risk_agent, supervisor) graph.add_edge(summarizer, END) app graph.compile() result app.invoke({source_material: source_text, iteration_count: 0})每个分析Agent执行完之后都会回到Supervisor重新路由。整个过程看起来像一个循环Supervisor判断缺谁派谁回来再判断。LangGraph不会限制这种循环反而提供了recursion_limit这类参数来控制最大执行步数防止意外死循环。汇总节点Summarizer的职责是把三个结果整合成一份连贯报告。它会读取三个result字段生成Markdown格式的完整报告存到final_report里。这里有一个关键选择汇总节点只读取三个分析结果字段不应再回头翻阅source_material做大段重组避免汇总阶段自己发明素材之外的细节。5. 跑通之后才是开始实测中的五个坑5.1 上下文越滚越长的解决方案第一次跑通全流程时我挺兴奋但很快发现越往后执行模型输出越离谱。排查了很久才找到原因我的State里有一个汇总列表字段每次循环都把三个Agent的完整输出追加进去导致历史文本被当作待分析的新材料传给其他节点上下文越滚越长模型开始把前面的分析结果当成新输入来再次分析。解决办法很直接把Agent的原始输出和精简摘要分开存储。原始分析全文只用于最后的报告生成路由判断和下游Agent只读取summary级别的精简字段。这样既保留了完整信息又避免无关上下文污染下一次判断。5.2 并行执行时的状态竞争曾经有一版设计里我想让三个分析Agent并行执行以提升速度。LangGraph对图内并行分支的支持当时还不够顺手于是我先用多线程方式同时调用三个Agent再手动合并结果。测试时发现如果三个线程几乎同时返回State的写入会出现竞争某个Agent的结果会莫名其妙丢失。后来我把方案改成节点内部可以并行调用多个工具但三个Agent仍然按顺序由Supervisor调度。实际损耗并不明显因为三个Agent之间本来就有先后依赖Supervisor要等前面的结果才能判断下一步而每个Agent内部的耗时大头在LLM调用顺序执行并不会显著变慢。如果想真正并行正确做法是使用LangGraph官方的并行节点能力让多个分支同时执行并在汇合点同步不要自己用线程去改同一个State。5.3 死循环保护与recursion_limit我前面提到过iteration_count这里再补充一个踩坑细节。起初我只是想在异常时用iteration_count兜底但并没有在编译时设置recursion_limit结果有一次条件判断写错图在Supervisor和缺失字段的Agent之间来回跳了上百次直到请求超时才发现。实际生产中我同时做了三层保护State里iteration_count递增并强制汇总compile前设置recursion_limit30超过步数直接抛异常路由函数里最后添加一个兜底分支任何未预期返回值都归一为进入汇总节点。这三层保护叠起来基本杜绝了死循环烧钱的问题。注意recursion_limit不是靠感觉随便设的。设太大会掩盖路由bug设太小会导致正常的多次分析被拦腰截断。我一般先统计正常任务的平均步数再留出两三倍余量作为上限。5.4 结构化输出的解析失败让Agent直接输出JSON非常不靠谱。我在初版代码中让模型以JSON格式回答结果经常混入好的这是分析结果这类开头语json.loads直接报错。后来改成让Agent调用自定义输出工具模型的回复被强制转换成工具参数格式稳定了一大截。但即使这样偶尔还会出现字段缺失或值类型不对的情况。所以我专门写了一个sanitize函数先尝试正常解析失败则用正则提取JSON片段再解析再失败则调用一次修复Prompt让模型重写最后还是失败就把结果降级为该维度分析失败并在报告中标注需人工确认。这套降级链虽然不优雅但保证了整个平台不会因为一个格式问题挂掉。5.5 调试技巧让每个Node都可以单独单测调试Multi-Agent图最难受的地方是出错时很难定位是哪个节点的问题。后来我养成了一个习惯每个Agent节点先脱离Graph单独测试。写临时脚本直接构造一份最小化的State字典调用对应的node函数只看它的返回值是否符合预期。这样能把问题精确缩小到某个节点的输出格式不对还是图结构路由错误。另一个实用技巧是在每个节点函数入口打印当前State的关键字段摘要。打印多了虽然刷屏但在调试阶段非常管用我能一眼看到某一步之后哪个字段被填充了、哪个字段还是空的。跑通之后再把这些print注释掉或者改成日志输出不影响线上逻辑。6. 从分析平台到通用编排底座扩展思考6.1 换模型的成本与收益平台跑稳定之后我开始琢磨成本优化。最直观的尝试是把三个分析Agent从旗舰模型换成更便宜的模型只保留Supervisor路由它基本不调LLM和Summarizer用旗舰模型。实测下来便宜模型在格式规范性和结论一致性上确实弱一些但因为每个Agent的Prompt足够聚焦、任务足够单一最终报告的质量损失并不大成本却下降了一半以上。这个结论让我意识到多Agent架构的价值不只是分工还在于它天然支持按节点差异化用模型。如果还是单Prompt单体模式想混用模型基本不可能。现在这个平台可以灵活地给风险Agent配更贵的模型给市场Agent配便宜模型互不影响。6.2 替换Agent内部的业务逻辑分析类Agent的节点结构是读材料、按Prompt生成结论。同样的节点外壳可以换成完全不同的业务逻辑比如接一个数据库查询Agent或者一个调用外部API的Agent。Supervisor的路由判断只需要看某个任务是否完成这个抽象状态并不关心节点内部到底调用了什么工具。所以在后续项目里我基本把这一套Supervisor 多个Worker 汇总器当成了通用编排底座。需要接入新能力时只需要新增一个节点、定义一个新的State字段、在路由表里加一个分支。这种扩展成本是单体Agent方案完全给不了的。6.3 路由策略升级规则兜底与LLM决策我前面说一开始不推荐直接用LLM做路由但这不意味着规则永远最好。当前平台上分析维度固定为三个硬编码路由最合适可一旦维度变成十几个而且用户的输入请求是动态的硬编码就会疲于应付。一个比较稳的升级路径是先用规则把大部分常见路径匹配掉规则匹配不到时再调一个LLM做意图路由。给LLM的输入只包含可用节点列表和State中缺失信息摘要让它返回一个节点名。为了防止LLM乱选在路由函数里设置白名单校验输出不合法就回到规则兜底。这样既保留了LLM的灵活性又不会被它的随机性拖垮。写到最后说一点个人体会。这个平台从需求到跑通第一版前后花了大概两周其中真正搭图只花了一天多剩下时间全部耗在Prompt调试、输出格式校验、循环保护和状态设计上。所以如果你准备做类似的事情我的建议是别急着写Agent逻辑先把State字段设计清楚把三个Agent的输出Schema定义好再谈LangGraph的图结构。状态定义清楚了图和路由其实都是顺水推舟的事。还有一个小技巧每个Agent的输出字段设计成加字段容易、删字段难后续扩展维度时你会感谢当初这个决定。
去年年中我接到一个让我头疼了大半个月的需求团队攒了一大批零散的调研素材希望系统能在十分钟内把它变成一份多维度的分析报告——要包含市场判断、技术路线评估和潜在风险提示。一开始我用了最直觉的方案把全部材料塞进一个超长Prompt让同一个大模型一口气输出所有分析。结果非常不稳定回答越到后面越糊甚至出现同一个数字前面说增长、后面说下降的情况。折腾了两周之后我转向LangGraph用Multi-Agent架构重新搭了整个流程。这篇文章就是这套分析平台从0到1的完整实战记录会覆盖为什么放弃单Agent、LangGraph的核心机制、整体架构设计、核心代码实现以及我当时踩过的五个坑和后续的扩展思路。内容适合准备做多Agent编排、正在评估LangGraph、或者遇到过单Agent任务一多就失灵这类问题的朋友。不管你是刚入门还是已经写过一些LangGraph示例这套实战路径应该都能给你一些可复用的参考。1. 为什么放弃一个Prompt跑到底单Agent的三大痛点1.1 上下文污染长对话里的事实漂移最开始的问题不是模型推理能力不够而是输入的上下文太乱。一份调研素材里往往既有市场规模数据、又有技术架构细节还有合规和风险提示。把这些全部塞进同一个Prompt模型的注意力会被摊薄分析市场部分的时候还好写到技术部分时已经开始丢信息等写到风险部分时经常出现编造数据和结论前后打架。我当时的调试方式很笨每次输出不对就去改Prompt措辞要求它记住前面的结论。但问题在于这个记住根本没法被强制。后来我统计了一下将近20%的回答里存在事实漂移——同一个指标前面说增长后面说下降。这不是模型不行而是任务结构本身不合理。打个比方让一个什么都会的资深顾问同时负责市场、技术、风控三份报告还要求他一次性写完。他当然能做但大概率每个部分只能做到七成深度而且写到后面会记不清自己前面说了什么。Multi-Agent的思路很朴素把三份报告拆给三个专职顾问各管各的最后汇总的人只做整合。1.2 工具调用的串行瓶颈第二个痛点是效率。如果分析流程还需要调用外部工具——比如查行业数据库、读取接口返回的文档——单Agent模式下这些调用天然是串行的。一个Agent要先把材料读一遍判断该调哪个工具等工具返回再接着写分析。整个过程像一个人既当项目经理、又当分析师、又当资料员所有事情排队做。我当时需要同时拉取市场数据和解析技术文档单Agent一次完整运行大概要跑七八分钟而且经常在最后一步超时重来。拆成三个Agent之后市场和技术两个维度可以各自行动整体耗时反而压缩到原来的一半左右。这也让我意识到多Agent不只是能力分工很多时候是并行度带来的收益。1.3 角色混在一个Prompt里导致的指令膨胀单Agent方案还有一个维护问题。随着需求细化我的Prompt越来越长十几条角色设定、三四段输出格式说明、若干条领域知识再加一堆禁止编造数据之类的红线。每次改一个维度的要求整个Prompt都要跟着调整而且重跑结果经常漂移——因为模型对不同指令的注意力权重并不稳定。等到要同时支持市场分析、技术分析、风险评估三个角色时我意识到Prompt已经彻底失控了。与其继续堆指令不如把职责拆开每个Agent只维护自己那一小块Prompt。于是LangGraph进入视野它天然支持把Agent当作图里的节点来编排每个节点职责单一状态流转全靠显式的State比在Prompt里反复提醒模型要可靠得多。2. LangGraph的核心机制把Agent当图来编排2.1 StateGraph、Node、Edge三个基本概念LangGraph的整体思路是把一次完整任务当成一个有向图每个节点是一段处理逻辑比如一个Agent调用或一个数据处理函数。节点之间用边连接数据靠一个全局State对象在各节点之间传递。State是一个典型的Python字典用TypedDict定义字段。每个节点函数接收当前State处理后返回一个字典里面只有它要更新的字段。LangGraph会自动把这些返回值合并进全局State交给下一个节点。边上可以加条件逻辑根据State里的某个字段决定下一步走到哪个节点也可以直接连到END表示收尾。我这里有一个特别小的示例当初就是跑通它才真正理解这套抽象from langgraph.graph import StateGraph, END from typing import TypedDict class DemoState(TypedDict): message: str def first(state: DemoState) - dict: print(first:, state[message]) return {message: hello} def second(state: DemoState) - dict: print(second:, state[message]) return {} graph StateGraph(DemoState) graph.add_node(first, first) graph.add_node(second, second) graph.add_edge(first, second) graph.add_edge(second, END) app graph.compile()调用app.invoke({message: start})就能看到first和second顺序执行。这个示例虽然简单但已经包含了LangGraph的全部核心概念State定义全局数据节点处理业务边控制流转方向。2.2 为什么选LangGraph而不是手写状态机可能有人会说不就是几个函数按顺序跑吗自己写个循环加判断不就行了我当时确实先尝试了手写版本。等到需要处理三个Agent谁先跑、谁跑完了、要不要重试、超时了怎么办这些问题时手写代码的维护成本很快就上来了。手写方案的问题集中在三处一是状态更新散落在各个分支里调试的时候要翻很多行代码才能弄清楚某一步改了什么字段二是条件判断一旦复杂循环出口很容易漏掉出现死循环自己还不知道三是如果要支持断点续跑、记录执行轨迹又得自己造轮子。LangGraph把这几件事收敛到了图结构里。State天然是一份全局快照每次节点运行前后都能看到完整状态条件边把下一步去哪显式画在图上而不是藏在代码分支里编译后的应用还支持设置最大递归深度防止异常循环。对于我这种需要快速迭代、频繁调整编排逻辑的场景这套抽象比手写状态机省心得多。2.3 环境准备与工程目录规划我这边的环境很朴素Python 3.11虚拟环境安装langgraph依赖包再加上大模型SDK。工程目录按职责分成了四块schemas/存放State定义和各个Agent的输出结构。agents/每个Agent一个文件负责各自的Prompt和调用逻辑。graph/编排逻辑组装StateGraph。utils/通用工具比如JSON清洗、字段校验。这个分法不是什么标准答案但很有效先想清楚State里有哪些字段再写Agent先画好节点和边再去填函数内部逻辑。目录结构最大的价值是让整张图一目了然——看到graph文件夹就知道流程的主干看到agents文件夹就知道换哪个Agent时只需要改哪个文件。3. 分析平台整体架构Supervisor 三个分析Agent3.1 场景设定从原始文本到多维度分析报告平台接收的输入很简单一段或几段调研素材文本。期望输出是一份Markdown格式的分析报告包含三个固定章节市场分析、技术分析、风险评估最后来一段综合结论指出当前最值得关注的方向和最需要警惕的风险点。我采用的设计模式是Supervisor主管模式一个Supervisor节点充当调度者它不直接做分析而是根据当前State判断还缺哪个维度的结果依次调度对应的分析Agent所有分析结果齐了之后进入一个汇总节点由它生成最终报告。这种模式的好处是分析Agent之间彼此隔离、没有依赖。市场Agent不需要关心技术Agent写了什么反过来也一样。隔离带来的稳定性提升很明显任何单个Agent的输出异常都可以单独修复不会因为一个环节出错导致整条链路全崩。3.2 状态定义数据在Agent之间如何流转State里需要承载几类信息原始素材、各分析Agent的结果、状态标记、最终报告。我定义了下面这样一份数据结构后面会展开代码细节source_material是唯一不变的核心输入market_result、tech_result、risk_result是三个Agent各自的独立输出字段彼此不交叉final_report是汇总节点生成的结果iteration_count用于记录循环次数防止路由逻辑出问题时死循环。为什么要把每个Agent的结果拆成独立字段而不是用一个analysis_results列表因为LangGraph默认的State更新是合并字典语义如果所有Agent都去写同一个列表字段很容易互相覆盖或者需要额外的合并逻辑。拆成独立字段是最省心的做法也让路由和调试时的状态一眼可见。3.3 路由逻辑谁来决定下一步派谁上场Supervisor节点的核心逻辑实际上非常直白。每个分析Agent完成后会把对应的结果字段更新进State。Supervisor在每次入口处检查这三个字段如果market_result为空就路由到市场Agent如果market_result已有但tech_result为空就路由到技术Agent如果tech_result也有了但risk_result还空就路由到风险Agent如果三个字段都非空就进入汇总节点汇总完成结束。这套硬编码规则的优点是可预期、可调试。我不建议在一开始就让Supervisor用大模型做意图路由因为LLM的判断会有随机性一旦选错Agent后面整条链路都会被带偏。规则能覆盖90%的常见路径剩下的边界情况可以后续再叠加模型决策。4. 核心代码实战从State定义到Agent编排4.1 定义State数据结构State是整个平台的数据总线字段设计直接决定后续代码好不好写。我最终用到如下结构from typing import TypedDict class AnalysisState(TypedDict): source_material: str market_result: str tech_result: str risk_result: str final_report: str iteration_count: intsource_material存放原始素材三个result字段存三个Agent的分析输出final_report存汇总结果iteration_count是安全衡量器用来限制循环次数。这里有一个实际项目里的教训一开始我没有iteration_count字段结果有一次路由判断写错条件图里循环了四十多次才因为超时中断。加入这个字段之后每次路由前都给它加一超过阈值就强制进汇总节点相当于给图加上一道熔断保护。4.2 用工厂函数实现三个分析Agent三个分析Agent的业务逻辑基本一致读取source_material按各自的角色Prompt生成分析再返回结构化结果。用工厂函数生成节点避免复制粘贴三份代码def create_analysis_node(result_field: str, agent_prompt: str): def node(state: AnalysisState) - dict: analysis llm.invoke( agent_prompt.format(source_materialstate[source_material]) ) return {result_field: analysis.content} return node实际项目中我还会让Agent调用一个自定义的输出工具让模型以JSON的形式返回结果而不是直接吐正文。这样能极大减少输出里混着解释性文字的问题。工具定义类似这样class AnalysisOutput(BaseModel): summary: str Field(description核心结论不超过300字) key_points: list[str] Field(description三个关键要点) confidence: float Field(description信心评分0到1之间)三个Agent的Prompt各自独立市场Agent强调市场规模、竞争格局、增长趋势技术Agent强调架构设计、技术选型、实现风险风险Agent强调合规、供应链、资金链等方面的隐患。每一个Prompt都不需要知道另外两个Agent的存在这从根本上消除了Prompt互相干扰的问题。4.3 实现Supervisor路由Supervisor节点本身不调用LLM只做路由判断def supervisor_route(state: AnalysisState) - str: state[iteration_count] state.get(iteration_count, 0) 1 if state[iteration_count] 4: return summarize if not state.get(market_result): return market if not state.get(tech_result): return tech if not state.get(risk_result): return risk return summarize这段逻辑很简单但有一个细节值得强调判断结果是否为空时要小心空字符串和无内容的返回值。我遇到过Agent返回了本部分没有发现值得关注的市场信息这样的文本它非空但本质上没有完成分析任务。后来统一约定Agent的输出必须包含summary字段的JSON如果summary为空或confidence过低Supervisor可以选择重新调度该Agent。这个校验逻辑虽然不算复杂但对最终报告质量影响很大。4.4 组合成图并invoke有了节点和路由逻辑组合成图就是最后一步from langgraph.graph import StateGraph, END graph StateGraph(AnalysisState) graph.add_node(supervisor, supervisor_node) graph.add_node(market_agent, create_analysis_node(market_result, MARKET_PROMPT)) graph.add_node(tech_agent, create_analysis_node(tech_result, TECH_PROMPT)) graph.add_node(risk_agent, create_analysis_node(risk_result, RISK_PROMPT)) graph.add_node(summarizer, summarizer_node) graph.set_entry_point(supervisor) graph.add_conditional_edges( supervisor, supervisor_route, { market: market_agent, tech: tech_agent, risk: risk_agent, summarize: summarizer, } ) graph.add_edge(market_agent, supervisor) graph.add_edge(tech_agent, supervisor) graph.add_edge(risk_agent, supervisor) graph.add_edge(summarizer, END) app graph.compile() result app.invoke({source_material: source_text, iteration_count: 0})每个分析Agent执行完之后都会回到Supervisor重新路由。整个过程看起来像一个循环Supervisor判断缺谁派谁回来再判断。LangGraph不会限制这种循环反而提供了recursion_limit这类参数来控制最大执行步数防止意外死循环。汇总节点Summarizer的职责是把三个结果整合成一份连贯报告。它会读取三个result字段生成Markdown格式的完整报告存到final_report里。这里有一个关键选择汇总节点只读取三个分析结果字段不应再回头翻阅source_material做大段重组避免汇总阶段自己发明素材之外的细节。5. 跑通之后才是开始实测中的五个坑5.1 上下文越滚越长的解决方案第一次跑通全流程时我挺兴奋但很快发现越往后执行模型输出越离谱。排查了很久才找到原因我的State里有一个汇总列表字段每次循环都把三个Agent的完整输出追加进去导致历史文本被当作待分析的新材料传给其他节点上下文越滚越长模型开始把前面的分析结果当成新输入来再次分析。解决办法很直接把Agent的原始输出和精简摘要分开存储。原始分析全文只用于最后的报告生成路由判断和下游Agent只读取summary级别的精简字段。这样既保留了完整信息又避免无关上下文污染下一次判断。5.2 并行执行时的状态竞争曾经有一版设计里我想让三个分析Agent并行执行以提升速度。LangGraph对图内并行分支的支持当时还不够顺手于是我先用多线程方式同时调用三个Agent再手动合并结果。测试时发现如果三个线程几乎同时返回State的写入会出现竞争某个Agent的结果会莫名其妙丢失。后来我把方案改成节点内部可以并行调用多个工具但三个Agent仍然按顺序由Supervisor调度。实际损耗并不明显因为三个Agent之间本来就有先后依赖Supervisor要等前面的结果才能判断下一步而每个Agent内部的耗时大头在LLM调用顺序执行并不会显著变慢。如果想真正并行正确做法是使用LangGraph官方的并行节点能力让多个分支同时执行并在汇合点同步不要自己用线程去改同一个State。5.3 死循环保护与recursion_limit我前面提到过iteration_count这里再补充一个踩坑细节。起初我只是想在异常时用iteration_count兜底但并没有在编译时设置recursion_limit结果有一次条件判断写错图在Supervisor和缺失字段的Agent之间来回跳了上百次直到请求超时才发现。实际生产中我同时做了三层保护State里iteration_count递增并强制汇总compile前设置recursion_limit30超过步数直接抛异常路由函数里最后添加一个兜底分支任何未预期返回值都归一为进入汇总节点。这三层保护叠起来基本杜绝了死循环烧钱的问题。注意recursion_limit不是靠感觉随便设的。设太大会掩盖路由bug设太小会导致正常的多次分析被拦腰截断。我一般先统计正常任务的平均步数再留出两三倍余量作为上限。5.4 结构化输出的解析失败让Agent直接输出JSON非常不靠谱。我在初版代码中让模型以JSON格式回答结果经常混入好的这是分析结果这类开头语json.loads直接报错。后来改成让Agent调用自定义输出工具模型的回复被强制转换成工具参数格式稳定了一大截。但即使这样偶尔还会出现字段缺失或值类型不对的情况。所以我专门写了一个sanitize函数先尝试正常解析失败则用正则提取JSON片段再解析再失败则调用一次修复Prompt让模型重写最后还是失败就把结果降级为该维度分析失败并在报告中标注需人工确认。这套降级链虽然不优雅但保证了整个平台不会因为一个格式问题挂掉。5.5 调试技巧让每个Node都可以单独单测调试Multi-Agent图最难受的地方是出错时很难定位是哪个节点的问题。后来我养成了一个习惯每个Agent节点先脱离Graph单独测试。写临时脚本直接构造一份最小化的State字典调用对应的node函数只看它的返回值是否符合预期。这样能把问题精确缩小到某个节点的输出格式不对还是图结构路由错误。另一个实用技巧是在每个节点函数入口打印当前State的关键字段摘要。打印多了虽然刷屏但在调试阶段非常管用我能一眼看到某一步之后哪个字段被填充了、哪个字段还是空的。跑通之后再把这些print注释掉或者改成日志输出不影响线上逻辑。6. 从分析平台到通用编排底座扩展思考6.1 换模型的成本与收益平台跑稳定之后我开始琢磨成本优化。最直观的尝试是把三个分析Agent从旗舰模型换成更便宜的模型只保留Supervisor路由它基本不调LLM和Summarizer用旗舰模型。实测下来便宜模型在格式规范性和结论一致性上确实弱一些但因为每个Agent的Prompt足够聚焦、任务足够单一最终报告的质量损失并不大成本却下降了一半以上。这个结论让我意识到多Agent架构的价值不只是分工还在于它天然支持按节点差异化用模型。如果还是单Prompt单体模式想混用模型基本不可能。现在这个平台可以灵活地给风险Agent配更贵的模型给市场Agent配便宜模型互不影响。6.2 替换Agent内部的业务逻辑分析类Agent的节点结构是读材料、按Prompt生成结论。同样的节点外壳可以换成完全不同的业务逻辑比如接一个数据库查询Agent或者一个调用外部API的Agent。Supervisor的路由判断只需要看某个任务是否完成这个抽象状态并不关心节点内部到底调用了什么工具。所以在后续项目里我基本把这一套Supervisor 多个Worker 汇总器当成了通用编排底座。需要接入新能力时只需要新增一个节点、定义一个新的State字段、在路由表里加一个分支。这种扩展成本是单体Agent方案完全给不了的。6.3 路由策略升级规则兜底与LLM决策我前面说一开始不推荐直接用LLM做路由但这不意味着规则永远最好。当前平台上分析维度固定为三个硬编码路由最合适可一旦维度变成十几个而且用户的输入请求是动态的硬编码就会疲于应付。一个比较稳的升级路径是先用规则把大部分常见路径匹配掉规则匹配不到时再调一个LLM做意图路由。给LLM的输入只包含可用节点列表和State中缺失信息摘要让它返回一个节点名。为了防止LLM乱选在路由函数里设置白名单校验输出不合法就回到规则兜底。这样既保留了LLM的灵活性又不会被它的随机性拖垮。写到最后说一点个人体会。这个平台从需求到跑通第一版前后花了大概两周其中真正搭图只花了一天多剩下时间全部耗在Prompt调试、输出格式校验、循环保护和状态设计上。所以如果你准备做类似的事情我的建议是别急着写Agent逻辑先把State字段设计清楚把三个Agent的输出Schema定义好再谈LangGraph的图结构。状态定义清楚了图和路由其实都是顺水推舟的事。还有一个小技巧每个Agent的输出字段设计成加字段容易、删字段难后续扩展维度时你会感谢当初这个决定。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
