LangGraph实战：用有向图重构LLM Agent工程化范式

张

张建站

2026/7/14 3:36:23

10分钟阅读

1. 这不是又一个“LLMAgent”的概念课而是一份能跑通、能调试、能上线的实战手记我从2023年中开始系统性地把LLM Agent落地到真实业务场景里——不是PoC演示不是Jupyter Notebook里的玩具demo而是每天处理3000条客服工单分类、自动填充维修报告、实时校验合同条款合规性的生产级流程。过程中踩过太多坑状态丢失、循环调用、工具链断裂、提示词漂移、异常无日志、重试逻辑失控……直到LangGraph出现我才第一次在代码层面真正“看见”了Agent的运行脉络。它不只是一套新API而是一种可建模、可中断、可回溯、可压测的Agent工程范式。这篇笔记标题里带“#1”但内容绝非泛泛而谈的“介绍”。我会直接从你打开VS Code后第一行该写什么开始讲起拆解LangGraph如何用有向图DAG结构替代传统while-loop轮询为什么State必须是不可变数据类、Node函数为何要严格遵循dict → dict签名、ConditionalEdge的判断逻辑怎么避免隐式状态污染、以及最关键的——当你的Agent在凌晨三点卡死在某个分支时你该看哪三行日志、改哪两个参数、加哪三个断点。如果你正被LangChain的AgentExecutor绕晕、被LlamaIndex的QueryEngine封装过深而无法干预中间步骤、或刚学完LangGraph文档却连最简单的“问天气→查API→总结回答”都跑不通那这篇就是为你写的。它不教你怎么背API而是带你亲手把Agent的“心跳”可视化出来。2. 为什么必须放弃“链式调用”转而拥抱“图式编排”2.1 传统Agent框架的隐性代价你永远不知道它卡在哪一步先说个真实案例。去年我们给某银行做智能尽调助手需求很明确输入企业名称→爬取工商信息→解析股权结构→比对关联方风险→生成简报。用LangChain v0.1的AgentExecutor实现后测试时一切正常上线首周平均响应时间从1.2秒飙升到8.7秒错误率从0.3%跳到12%。排查三天才发现当工商API返回空数据时Tool层抛出异常但AgentExecutor的handle_tool_error回调没被触发——因为它的错误捕获逻辑藏在_call方法的try-catch嵌套第5层且默认重试3次每次重试都重新初始化整个Agent状态。更致命的是所有中间步骤如“已获取工商数据”、“正在解析JSON”完全不可见日志里只有两行“Starting agent run”和“Agent finished with error”。你根本没法区分这是网络超时、模型幻觉还是JSON Schema校验失败。提示LangChain的AgentExecutor本质是状态黑盒轮询驱动。它用一个while循环不断调用LLM靠agent_scratchpad拼接历史靠intermediate_steps存临时结果。这种设计在单步任务如“计算123*456”中足够轻量但一旦涉及多工具协同、条件分支、人工审核介入、长周期异步等待就会迅速失控——因为你无法在任意时刻暂停、检查、修改、重启Agent的状态。2.2 LangGraph的破局点把Agent变成一张“活的流程图”LangGraph的核心思想极其朴素Agent不是一段代码而是一个有状态的图Stateful Graph。这个图由三类元素构成Nodes节点代表具体执行单元可以是LLM调用、工具函数、人工审核接口、甚至数据库查询。每个Node接收当前State返回更新后的State。Edges边定义节点间的流转关系。分为两类add_edge(node_a, node_b)无条件直连类似函数调用add_conditional_edges(node_a, route_function)根据route_function的返回值动态选择下一节点类似if-else或switchState状态贯穿全图的唯一数据载体必须是不可变对象immutable通常用TypedDict或pydantic.BaseModel定义。所有Node只能读取并返回新State不能修改原State。这种设计带来的改变是根本性的维度传统AgentLangChainLangGraph Agent状态可见性隐藏在intermediate_steps等私有属性中需hack源码才能打印State是显式参数每步执行后可直接print(state)或写入监控系统执行可控性while循环由框架控制无法在任意节点暂停/恢复可调用app.invoke()单步执行或app.stream()流式迭代精确控制生命周期错误定位异常堆栈指向AgentExecutor._call需层层反推错误直接抛在具体Node函数内堆栈清晰指向weather_tool.py:42分支逻辑依赖LLM输出特定字符串如TOOL:search_weather易受提示词扰动route_function是纯Python逻辑可接入规则引擎、数据库查询、甚至另一个LLM可测试性需启动完整LLM链路mock成本高每个Node可独立单元测试State输入/输出可完全mock我实测过同样一个“用户问‘上海今天热吗’→查天气→判断是否需提醒带伞”的流程在LangChain中调试需反复修改system prompt、观察LLM输出token在LangGraph中我只需在weather_node里加一行logger.info(fCalling weather API with {state[city]})再在route_after_weather里打个断点就能看到state里temperature: 32.5, precipitation_chance: 85然后手动返回suggest_umbrella——整个过程不到10秒。2.3 为什么State必须不可变一次内存泄漏的血泪教训很多人初学LangGraph时会忽略State的设计规范直接用dict甚至dataclass。这会导致灾难性后果。去年我们有个Agent需要处理PDF合同流程是parse_pdf → extract_clauses → validate_risk → generate_summary。初期用dict作为State# ❌ 危险写法可变dict导致状态污染 def parse_pdf_node(state: dict) - dict: state[pages] pdf_to_text(state[pdf_path]) # 直接修改原dict return state上线后发现当多个用户并发请求时A用户的PDF内容会“泄露”到B用户的state[pages]里。排查两天才定位到问题——Python中dict是可变对象app.invoke()传入的state引用被多个Node共享parse_pdf_node的修改直接影响后续所有Node。更隐蔽的是app.stream()流式调用时state会被多次复用导致上一轮的summary残留覆盖下一轮的clauses。LangGraph强制要求State为不可变类型正是为了解决这个问题。正确做法是# ✅ 安全写法使用TypedDict 显式拷贝 from typing import TypedDict, Optional class AgentState(TypedDict): user_query: str pdf_path: str pages: Optional[str] clauses: Optional[list] risk_score: Optional[float] summary: Optional[str] def parse_pdf_node(state: AgentState) - AgentState: # 创建新字典不修改原state return { **state, # 展开原有字段 pages: pdf_to_text(state[pdf_path]), # 只更新需要的字段 }或者用pydantic.BaseModel推荐自带验证from pydantic import BaseModel class AgentState(BaseModel): user_query: str pdf_path: str pages: str | None None clauses: list | None None risk_score: float | None None summary: str | None None def parse_pdf_node(state: AgentState) - AgentState: # copy()创建新实例原state完全隔离 return state.copy(update{pages: pdf_to_text(state.pdf_path)})注意state.copy(update{...})是pydantic的深拷贝确保即使clauses是嵌套列表也不会共享引用。我在金融场景中处理过含200嵌套字典的合同条款用copy()后内存占用稳定在12MB以内若用dict(state)浅拷贝GC压力会飙升QPS下降40%。3. 从零搭建第一个LangGraph Agent天气查询器的逐行实现3.1 环境准备与依赖锁定为什么pip install langgraph不够用LangGraph虽是LangChain生态一员但版本兼容性极敏感。我踩过的最大坑是langgraph0.1.19与langchain-core0.1.42组合下add_conditional_edges会静默忽略route_function的返回值始终走默认分支。原因在于langchain-core的Runnable基类在0.1.42版重构了invoke方法签名而LangGraph 0.1.19未适配。实操建议已验证通过# 创建干净虚拟环境强烈推荐 python -m venv langgraph-env source langgraph-env/bin/activate # Linux/Mac # langgraph-env\Scripts\activate # Windows # 安装精确版本2024年10月最新稳定组合 pip install --upgrade pip pip install langgraph0.1.25 \ langchain-core0.1.51 \ langchain-openai0.1.16 \ openai1.35.11 \ httpx0.27.0 \ pydantic2.7.1提示httpx0.27.0是关键。LangGraph底层用httpx.AsyncClient管理异步请求0.27.0版修复了连接池复用bug。若用0.26.x高并发时会出现Connection pool is full错误导致Agent卡死。验证安装是否成功# test_install.py from langgraph.graph import StateGraph from langgraph.checkpoint.memory import MemorySaver print(✅ LangGraph imports successfully) print(f✅ Version: {StateGraph.__module__})运行无报错即表示环境就绪。3.2 定义State用Pydantic建模比TypedDict更安全我们以“天气查询建议”为例State需承载用户原始问题、解析后的城市名、API返回的天气数据、最终建议。用pydantic.BaseModel定义# state.py from pydantic import BaseModel, Field from typing import Optional, Dict, Any class WeatherData(BaseModel): temperature: float Field(..., description当前温度单位摄氏度) condition: str Field(..., description天气状况如晴、多云、小雨) precipitation_chance: float Field(..., description降水概率0.0~1.0) humidity: int Field(..., description湿度百分比0~100) class AgentState(BaseModel): # 必填字段用户原始输入 user_query: str Field(..., description用户自然语言提问如上海今天热吗) # 可选字段各环节产出 city: Optional[str] Field(None, description解析出的城市名) weather_data: Optional[WeatherData] Field(None, description天气API返回数据) suggestion: Optional[str] Field(None, description最终给用户的建议如今天较热建议穿短袖) # 元数据字段用于调试和监控 step_count: int Field(0, description当前执行步数用于防死循环) error_message: Optional[str] Field(None, description最近一次错误信息)为什么不用TypedDictTypedDict仅提供类型提示运行时无校验。而pydantic.BaseModel在state.copy(update{...})时会自动验证字段类型。例如若天气API返回temperature: 32.5字符串而非floatBaseModel会在构造WeatherData时抛出ValidationError错误堆栈精准指向weather_node而非等到generate_suggestion_node里state.weather_data.temperature 30时报TypeError——后者调试成本高3倍以上。3.3 编写Nodes每个函数都是纯逻辑拒绝副作用LangGraph的Node必须是纯函数Pure Function输入State输出State不读写全局变量、不操作文件、不发HTTP请求除非封装在工具内。我们将流程拆为4个NodeNode 1解析城市名LLM调用# nodes.py from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from state import AgentState llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 提示词模板强制输出纯城市名避免LLM画蛇添足 prompt ChatPromptTemplate.from_messages([ (system, 你是一个精准的城市名提取器。请从用户问题中提取唯一城市名只输出城市名不要任何标点、解释或额外文字。例如北京明天会下雨吗 → 北京上海和杭州哪个更热 → 上海), (human, {user_query}) ]) extract_city_chain prompt | llm | StrOutputParser() def extract_city_node(state: AgentState) - AgentState: try: city extract_city_chain.invoke({user_query: state.user_query}) # 清洗输出去除空格、换行、引号 city city.strip().strip(\).replace(\n, ) return state.copy(update{city: city, step_count: state.step_count 1}) except Exception as e: return state.copy(update{ error_message: fCity extraction failed: {str(e)}, step_count: state.step_count 1 })Node 2调用天气API工具函数import httpx from state import AgentState, WeatherData # 封装天气API此处用Mock实际替换为真实API async def get_weather_async(city: str) - WeatherData: # 实际项目中这里调用和风天气、OpenWeatherMap等API # 为演示返回固定数据 if city in [上海, 北京, 广州]: return WeatherData( temperature32.5, condition晴, precipitation_chance0.05, humidity65 ) else: return WeatherData( temperature28.0, condition多云, precipitation_chance0.3, humidity72 ) async def weather_api_node(state: AgentState) - AgentState: if not state.city: return state.copy(update{ error_message: No city extracted, step_count: state.step_count 1 }) try: weather_data await get_weather_async(state.city) return state.copy(update{ weather_data: weather_data, step_count: state.step_count 1 }) except Exception as e: return state.copy(update{ error_message: fWeather API call failed: {str(e)}, step_count: state.step_count 1 })Node 3生成建议LLM调用from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser suggestion_prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的天气顾问。根据提供的天气数据用中文给出简洁、实用的穿衣/出行建议。要求1. 仅输出建议不要重复天气数据2. 语气亲切像朋友提醒3. 若温度30℃且降水概率20%建议防晒若降水概率70%建议带伞其他情况给出通用建议。), (human, 城市{city}温度{temp}℃天气{condition}降水概率{precip}%湿度{hum}%) ]) suggestion_chain suggestion_prompt | llm | StrOutputParser() def generate_suggestion_node(state: AgentState) - AgentState: if not state.weather_data: return state.copy(update{ error_message: No weather data to generate suggestion, step_count: state.step_count 1 }) try: suggestion suggestion_chain.invoke({ city: state.city, temp: state.weather_data.temperature, condition: state.weather_data.condition, precip: int(state.weather_data.precipitation_chance * 100), hum: state.weather_data.humidity }) return state.copy(update{ suggestion: suggestion.strip(), step_count: state.step_count 1 }) except Exception as e: return state.copy(update{ error_message: fSuggestion generation failed: {str(e)}, step_count: state.step_count 1 })Node 4错误处理兜底节点def error_handler_node(state: AgentState) - AgentState: # 当发生错误时返回友好提示 if state.error_message: fallback_msg f抱歉服务暂时遇到问题{state.error_message[:50]}... return state.copy(update{suggestion: fallback_msg}) return state3.4 构建Graph用add_conditional_edges实现智能路由这才是LangGraph的灵魂所在。我们定义路由逻辑当weather_data存在时走generate_suggestion否则走error_handler。# graph.py from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver from nodes import ( extract_city_node, weather_api_node, generate_suggestion_node, error_handler_node ) from state import AgentState # 初始化图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(extract_city, extract_city_node) workflow.add_node(get_weather, weather_api_node) workflow.add_node(generate_suggestion, generate_suggestion_node) workflow.add_node(error_handler, error_handler_node) # 设置入口点 workflow.set_entry_point(extract_city) # 定义条件边从extract_city出发 # 如果city为空直接去error_handler workflow.add_conditional_edges( extract_city, lambda state: error_handler if not state.city else get_weather, { error_handler: error_handler, get_weather: get_weather } ) # 从get_weather出发检查weather_data是否存在 workflow.add_conditional_edges( get_weather, lambda state: generate_suggestion if state.weather_data else error_handler, { generate_suggestion: generate_suggestion, error_handler: error_handler } ) # 从generate_suggestion和error_handler都直接结束 workflow.add_edge(generate_suggestion, END) workflow.add_edge(error_handler, END) # 添加记忆检查点支持断点续跑 checkpointer MemorySaver() app workflow.compile(checkpointercheckpointer)关键细节解析lambda state: ...是路由函数必须返回字符串对应add_conditional_edges的path_map键。path_map中的键如generate_suggestion必须与图中已注册的节点名完全一致大小写敏感。END是LangGraph预定义的终止节点调用app.invoke()时会在此停止。3.5 运行与调试三种调用方式的适用场景方式1app.invoke()—— 单次完整执行适合API后端# run_invoke.py from graph import app from state import AgentState # 输入初始状态 initial_state AgentState(user_query上海今天热吗) # 执行 result app.invoke(initial_state) print( 最终结果 ) print(f城市: {result.city}) print(f温度: {result.weather_data.temperature}℃) print(f建议: {result.suggestion}) print(f总步数: {result.step_count}) # 输出: # 最终结果 # 城市: 上海 # 温度: 32.5℃ # 建议: 今天较热紫外线强建议穿短袖并涂抹防晒霜哦~ # 总步数: 4方式2app.stream()—— 流式观察每一步调试神器# run_stream.py from graph import app from state import AgentState initial_state AgentState(user_query北京明天下雨吗) # 流式执行每步输出 for step in app.stream(initial_state): node_name list(step.keys())[0] # 获取当前执行的节点名 state list(step.values())[0] # 获取该节点输出的state print(f\n--- 步骤 {state.step_count}: {node_name} ---) print(f城市: {state.city or N/A}) print(f温度: {state.weather_data.temperature if state.weather_data else N/A}℃) print(f错误: {state.error_message or None})输出效果--- 步骤 1: extract_city --- 城市: 北京温度: N/A℃ 错误: None --- 步骤 2: get_weather --- 城市: 北京温度: 28.0℃ 错误: None --- 步骤 3: generate_suggestion --- 城市: 北京温度: 28.0℃ 错误: None实操心得stream()是调试核心。当Agent卡住时立即改用stream()看最后停在哪一步。如果停在get_weather且state.error_message为空说明API调用超时——此时应检查httpx.AsyncClient的timeout配置如果停在extract_city且state.city为空说明LLM没提取出城市需优化提示词。方式3app.get_state()—— 断点续跑生产必备# run_checkpoint.py from graph import app from state import AgentState # 第一次执行到get_weather后暂停 initial_state AgentState(user_query广州今天湿度大吗) config {configurable: {thread_id: test-001}} result app.invoke(initial_state, configconfig) # 查看当前状态可用于前端展示“正在查询天气...” current_state app.get_state(config) print(f当前节点: {current_state.values.city}) # 广州 # 模拟网络延迟后继续执行 final_result app.invoke(None, configconfig) # 传None表示从断点继续 print(f最终建议: {final_result.suggestion})MemorySaver会将中间状态存于内存thread_id作为会话ID。生产环境应替换为PostgresSaver或MongoDBSaver。4. 生产级避坑指南那些文档里不会写的12个致命细节4.1 Node函数的超时与重试别让一个慢API拖垮整个AgentLangGraph默认不提供Node级超时。若get_weather_async因网络问题卡住30秒app.invoke()会同步阻塞30秒。解决方案# 在weather_api_node中添加超时 import asyncio from httpx import Timeout async def weather_api_node(state: AgentState) - AgentState: if not state.city: return state.copy(update{error_message: No city}) try: # 设置10秒超时 timeout Timeout(10.0, connect5.0, read5.0) async with httpx.AsyncClient(timeouttimeout) as client: # ... 调用API pass except httpx.TimeoutException: return state.copy(update{error_message: Weather API timeout}) except Exception as e: return state.copy(update{error_message: fWeather API error: {e}})重试策略推荐对网络类Node用tenacity库实现指数退避from tenacity import retry, stop_after_attempt, wait_exponential retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max10) ) async def get_weather_async(city: str) - WeatherData: # ... API调用逻辑 pass4.2 条件边的“幽灵分支”route_function返回None的灾难这是新手最高频的Bug。看这段错误代码# ❌ 错误route_function可能返回None workflow.add_conditional_edges( get_weather, lambda state: generate_suggestion if state.weather_data else None, # ← 返回None {generate_suggestion: generate_suggestion} )当state.weather_data为None时lambda返回NoneLangGraph找不到对应分支直接抛出KeyError: None且错误堆栈不指向你的lambda而是指向langgraph/graph.py内部极难定位。正确写法必须覆盖所有可能返回值# ✅ 正确显式声明default分支 def route_after_weather(state: AgentState) - str: if state.weather_data: return generate_suggestion else: return error_handler # 明确返回已注册的节点名 workflow.add_conditional_edges( get_weather, route_after_weather, { generate_suggestion: generate_suggestion, error_handler: error_handler } )4.3 State字段的“空值陷阱”None vs 空字符串 vs 未定义Pydantic的Optional[str]字段在state.copy(update{})时行为微妙# 初始state: cityNone state AgentState(user_querytest) # ❌ 错误这样写不会清除city字段 state.copy(update{city: None}) # city仍为None无变化 # ✅ 正确用unset显式清除 from pydantic import Undefined state.copy(update{city: Undefined}) # city变为未设置状态但在LangGraph中更推荐统一用None表示“未获取”并在Node中显式判断def generate_suggestion_node(state: AgentState) - AgentState: # 安全判断用is None而非not state.city避免空字符串误判 if state.weather_data is None: return state.copy(update{error_message: No weather data})4.4 内存泄漏的根源State中存了不该存的东西曾有个Agent在处理大文件时OOM崩溃。排查发现AgentState里存了pdf_bytes: bytes而MemorySaver会序列化整个State存入内存。10MB PDF × 100并发 1GB内存瞬间占满。解决方案绝不存二进制数据pdf_bytes应存为临时文件路径/tmp/xxx.pdfNode执行完立即os.remove大文本截断pages: str字段限制长度超长时存摘要用weakref管理缓存对需复用的LLM结果用weakref.WeakValueDictionary缓存避免强引用4.5 日志埋点的黄金位置3个必打日志点没有日志的Agent等于盲人开车。我在每个Node入口和出口都加日志import logging logger logging.getLogger(__name__) def extract_city_node(state: AgentState) - AgentState: logger.info(f[NODE:extract_city] START | query{state.user_query[:50]}) try: # ... 业务逻辑 logger.info(f[NODE:extract_city] SUCCESS | city{city}) return state.copy(update{city: city}) except Exception as e: logger.error(f[NODE:extract_city] ERROR | {e}, exc_infoTrue) return state.copy(update{error_message: str(e)})三个黄金日志点Node入口记录输入关键字段如user_query前50字符Node出口记录输出关键字段如city值条件边路由前记录route_function的输入和返回值logger.debug(fROUTE:get_weather → {next_node})4.6 并发安全MemorySaver不是为高并发设计的MemorySaver是线程安全的但不是进程安全的。若用Gunicorn启多个Worker每个Worker有自己的内存副本thread_id状态无法共享。生产方案小规模100 QPS用RedisSaver配置redis://localhost:6379/0中大规模用PostgresSaver建表语句官方已提供关键业务自定义Saver将State存入分布式缓存如etcd4.7 提示词工程的隐藏战场Node间的数据契约很多团队把提示词写在Node函数里导致修改提示词需改代码无法A/B测试不同Node的提示词风格不一致如一个用“请”一个用“务必”无法集中管理、版本控制最佳实践提示词存为.txt文件按Node命名prompts/extract_city.txtNode中用jinja2模板渲染from jinja2 import Template prompt_template Template(open(prompts/extract_city.txt).read()) prompt prompt_template.render(user_querystate.user_query)4.8 图的可视化别信文档里的app.get_graph().draw_mermaid_png()LangGraph官方文档说draw_mermaid_png()可导出图片但实际需安装graphviz和pangoWindows上极易失败。更可靠的是用draw_mermaid_ascii()# 可视化图结构纯文本100%可用 print(app.get_graph().draw_mermaid_ascii())输出flowchart TD A[extract_city] -- B[get_weather] B -- C[generate_suggestion] B -- D[error_handler] C -- E[END] D -- E[END]4.9 错误传播的边界何时该捕获何时该抛出Node内异常处理原则网络/IO错误必须捕获返回error_message让图走到error_handler逻辑错误如state.city类型错误应抛出暴露给开发者而非吞掉LLM输出解析失败捕获记录error_message避免图中断4.10 性能瓶颈诊断用cProfile定位慢Node当整体延迟高时用Python内置分析器import cProfile import pstats pr cProfile.Profile() pr.enable() result app.invoke(initial_state) pr.disable() stats pstats.Stats(pr) stats.sort_stats(cumulative) stats.print_stats(10) # 打印最慢的10个函数重点关注weather_api_node和generate_suggestion_node的耗时。4.11 状态爆炸避免在State中存冗余字段曾有个Agent的State包含user_query,cleaned_query,normalized_query,query_embedding导致序列化体积翻4倍。State设计铁律只存必要字段下游Node真正需要的衍生字段如embedding应在Node内计算用完即弃大字段1KB存外部存储State中只存ID4.12 测试策略单元测试覆盖3类场景每个Node必须有3个测试用例正常流程输入有效state验证输出state字段正确边界输入state.user_query验证返回error_message异常模拟mock LLM/API抛异常验证错误处理逻辑用pytest和unittest.mock即可无需启动真实LLM。5. 后续演进从单体Agent到Agent集群的架构跃迁当你跑通第一个LangGraph Agent后真正的挑战才开始如何支撑100不同业务线的Agent如何让Agent之间协作如何做灰度发布我在实践中沉淀出三条必经之路5.1 Agent即服务AaaS抽象公共能力为可插拔模块把高频功能拆成独立微服务entity_extractor_service统一NER服务返回JSONtool_router_service基于意图识别的工具分发中心audit_logger_service所有Agent调用的审计日志网关LangGraph的Node可直接调用这些HTTP服务State字段保持一致实现能力复用。5.2 图嵌套用Subgraph管理复杂子流程当“合同审查”Agent本身包含“条款抽取→风险识别→法规匹配→摘要生成”4步时不应全写在一个Graph里。正确做法# contract_subgraph.py contract_workflow StateGraph(ContractState) # ... 添加contract相关nodes contract_app contract_workflow.compile() # 主graph中调用子图 def run_contract_review_node(state: AgentState) - AgentState: # 将state映射到ContractState contract_state ContractState(**state.dict()) result contract_app.invoke(contract_state) # 将result映射回AgentState return state.copy(update{contract_report: