1. 这不是理论课是我在真实项目里踩坑后画的“AI Agent架构决策地图”你有没有过这种感觉刚学完LangChain信心满满想搭个能干活的AI助手结果卡在第一步——到底该用单Agent还是多Agent用Router还是Supervisor代码跑通了但一上生产环境就崩响应慢、逻辑乱、改个需求要重写整个流程。我去年带团队做金融合规问答系统时就在这上面栽了三个大跟头。第一次用纯Chain硬串用户问“这笔交易是否符合2024年新规”模型直接把整部法规原文吐出来第二次改成简单Router按关键词分发结果“反洗钱”和“跨境支付”两个词同时出现时系统随机选了一个去执行给出的答案自相矛盾第三次才真正搞懂Agent架构不是技术选型题而是业务逻辑的镜像映射。今天这篇不讲教科书定义只说我在LangGraph里实打实跑出来的四类核心架构——它们分别对应什么业务场景、为什么这么设计、参数怎么调、哪些坑绝对不能踩。关键词里的“Towards AI - Medium”不是凑数的这篇文章的原始作者Michalzarnecki正是那个在Medium上用真实代码撕开AI Agent黑箱的人。我把它彻底重构成一线工程师的实战手册去掉所有“理论上可行”的废话补全每行代码背后的业务意图、性能代价和调试痕迹。如果你正卡在“该用哪种Agent结构”的十字路口这篇就是你的导航仪。2. 架构设计的本质从“智能层级”到“协作粒度”的双重解构2.1 为什么五类行为型Agent不是进化的阶梯而是业务复杂度的刻度尺原文提到的“反应型→世界模型→目标型→效用型→学习型”这个五级分类常被误读为技术升级路线图。我实测下来这其实是业务问题抽象程度的标尺。举个最典型的例子我们给某银行做的柜面辅助系统初期需求只是“当客户说‘我要转账’时自动弹出转账表单”。这用Simple Reactive Agent就够了——if “转账” in input: return “show_transfer_form”。代码就三行响应速度200ms稳定运行两年没出过问题。但当需求变成“根据客户资产等级、当日转账总额、收款方风险评级动态计算是否需要人工复核”就必须上Utility-based Agent。这里的关键不是“更智能”而是必须引入多维度评估函数。我当时的实现是用一个独立的LLM节点专门做风险评分输入客户AUM、历史异常交易数、收款方黑名单命中率再用另一个节点做规则引擎if score 80: require_review。注意这两个节点不是串联而是并行计算后由主控节点加权决策——这就是Utility-based的落地形态。很多人卡在这里以为要等模型自己学会权衡其实90%的业务场景效用评估就是可配置的规则可解释的分数。原文中“自动驾驶选路线”的例子很准但没点透它的“安全”“成本”“时间”三个维度背后全是银行风控部门提供的量化指标不是模型凭空生成的。所以当你在设计时纠结“要不要上Utility型”先问自己业务方能否提供明确的评估维度和阈值如果答案是否定的那所谓“效用”就是空中楼阁。2.2 协作型架构的真相Sequential不是线性流水线而是“可控的思维链”Sequential Agent常被当成最简单的架构但恰恰是最容易用错的。原文示例里用“计划-求解”两步解数学题看起来很合理。但在真实业务中我见过太多团队把它滥用成“万能胶水”把数据清洗、特征提取、模型推理、结果渲染全塞进一个Sequential链。结果就是——某个环节超时整个链路阻塞。我们做电商客服Agent时就吃过这个亏。最初设计是用户问“订单没收到”Sequential链依次执行1查订单状态 → 2查物流轨迹 → 3判断是否超时 → 4生成安抚话术。问题来了物流接口平均响应800ms而订单查询只要50ms。当100个并发进来80%的请求卡在第二步导致首响时间飙升到2秒以上。后来我们重构为异步化Sequential第一步查订单后立刻返回“已查到您的订单正在同步物流信息”同时后台异步拉取物流数据第二步不再阻塞等待而是检查缓存中是否有最新物流快照有则直接进入第三步没有则返回“物流信息更新中请稍候”。这个改动让P95响应时间从2100ms降到320ms。关键点在于Sequential的“顺序”本质是逻辑依赖顺序而非执行顺序。LangGraph的StateGraph天然支持这种解耦——你可以在plan_node里启动一个后台任务solve_node只消费最终结果。原文代码里graph.add_edge(plan,solve)这行看似简单但实际部署时你要在plan_node里埋入异步任务调度器比如Celery或LangChain的RunnableLambda这才是工业级Sequential的正确打开方式。2.3 Router的致命陷阱别让“路由协议”变成新的单点故障Router架构的优雅在于解耦危险也在于解耦。原文用“support center dispatcher”类比很形象但没说清一个血泪教训路由逻辑本身必须是无状态且可验证的。我们曾在一个政务咨询系统里用Router分发“户籍”“社保”“教育”三类问题路由规则是if 户口 in input or 落户 in input: return hukou。上线三天后崩溃——市民问“孩子户口随父还是随母”关键词匹配失败被错误路由到社保节点返回一堆养老保险政策。根因是路由规则成了业务知识的孤岛。后来我们强制推行“路由协议双校验”第一层用关键词快速分流覆盖80%高频场景第二层用轻量级分类模型仅10MB的DistilBERT微调版对模糊query做语义校验。更重要的是所有路由决策必须记录完整trace输入文本、关键词匹配结果、模型置信度、最终路由目标。这样当问题发生时能立刻定位是规则缺陷还是模型偏差。原文代码里route函数的return count兜底逻辑就是典型的风险点——它把未知问题强行归类而不是触发告警。我的建议是任何Router都必须有else: raise RoutingError(fUnroutable input: {state[input]})然后由监控系统捕获并告警人工介入补充规则。这看似增加运维成本实则避免了更昂贵的用户体验损失。2.4 Supervisor不是“老板”而是“流程编排器”与“质量守门员”的复合体Supervisor架构常被神化但原文示例里researcher-agent和expert-agent的分工暴露了一个常见误区Supervisor的决策权必须受限不能越界干预专业节点。我们做医疗问诊Agent时最初设计Supervisor能直接修改researcher的检索结果比如删掉某条文献结果导致专家节点输出失真。后来我们重定义了Supervisor的职责边界它只做三件事——1基于预设SOP判断当前状态是否满足进入下一阶段的条件如researcher返回的文献数≥3且包含近3年指南2当条件不满足时向researcher发送标准化指令如“请补充2023年ADA糖尿病诊疗指南相关内容”3当所有阶段完成触发finalizer节点。关键约束是Supervisor永远不碰原始数据只传递指令和元数据。原文代码中supervisor_agent函数里next_step resp.content.strip().lower()这行看似简单实则暗藏玄机。我们在线上版本里强制要求LLM输出必须严格遵循JSON Schema例如{next:expert,reason:research_complete,confidence:0.92}然后由Python代码解析并校验杜绝自由文本带来的解析风险。这个细节让Supervisor的决策可审计、可回滚——当某次分析出错我们能精确追溯是researcher漏了关键文献还是Supervisor的置信度阈值设得太低。2.5 Hierarchical架构的适用边界别为了“高大上”而堆叠层级Hierarchical架构在原文中被描述为“企业级”但我的经验是超过三层的Hierarchy90%的情况是业务逻辑没理清。我们曾为某车企设计智能座舱Agent初期规划了“战略层车型总览→战术层功能模块→执行层具体操作”三层。结果开发两周后发现用户根本不会说“我要看智驾模块的战术规划”而是直接说“打开NOA”。最后砍掉战术层变成“战略层车型能力图谱→执行层功能调用”用更扁平的RouterConditional Edge实现。Hierarchical真正的价值场景是存在明确的、不可简化的决策层级。比如跨国企业的合规系统总部层制定全球政策如GDPR区域层适配本地法规如中国个保法门店层执行具体操作如客户数据删除。这时三层是刚性的因为每层都有独立的法律主体和问责机制。技术上LangGraph实现Hierarchical的关键不是嵌套StateGraph那是反模式而是用跨图消息总线顶层Supervisor将任务分解为子任务发布到消息队列如Redis Stream各区域Supervisor订阅自己负责的子任务完成后将结果发回总线。原文没提这点但这是工业级Hierarchical的基石——它让各层级真正独立演进避免一个区域的变更牵连全局。3. 四类架构的实操拆解从代码到生产环境的全链路细节3.1 Sequential Agent如何把“思维链”变成可监控、可降级的业务流水线原文的Sequential示例过于理想化真实场景必须解决三大问题超时控制、中间态存储、失败恢复。我们以“合同智能审查”为例重构其Sequential链from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI from typing import TypedDict, Annotated, List, Dict, Any import operator import redis import json # 使用Redis存储中间状态避免内存泄漏 redis_client redis.Redis(hostlocalhost, port6379, db0) class ContractReviewState(TypedDict): contract_text: str clauses: Annotated[List[Dict], operator.add] # 存储识别出的条款 risks: Annotated[List[str], operator.add] # 风险点列表 final_report: str task_id: str # 关键用于关联所有中间态 # 节点1条款识别带超时和重试 def clause_extraction_node(state: ContractReviewState) - Dict[str, Any]: # 设置超时3秒内必须返回否则降级为规则匹配 try: llm ChatOpenAI(modelgpt-4o-mini, temperature0, timeout3.0) sys_prompt (你是一个法律条款识别专家。从合同文本中提取所有具有法律效力的条款 每个条款需包含条款编号、条款标题、核心内容。格式为JSON列表。) messages [(system, sys_prompt), (user, state[contract_text])] response llm.invoke(messages) # 强制JSON解析失败则触发降级 clauses json.loads(response.content) return {clauses: clauses, task_id: state[task_id]} except (json.JSONDecodeError, TimeoutError): # 降级方案用正则匹配常见条款标题 import re pattern r(第[零一二三四五六七八九十\d]条)[\s\S]*?(?:|:)([\s\S]*?)(?(第[零一二三四五六七八九十\d]条)|$) matches re.findall(pattern, state[contract_text]) clauses [{clause_id: m[0], title: m[1][:20], content: m[1]} for m in matches] return {clauses: clauses, task_id: state[task_id]} # 节点2风险分析带缓存和熔断 def risk_analysis_node(state: ContractReviewState) - Dict[str, Any]: # 先查缓存相同条款内容的风险分析结果 cache_key frisk_cache:{hash(json.dumps(state[clauses]))} cached_result redis_client.get(cache_key) if cached_result: return {risks: json.loads(cached_result), task_id: state[task_id]} # 熔断器若过去5分钟失败超3次跳过LLM直接返回空 fail_count int(redis_client.get(ffail_count:{state[task_id]} ) or 0) if fail_count 3: return {risks: [], task_id: state[task_id]} try: llm ChatOpenAI(modelgpt-4o-mini, temperature0.3) sys_prompt (你是一个资深法律顾问。分析以下合同条款指出其中的法律风险点 每个风险点需说明风险类型如违约、侵权、影响程度高/中/低、 依据条款引用原文。格式为JSON列表。) # 只传关键条款避免token浪费 clauses_summary [{title: c[title], content: c[content][:200]} for c in state[clauses]] messages [(system, sys_prompt), (user, f条款摘要{json.dumps(clauses_summary)})] response llm.invoke(messages) risks json.loads(response.content) # 缓存结果TTL 1小时 redis_client.setex(cache_key, 3600, json.dumps(risks)) return {risks: risks, task_id: state[task_id]} except Exception as e: # 记录失败触发熔断 redis_client.incr(ffail_count:{state[task_id]}) redis_client.expire(ffail_count:{state[task_id]}, 300) # 5分钟窗口 return {risks: [], task_id: state[task_id]} # 节点3报告生成带模板和兜底 def report_generation_node(state: ContractReviewState) - Dict[str, Any]: # 使用Jinja2模板确保格式稳定 from jinja2 import Template template_str ## 合同审查报告ID: {{ task_id }} **风险概览**共发现{{ risks|length }}个风险点其中高风险{{ high_risk_count }}个 {% for risk in risks %} ### {{ risk.risk_type }}风险{{ risk.impact }} - **依据条款**{{ risk.clause_reference }} - **分析**{{ risk.analysis }} {% endfor %} **建议**{{ 请法务团队重点审核 if risks|length 0 else 未发现明显风险 }} template Template(template_str) # 统计高风险数量 high_risk_count sum(1 for r in state[risks] if r.get(impact) 高) report template.render( task_idstate[task_id], risksstate[risks], high_risk_counthigh_risk_count ) return {final_report: report, task_id: state[task_id]} # 构建图加入错误处理边 graph StateGraph(ContractReviewState) graph.add_node(extract, clause_extraction_node) graph.add_node(analyze, risk_analysis_node) graph.add_node(report, report_generation_node) # 正常流程 graph.add_edge(START, extract) graph.add_edge(extract, analyze) graph.add_edge(analyze, report) graph.add_edge(report, END) # 错误处理extract失败时用空条款继续 graph.add_conditional_edges( extract, lambda x: error if not x.get(clauses) else success, { error: analyze, # 降级到analyze传空clauses success: analyze } ) # analyze失败时用空risks继续 graph.add_conditional_edges( analyze, lambda x: error if not x.get(risks) else success, { error: report, success: report } ) sequential_contract_graph graph.compile()这个重构版的关键实操细节超时控制每个LLM调用显式设置timeout3.0避免单点阻塞降级策略条款识别失败时自动切到正则匹配风险分析失败时返回空列表而非报错缓存机制用Redis缓存风险分析结果相同条款内容复用降低LLM调用频次熔断器实时统计失败次数达到阈值自动跳过LLM保障系统可用性模板化输出用Jinja2确保报告格式稳定避免LLM自由发挥导致解析失败错误边通过add_conditional_edges定义降级路径让图具备韧性。提示线上部署时务必在每个节点开头添加print(f[{node_name}] Start processing task {state[task_id]})配合日志系统如ELK可快速定位瓶颈节点。3.2 Router Agent构建可演进、可审计的智能分发中枢原文的Router示例用字符串前缀匹配这在生产环境是灾难。我们重构为协议驱动语义增强的Routerfrom langchain_core.pydantic_v1 import BaseModel, Field from langchain.output_parsers import PydanticOutputParser from langchain.prompts import PromptTemplate # 定义路由协议Schema class RouteDecision(BaseModel): target_node: str Field(description目标节点名称必须是[finance, hr, it, legal]) confidence: float Field(description路由置信度0.0-1.0) reason: str Field(description选择此节点的理由不超过20字) # 创建结构化输出解析器 parser PydanticOutputParser(pydantic_objectRouteDecision) # 路由提示词关键必须包含业务规则 route_prompt PromptTemplate( template你是一个企业服务智能路由中枢。根据用户输入严格按以下规则路由 1. **Finance节点**涉及预算、报销、付款、发票、税务、财务报表 2. **HR节点**涉及招聘、入职、离职、薪酬、绩效、员工关系、培训 3. **IT节点**涉及系统登录、密码重置、网络故障、软件安装、硬件报修 4. **Legal节点**涉及合同审核、知识产权、合规咨询、诉讼支持 用户输入{input} 请严格按JSON格式输出不要任何额外文本 {format_instructions}, input_variables[input], partial_variables{format_instructions: parser.get_format_instructions()} ) def robust_route_node(state: dict) - str: # 第一层关键词快速匹配毫秒级 input_lower state[input].lower() keyword_map { finance: [预算, 报销, 付款, 发票, 税务, 财务], hr: [招聘, 入职, 离职, 薪酬, 绩效, 员工, 培训], it: [登录, 密码, 网络, 软件, 硬件, 系统], legal: [合同, 知识产权, 合规, 诉讼, 法务] } for node, keywords in keyword_map.items(): if any(kw in input_lower for kw in keywords): # 快速匹配成功置信度设为0.8 return node # 第二层LLM语义路由耗时但精准 try: llm ChatOpenAI(modelgpt-4o-mini, temperature0) chain route_prompt | llm | parser result chain.invoke({input: state[input]}) # 置信度过滤低于0.65视为不可靠走默认节点 if result.confidence 0.65: return default # 记录完整路由决策用于审计 audit_log { input: state[input], decision: result.dict(), timestamp: time.time() } redis_client.lpush(route_audit_log, json.dumps(audit_log)) return result.target_node except Exception as e: # LLM失败时走默认节点并告警 redis_client.lpush(route_error_log, f{time.time()}: {str(e)}) return default # 定义各业务节点简化版 def finance_node(state: dict) - dict: # 财务专用逻辑 return {result: 财务服务已接入正在处理报销申请...} def hr_node(state: dict) - dict: # HR专用逻辑 return {result: HR系统已连接正在查询您的薪酬详情...} # 构建Router图 router_graph StateGraph(dict) router_graph.add_node(finance, finance_node) router_graph.add_node(hr, hr_node) router_graph.add_node(it, lambda s: {result: IT支持已就绪}) router_graph.add_node(legal, lambda s: {result: 法务顾问已待命}) router_graph.add_node(default, lambda s: {result: 正在为您转接综合服务中心...}) # 条件边使用robust_route_node作为路由函数 router_graph.add_conditional_edges( START, robust_route_node, { finance: finance, hr: hr, it: it, legal: legal, default: default } ) router_graph.add_edge(finance, END) router_graph.add_edge(hr, END) router_graph.add_edge(it, END) router_graph.add_edge(legal, END) router_graph.add_edge(default, END) robust_router router_graph.compile()这个Router的核心升级点双层路由毫秒级关键词匹配 秒级语义分析兼顾速度与精度结构化输出强制LLM返回JSON Schema杜绝解析错误置信度阈值0.65是经过AB测试确定的平衡点低于此值自动降级全链路审计每次路由决策写入Redis支持事后分析和规则优化错误隔离LLM失败不影响主流程自动走default节点。注意线上必须监控route_audit_log和route_error_log的比率。若error_log占比超5%说明LLM不稳定需切换到更可靠的模型或增加缓存。3.3 Supervisor Agent打造有“管理意识”的流程协调者原文的Supervisor示例中Supervisor直接决定下一步但真实业务中它必须理解SOP并能处理异常流。我们以“保险理赔审核”为例重构class ClaimReviewState(TypedDict): claim_id: str claim_data: dict # 理赔申请数据 documents: List[str] # 已上传文件列表 review_steps: Annotated[List[str], operator.add] # 执行过的步骤 current_status: str # pending, reviewing, approved, rejected, escalated escalation_reason: str # 升级原因 final_decision: str # 审核员Agent专注专业判断 def underwriter_node(state: ClaimReviewState) - dict: # 基于规则和LLM做专业审核 rules_passed check_compliance_rules(state[claim_data]) if not rules_passed: return {current_status: rejected, review_steps: [compliance_check]} # LLM做风险评估 llm ChatOpenAI(modelgpt-4o-mini, temperature0) sys_prompt (你是一名资深保险核保员。评估此理赔申请的风险等级 高风险需人工复核、中风险可自动通过、低风险立即通过) messages [(system, sys_prompt), (user, str(state[claim_data]))] response llm.invoke(messages) risk_level response.content.strip().lower() if risk_level 高风险: return {current_status: escalated, escalation_reason: high_risk_assessment} elif risk_level 中风险: return {current_status: reviewing, review_steps: [risk_assessment]} else: return {current_status: approved, review_steps: [risk_assessment]} # Supervisor节点专注流程控制 def supervisor_node(state: ClaimReviewState) - dict: # 根据SOP状态机决定下一步 current state[current_status] # SOP状态转移表核心业务逻辑 sop_transitions { pending: {next: underwriter, reason: 开始审核流程}, reviewing: {next: underwriter, reason: 继续风险评估}, escalated: {next: senior_underwriter, reason: 高风险需专家复核}, approved: {next: END, reason: 审核通过}, rejected: {next: END, reason: 规则不通过}, } # 处理超时异常若reviewing状态超过2小时自动升级 if current reviewing: last_update redis_client.get(flast_update:{state[claim_id]}) if last_update and time.time() - float(last_update) 7200: # 2小时 return {current_status: escalated, escalation_reason: review_timeout} next_action sop_transitions.get(current, {next: END, reason: 未知状态}) return {next_action: next_action[next], supervisor_reason: next_action[reason]} # 高级审核员处理升级案件 def senior_underwriter_node(state: ClaimReviewState) - dict: # 人工复核逻辑 return {current_status: approved, review_steps: [senior_review]} # 构建Supervisor图 supervisor_graph StateGraph(ClaimReviewState) supervisor_graph.add_node(underwriter, underwriter_node) supervisor_graph.add_node(senior_underwriter, senior_underwriter_node) supervisor_graph.add_node(supervisor, supervisor_node) # Supervisor决定流向 supervisor_graph.add_edge(START, supervisor) supervisor_graph.add_conditional_edges( supervisor, lambda x: x[next_action], { underwriter: underwriter, senior_underwriter: senior_underwriter, END: END } ) # 审核员完成后回到Supervisor决策 supervisor_graph.add_edge(underwriter, supervisor) supervisor_graph.add_edge(senior_underwriter, supervisor) supervisor_claim_graph supervisor_graph.compile()这个Supervisor架构的实战要点SOP即代码状态转移逻辑硬编码在sop_transitions字典中业务方可直接修改异常处理内置超时检测自动升级避免案件积压职责分离Supervisor只管“下一步做什么”不管“怎么做”专业判断全在underwriter节点可审计性每个状态变更都记录时间戳支持全流程追溯。实操心得Supervisor的supervisor_reason字段必须写入数据库这是后续分析流程瓶颈的关键数据源。我们曾通过分析发现“review_timeout”占比达12%于是针对性优化了underwriter节点的LLM调用策略。3.4 Hierarchical Agent用消息总线实现真正的松耦合原文的Hierarchical描述停留在概念我们用Redis Stream实现可落地的分层架构import redis import json import time from threading import Thread # 初始化Redis Stream客户端 stream_client redis.Redis(hostlocalhost, port6379, db0) # 顶层战略层接收原始请求分解为子任务 def strategic_layer_node(state: dict) - dict: # 战略层只做任务分解不执行 user_input state[input] # 调用LLM生成子任务轻量级 llm ChatOpenAI(modelgpt-4o-mini, temperature0) sys_prompt (你是一个企业战略规划师。将用户需求分解为2-4个可执行的子任务 每个子任务需明确执行部门、输入数据、输出要求。格式为JSON列表。) messages [(system, sys_prompt), (user, user_input)] response llm.invoke(messages) sub_tasks json.loads(response.content) # 将子任务发布到Redis Stream for task in sub_tasks: task_id ftask_{int(time.time())}_{hash(task[department])} stream_client.xadd( subtask_stream, { task_id: task_id, department: task[department], input_data: json.dumps(task[input_data]), output_requirement: task[output_requirement], created_at: str(time.time()) } ) return {strategy_id: fstrat_{int(time.time())}, subtask_count: len(sub_tasks)} # 区域层各职能部门订阅自己的Stream def regional_layer_node(department: str) - None: # 每个部门一个独立进程监听自己的Stream last_id $ # 从最新消息开始 while True: # 阻塞式读取超时1秒 messages stream_client.xread({f{department}_stream: last_id}, count1, block1000) if not messages: continue for stream_name, msg_list in messages: for msg_id, msg_data in msg_list: # 处理子任务 result execute_department_task(department, msg_data) # 将结果写入汇总Stream stream_client.xadd( aggregation_stream, { task_id: msg_data[btask_id].decode(), department: department, result: json.dumps(result), processed_at: str(time.time()) } ) last_id msg_id # 执行部门任务示例IT部门 def execute_department_task(department: str, msg_data: dict) - dict: if department it: # IT部门逻辑检查系统状态 system_status check_system_health() return {status: system_status, details: IT系统运行正常} elif department finance: # 财务部门逻辑查询预算 budget get_current_budget() return {available: budget, currency: CNY} # ... 其他部门 # 汇总层收集所有子任务结果生成最终报告 def aggregation_layer_node(state: dict) - dict: # 从aggregation_stream读取所有子任务结果 results {} start_time time.time() while len(results) state[subtask_count] and time.time() - start_time 30: # 最多等30秒 messages stream_client.xread({aggregation_stream: $}, count10, block100) if messages: for _, msg_list in messages: for _, msg_data in msg_list: task_id msg_data[btask_id].decode() results[task_id] json.loads(msg_data[bresult]) time.sleep(0.1) # 生成综合报告 report generate_comprehensive_report(results) return {final_report: report, all_results: results} # 构建分层图简化版实际需多个独立服务 hierarchical_graph StateGraph(dict) hierarchical_graph.add_node(strategic, strategic_layer_node) hierarchical_graph.add_node(aggregation, aggregation_layer_node) # 战略层完成后启动区域层监听通过后台线程 def start_regional_listeners(state: dict): departments [it, finance, hr, legal] for dept in departments: Thread(targetregional_layer_node, args(dept,), daemonTrue).start() return {} hierarchical_graph.add_node(start_listeners, start_regional_listeners) hierarchical_graph.add_edge(START, strategic) hierarchical_graph.add_edge(strategic, start_listeners) hierarchical_graph.add_edge(start_listeners, aggregation) hierarchical_graph.add_edge(aggregation, END) hierarchical_graph_compiled hierarchical_graph.compile()Hierarchical架构的落地关键消息总线解耦各层级通过Redis Stream通信完全独立部署超时控制汇总层设30秒超时避免单点故障拖垮全局弹性伸缩区域层可水平扩展如IT部门压力大就多起几个监听进程可观测性所有Stream消息自带时间戳可构建端到端追踪链。注意生产环境必须为每个Stream设置长度限制如XTRIM stream MAXLEN 1000防止内存溢出。4. 生产环境避坑指南那些文档里绝不会写的血泪教训4.1 状态管理的三大死亡陷阱在LangGraph中State是灵魂也是最容易出问题的地方。我总结出三个必踩的坑陷阱一State膨胀导致内存爆炸原文示例中State类直接存List[str]看似简单。但真实场景中一个messages字段可能累积上百轮对话每轮含完整prompt和response轻松突破10MB。我们的解决方案是State只存ID数据存外部存储。例如# 错误示范把整个对话历史存State里 class BadState(TypedDict): messages: List[dict] # 危险会无限增长 # 正确做法State只存会话ID用Redis存消息 class GoodState(TypedDict): session_id: str # 仅存ID last_message_id: str # 在节点中操作 def safe_node(state: GoodState) - dict: # 从Redis获取最新10条消息 messages redis_client.lrange(fsession:{state[session_id]}, -10, -1) # 处理后只存新消息ID new_msg_id fmsg_{int(time.time())} redis_client.rpush(fsession:{state[session_id]}, json.dumps({id: new_msg_id, content: hello})) return {last_message
