尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

LangGraph:AI应用的状态机编排协议与工程实践

LangGraph:AI应用的状态机编排协议与工程实践 1. 为什么现在必须认真对待 LangGraph不是又一个“LLM 工具库”而是重构 AI 应用开发范式的底层协议LangGraph 这个词最近在技术社区里出现的频率已经明显超过了“LangChain”——但很多人点开文档的第一反应是“这不就是把 Chain 换了个名字加了点箭头”我去年底在给一家做智能客服中台的客户做架构评审时也这么以为。直到他们用 LangChain 写了三个月的多跳问答流程最后卡死在“用户中途改口、需要回溯状态、同时调用三个异步工具并等待其中两个返回后才决定第三个是否执行”这个场景上。团队写了 2700 行胶水代码测试覆盖率不到 43%每次加一个新分支逻辑就得重跑全部集成测试。后来我们用 LangGraph 重写核心路由模块最终交付版本只有 890 行状态可追踪、分支可调试、失败可重放——而且上线后第一个月就捕获了 14 个此前被胶水代码掩盖的业务逻辑冲突。这不是工具升级是开发范式迁移。LangGraph 的本质不是“让 LLM 调用更方便”而是把 AI 应用从线性脚本驱动转向状态机驱动。它强制你回答三个问题当前系统处于什么状态哪些动作可以合法触发触发后状态如何迁移这和前端 React 的状态管理、后端微服务的状态编排、甚至嵌入式系统的有限状态机FSM一脉相承——只是这次状态里存的是 LLM 的思考痕迹、工具调用结果、用户意图置信度而“动作”是模型生成、函数调用、人工审核或条件跳转。所以别把它当“LangChain 的增强版”。LangChain 是帮你把 prompt 拼得更稳、把向量库连得更牢LangGraph 是帮你把整个 AI 工作流画成一张可执行、可中断、可回滚、可监控的图。它的核心价值不在“能做什么”而在“做错时你能看清哪里错了”。当你看到控制台输出State: {messages: [...], tool_calls: [...], retry_count: 2, user_intent: cancel_order}而不是Error: list index out of range in _process_response()你就知道调试成本降了多少。关键词里反复出现的“langgraph 和 langchain 的区别”其实问错了重点。LangChain 是组件库Component LibraryLangGraph 是编排协议Orchestration Protocol。就像 jQuery 和 React 的关系——你依然可以用 jQuery 写 DOM 操作但 React 强制你用声明式状态描述 UI。LangGraph 不禁止你用 LangChain 的工具但它要求你把所有操作都注册为图中的节点并显式定义它们之间的边。这种约束恰恰是复杂 AI 应用可维护性的起点。提示如果你的项目还停留在“用户输入 → prompt 模板 → LLM 调用 → 输出解析”的单跳模式LangGraph 可能显得过度设计。但只要涉及多轮对话、条件分支、工具协同、人工干预或状态持久化它就不是“可选”而是“必需”。这不是技术炫技是工程负债的止损点。2. 图结构不是抽象概念从一个真实电商客服 Agent 看 LangGraph 的节点与边如何落地我们拿一个具体场景切入电商客服 Agent需支持“查订单 → 申请退货 → 选择退款方式 → 生成工单”全流程且要处理“用户突然说‘等等我要换货’”这类中断。用传统方式你会写一堆 if-elif-else 嵌套状态靠全局变量或闭包传递出错时只能靠日志拼凑执行路径。LangGraph 把这一切变成一张可读、可验、可调试的图。2.1 节点Node每个节点是一个纯函数只做一件事且必须返回新状态LangGraph 的节点不是类方法不是异步任务而是接收当前状态、返回新状态的纯函数。以“查订单”节点为例from typing import TypedDict, Annotated, Sequence from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver import operator class AgentState(TypedDict): messages: Annotated[Sequence[str], operator.add] # 自动合并消息列表 order_id: str user_phone: str current_step: str # lookup_order, request_refund, choose_method, create_ticket refund_method: str # original_payment, store_credit, bank_transfer has_confirmed: bool def lookup_order_node(state: AgentState) - AgentState: # 注意这里不直接调用数据库而是返回一个待执行的操作指令 # 真正的 DB 查询由图引擎在后续调度中执行解耦逻辑与执行 return { current_step: lookup_order, messages: [正在查询您的订单信息...], # 关键返回新状态不修改原 state order_id: extract_order_id_from_last_message(state[messages][-1]), user_phone: state.get(user_phone, ) }这个函数的关键特征无副作用不修改传入的state只返回新字段。LangGraph 内部用operator.add合并messages确保历史消息累积。职责单一只解析用户消息提取订单号不查库、不校验、不回复。查库是另一个节点的事。状态显式化current_step字段明确标识当前所处环节这是图能做条件跳转的基础。对比 LangChain 的RunnableLangChain 的Runnable可以链式调用但状态隐含在对象属性中LangGraph 的节点必须显式返回完整状态切片强制你思考“此刻系统需要记住什么”。2.2 边Edge边不是固定连接而是带条件的动态路由图的边决定了下一个节点是谁。LangGraph 支持两种边无条件边builder.add_edge(node_a, node_b)条件边builder.add_conditional_edges(node_a, route_function, {True: node_b, False: node_c})这才是 LangGraph 的灵魂。回到电商场景“查订单”后该去哪取决于订单是否存在、是否可退、用户是否已登录def route_after_lookup(state: AgentState) - str: # 检查订单是否存在状态中应有 order_id if not state.get(order_id): return ask_for_order_id # 节点名 # 模拟查库结果实际中此处会触发 DB 节点 order_status mock_db_query_order_status(state[order_id]) if order_status not_found: return order_not_found elif order_status delivered: return request_refund # 允许退货 elif order_status shipped: return cannot_refund_yet # 需等签收 else: return unexpected_status # 构建图时注册条件边 builder.add_conditional_edges( lookup_order_node, route_after_lookup, { ask_for_order_id: ask_for_order_id, order_not_found: order_not_found, request_refund: request_refund_node, cannot_refund_yet: cannot_refund_yet_node, unexpected_status: handle_error_node } )注意route_after_lookup函数返回的是字符串节点名不是节点对象。LangGraph 在运行时根据返回值动态选择下一跳。这意味着路由逻辑完全独立于节点实现可单独单元测试新增一种订单状态只需修改route_after_lookup的分支无需改动任何节点代码所有分支路径在图构建时就可静态分析IDE 能提示缺失的节点连接。2.3 状态State状态是图的唯一真相源必须精心设计LangGraph 的状态设计是项目成败的关键。很多初学者失败不是因为不会写节点而是状态设计太随意。我们的AgentState定义看似简单实则经过三次迭代第一版失败{messages: [], data: {}}→ 问题data里塞了订单详情、用户信息、临时 token后期无法区分哪些是业务数据、哪些是中间计算结果导致状态膨胀不可控。第二版改进按领域拆分{messages: [], order_context: {}, user_profile: {}, session_meta: {}}→ 问题字段太多节点间传递冗余数据且order_context里混着原始 API 返回和加工后的字段难以追踪数据血缘。第三版生产级class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add] # 业务实体只存 ID具体内容由专用节点加载懒加载 order_id: Optional[str] user_id: Optional[str] # 控制流明确标识当前步骤、重试次数、中断标记 current_step: Literal[ lookup_order, request_refund, choose_method, create_ticket, await_human_review ] retry_count: Annotated[int, operator.add] # 自动累加 # 用户显式意图由 NLU 节点识别用于覆盖默认流程 override_intent: Optional[Literal[cancel, switch_to_exchange, escalate]] # 工具调用记录用于审计和重放 tool_calls: Annotated[Sequence[ToolCall], operator.add]关键设计原则ID 优先内容后置状态里只存order_id不存订单详情。详情由load_order_details_node在需要时加载避免状态污染。控制流字段显式化current_step和override_intent让图引擎能精准决策而非靠解析messages猜意图。审计友好tool_calls记录每次工具调用的参数和时间戳故障时可精确重放。类型严格用Literal限定current_step取值IDE 能自动补全编译期捕获非法状态。注意状态设计没有银弹。我们曾因retry_count初始值设为0导致首次失败就触发重试后来改为Annotated[int, operator.add]并初始化为0确保每次失败自动 1。这种细节只有在真实压测中才会暴露。3. 从零搭建第一个可调试的 LangGraph Agent避开新手最常踩的五个深坑光看概念不够动手才是检验理解的唯一标准。下面带你手写一个极简但可调试的天气查询 Agent它能处理“北京天气”、“上海明天天气”、“深圳后天温度多少度”三种请求并在解析失败时自动重试。我会逐行解释每一步背后的工程考量以及那些文档里不会写的坑。3.1 环境准备版本锁定比什么都重要LangGraph 生态更新极快0.1.x 和 0.2.x 的 API 差异足以让你重写一半代码。我们锁定生产环境版本pip install langgraph0.2.45 langchain0.3.7 langchain-openai0.2.12 pydantic2.9.2为什么选这些版本langgraph0.2.45这是首个稳定支持checkpointer断点续跑和interrupt人工干预的版本0.2.40 之前MemorySaver有并发 bug。langchain0.3.7与 LangGraph 0.2.x 兼容性最佳0.4.x 开始引入RunnableConfig新参数旧代码需大量适配。pydantic2.9.2LangGraph 依赖 Pydantic v2但2.10.0的BaseModel.model_dump()默认exclude_unsetTrue会导致状态序列化丢失默认值引发图执行异常。提示永远在requirements.txt中锁定小版本号如0.2.45而非0.2.*。我见过团队因pip install langgraph自动升级到0.2.50导致add_conditional_edges接口签名变更CI 环境全量失败。3.2 定义状态与节点从“能跑通”到“可维护”的跨越from typing import TypedDict, Annotated, Optional, Literal, Sequence from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver import operator # 1. 精确的状态定义避坑点1不要用 dict class WeatherState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add] # 必须用 Annotated operator.add location: Optional[str] # 城市名 date: Optional[Literal[today, tomorrow, day_after_tomorrow]] # 避免字符串硬编码 temperature_unit: Literal[celsius, fahrenheit] celsius # 默认值必须是 Literal parse_attempts: Annotated[int, operator.add] 0 # 重试计数器初始值 0 # 2. 解析用户意图的节点避坑点2节点必须返回完整状态切片 def parse_intent_node(state: WeatherState) - WeatherState: last_msg state[messages][-1].content if state[messages] else # 简单规则匹配实际用 LLM 或 NLU 模型 if 北京 in last_msg: location Beijing elif 上海 in last_msg: location Shanghai elif 深圳 in last_msg: location Shenzhen else: location None if 明天 in last_msg: date tomorrow elif 后天 in last_msg: date day_after_tomorrow else: date today # 关键返回新状态不修改原 state return { location: location, date: date, parse_attempts: state.get(parse_attempts, 0) 1 # 显式累加 } # 3. 调用天气 API 的节点避坑点3API 调用必须包装为节点不能在路由函数里 def call_weather_api_node(state: WeatherState) - WeatherState: if not state.get(location): return {messages: [AIMessage(content请告诉我您想查询哪个城市的天气)]} # 模拟 API 调用实际中用 requests 或 langchain.tools weather_data mock_weather_api(state[location], state[date]) return { messages: [ AIMessage(contentf{state[location]} {state[date]} 天气{weather_data[condition]}{weather_data[temp]}°C) ] }避坑点详解坑1用dict代替TypedDictdict无法被 LangGraph 的类型检查器识别导致 IDE 无提示、运行时字段拼写错误难发现。TypedDict提供静态类型VS Code 能实时校验state[locaton]这种笔误。坑2节点内修改原 stateLangGraph 依赖不可变性做状态快照。若在parse_intent_node里写state[location] Beijing后续MemorySaver保存的状态会是脏数据断点续跑时出错。坑3在路由函数里调用 API条件边函数route_function必须是纯函数无 IO、无副作用。若在里面调用 API图执行将失去可预测性且无法被checkpointer捕获中间状态。3.3 构建图与添加边条件路由的正确写法def should_retry_parse(state: WeatherState) - bool: 判断是否需要重试解析避坑点4路由函数必须返回 bool 或 str不能返回 None # 如果 location 为空且重试次数 2则重试 return state.get(location) is None and state.get(parse_attempts, 0) 2 def route_after_parse(state: WeatherState) - str: 路由函数返回节点名字符串避坑点5返回值必须是已注册的节点名 if state.get(location): return call_weather_api_node else: return ask_location_node # 这个节点必须存在 # 构建图 builder StateGraph(WeatherState) # 注册节点 builder.add_node(parse_intent_node, parse_intent_node) builder.add_node(call_weather_api_node, call_weather_api_node) builder.add_node(ask_location_node, lambda s: {messages: [AIMessage(content请问您想查询哪个城市的天气)]}) # 添加边 builder.add_edge(START, parse_intent_node) # START 是内置节点 builder.add_conditional_edges( parse_intent_node, route_after_parse, # 先判断是否成功 { call_weather_api_node: call_weather_api_node, ask_location_node: ask_location_node } ) # 添加重试边当 should_retry_parse 为 True 时回到 parse_intent_node builder.add_conditional_edges( parse_intent_node, should_retry_parse, # 第二个条件边检查是否重试 {True: parse_intent_node, False: END} # 注意False 时直接结束 ) # 连接 API 节点到结束 builder.add_edge(call_weather_api_node, END) # 编译图避坑点6必须调用 compile() 才能运行 graph builder.compile(checkpointerMemorySaver())关键避坑点坑4路由函数返回NoneLangGraph 要求路由函数必须返回明确的布尔值或字符串。返回None会导致KeyError: None错误信息晦涩。务必用return True/False或return node_name。坑5返回未注册的节点名route_after_parse返回ask_location_node但若忘记调用builder.add_node(ask_location_node, ...)运行时报错ValueError: Node ask_location_node not found且堆栈不指向路由函数调试困难。坑6忘记调用compile()builder只是图定义graph builder.compile()才生成可执行对象。新手常直接调用builder.invoke(...)报错AttributeError: StateGraph object has no attribute invoke。3.4 运行与调试用stream()看清每一步发生了什么# 初始化内存检查点模拟用户会话 config {configurable: {thread_id: 123}} # 输入用户消息 initial_input {messages: [HumanMessage(content北京明天天气怎么样)]} # 流式执行观察每一步状态变化 for output in graph.stream(initial_input, config, stream_modevalues): print( 当前状态 ) print(f消息: {[m.content for m in output[messages]]}) print(f位置: {output.get(location)}) print(f日期: {output.get(date)}) print(f重试次数: {output.get(parse_attempts, 0)}) print() # 获取最终结果 final_state graph.invoke(initial_input, config) print(最终回复:, final_state[messages][-1].content)输出示例 当前状态 消息: [北京明天天气怎么样] 位置: None 日期: None 重试次数: 0 当前状态 消息: [北京明天天气怎么样] 位置: Beijing 日期: tomorrow 重试次数: 1 当前状态 消息: [北京明天天气怎么样, 北京 tomorrow 天气晴25°C] 位置: Beijing 日期: tomorrow 重试次数: 1调试价值你能清晰看到parse_attempts从 0 变 1确认重试逻辑生效messages列表逐步增长验证消息累积正确每次状态变更都对应一个节点执行故障时可精确定位到哪一步出错。实战心得在开发阶段永远用stream_modevalues而非invoke()。invoke()只返回最终结果stream()让你像看手术直播一样观察图的每一次心跳。我们曾用此方法发现MemorySaver在高并发下状态覆盖 bug若只用invoke()这个问题会潜伏数周。4. 生产级部署如何让 LangGraph Agent 在 Kubernetes 上稳定扛住每秒 200 请求写完本地可跑的 Demo 只是开始。真正的挑战在于如何让这个图在生产环境高可用、可观测、可扩缩我们为某金融客户部署的风控决策 Agent峰值 QPS 217平均延迟 89msSLA 99.99%。以下是经过压测验证的核心配置。4.1 状态持久化MemorySaver是玩具PostgresSaver才是生产标配MemorySaver仅适用于单机开发。生产必须用持久化检查点Checkpoint否则 Pod 重启后会话状态全丢。from langgraph.checkpoint.postgres import PostgresSaver import asyncpg # 初始化 PostgreSQL 连接池使用 asyncpg非 SQLAlchemy async def init_checkpointer(): connection_string postgresql://user:passlocalhost:5432/langgraph_db pool await asyncpg.create_pool(connection_string) # 创建表LangGraph 会自动执行 DDL saver PostgresSaver(pool) await saver.setup() # 必须调用创建 checkpoint 表 return saver # 在 FastAPI 启动时初始化 app.on_event(startup) async def startup_event(): app.state.checkpointer await init_checkpointer()PostgreSQL 表结构关键字段字段类型说明thread_idVARCHAR(255)会话唯一 ID作为主键checkpointJSONB序列化后的状态快照压缩存储parent_tsTIMESTAMP父检查点时间戳用于状态回溯pending_sendsJSONB待发送的工具调用支持断点续调性能优化点JSONB类型支持 GIN 索引WHERE thread_id ?查询毫秒级响应checkpoint字段启用pglz压缩10KB 状态压缩后仅 2.3KB我们为thread_id添加唯一索引并设置checkpoint字段 TTL 为 7 天自动清理。注意不要用 Redis 做检查点。Redis 的SET操作在高并发下易发生覆盖LangGraph 的PostgresSaver通过INSERT ... ON CONFLICT DO UPDATE保证原子性这是金融场景的底线。4.2 并发控制GIL 不是瓶颈LLM API 才是Python 的 GIL 在 LangGraph 场景下影响极小——因为真正耗时的是 LLM API 调用网络 IO而非 CPU 计算。瓶颈在于OpenAI API 的 rate limit如gpt-4-turbo10K TPM自建 LLM 服务的 GPU 显存占用数据库连接池耗尽。解决方案LLM 限流用langchain_community.utils.math的RateLimiter包装 LLMfrom langchain_community.utils.math import RateLimiter from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4-turbo) rate_limited_llm RateLimiter( llm, max_calls100, # 每分钟最多 100 次 period60 )数据库连接池asyncpg连接池大小设为min(20, CPU_CORES * 4)避免连接争抢。图执行并发LangGraph 默认单线程执行图。若需并行执行多个会话用asyncio.gather# 同时处理 10 个用户请求 tasks [ graph.ainvoke({messages: [HumanMessage(contentmsg)]}, config) for msg in user_messages[:10] ] results await asyncio.gather(*tasks)4.3 可观测性埋点不是可选是 SLO 的基石没有监控的 LangGraph 就是黑盒。我们在每个节点入口/出口打点import time from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter # 初始化 OpenTelemetry provider TracerProvider() exporter OTLPSpanExporter(endpointhttp://otel-collector:4318/v1/traces) provider.add_span_processor(BatchSpanProcessor(exporter)) # 节点装饰器 def instrument_node(node_name: str): def decorator(func): def wrapper(state: WeatherState): tracer trace.get_tracer(__name__) with tracer.start_as_current_span(f{node_name}.execute) as span: span.set_attribute(state.keys, list(state.keys())) start_time time.time() try: result func(state) span.set_attribute(status, success) return result except Exception as e: span.set_attribute(status, error) span.record_exception(e) raise finally: span.set_attribute(duration_ms, (time.time() - start_time) * 1000) return wrapper return decorator instrument_node(parse_intent_node) def parse_intent_node(state: WeatherState) - WeatherState: # 原逻辑不变 ...关键指标看板langgraph_node_duration_seconds_bucket各节点 P95 延迟定位慢节点langgraph_state_size_bytes状态序列化后大小预警状态膨胀langgraph_checkpoint_errors_total检查点失败次数数据库故障早期信号langgraph_tool_call_failures_total工具调用失败率区分 LLM 错误 vs 服务错误。4.4 滚动升级与灰度如何零停机发布新图版本LangGraph 的图是代码升级即发版。我们采用双版本并行策略# v1_graph.py旧版 v1_graph builder_v1.compile(checkpointerapp.state.checkpointer) # v2_graph.py新版新增汇率查询节点 v2_builder StateGraph(WeatherState) v2_builder.add_node(parse_intent_node, v2_parse_intent_node) # 新解析逻辑 v2_builder.add_node(get_exchange_rate_node, get_exchange_rate_node) # 新节点 ... v2_graph v2_builder.compile(checkpointerapp.state.checkpointer) # FastAPI 路由根据 header 路由 app.post(/chat) async def chat_endpoint(request: Request): # 读取 header X-Graph-Version version request.headers.get(X-Graph-Version, v1) if version v2: graph v2_graph else: graph v1_graph return await graph.ainvoke(...)灰度发布流程新版图部署到 5% 流量通过 Istio VirtualService监控v2_graph的langgraph_node_duration_seconds是否显著升高对比v1和v2的langgraph_tool_call_failures_total确认新节点稳定性72 小时无异常后切流至 100%。经验教训我们曾因新图中get_exchange_rate_node的超时设置为 30s旧版是 5s导致整体延迟飙升。可观测性指标在灰度期就捕获了 P95 延迟从 89ms 升至 3.2s避免了全量事故。5. LangGraph 与 LangChain 的共生关系何时用谁怎么组合搜索热词里高频出现“langchain 和 langgraph 的区别”但现实项目中它们不是二选一而是分工协作。LangChain 是“零件库”LangGraph 是“装配线”。理解它们的边界才能避免重复造轮子。5.1 LangChain 的不可替代性它负责“怎么做”LangGraph 负责“什么时候做”LangChain 提供了 LangGraph 无法替代的基础设施文档加载与切分UnstructuredPDFLoader、RecursiveCharacterTextSplitter向量存储与检索Chroma、PGVector的封装Retriever抽象工具集成DuckDuckGoSearchRun、WikipediaQueryRun等开箱即用工具LLM 抽象层统一ChatOpenAI、ChatAnthropic、Ollama的调用接口。LangGraph 则负责流程编排决定“先查知识库再调用工具最后生成回复”状态管理在多轮中记住用户偏好、历史工具调用结果错误恢复当工具调用失败自动降级到备用方案。典型组合模式# LangChain 工具开箱即用 from langchain_community.tools import DuckDuckGoSearchRun search_tool DuckDuckGoSearchRun() # LangGraph 节点中调用 LangChain 工具 def search_node(state: AgentState) - AgentState: query state[search_query] # LangChain 工具在此执行 results search_tool.invoke(query) return {search_results: results} # LangGraph 图中注册该节点 builder.add_node(search_node, search_node)5.2 避免常见组合陷阱三个必须警惕的反模式反模式1在 LangGraph 节点里写 LangChain Chain# ❌ 错误把 LangChain Chain 当节点破坏状态可见性 def bad_node(state: State): chain prompt | llm | parser # LangChain Chain result chain.invoke({input: state[messages][-1].content}) return {answer: result} # ✅ 正确拆解 Chain 为独立节点暴露中间状态 builder.add_node(format_prompt_node, format_prompt) # 只做 prompt 格式化 builder.add_node(call_llm_node, call_llm) # 只做 LLM 调用 builder.add_node(parse_response_node, parse_response) # 只做解析反模式2用 LangChain Memory 替代 LangGraph StateLangChain 的ConversationBufferMemory是为单轮 Chain 设计的无法支持 LangGraph 的多分支、状态回溯。强行混合会导致Memory中的消息与State中的messages不一致checkpointer无法保存Memory状态断点续跑失效。反模式3认为 LangGraph 能替代 LangChain 的所有功能LangGraph 没有内置 PDF 解析、没有向量检索器、没有工具市场。试图自己实现这些等于放弃 LangChain 数年的生态积累。正确的做法是LangChain 做“能力”LangGraph 做“编排”。5.3 未来演进LangGraph 正在吞噬 LangChain 的边界LangGraph 团队已在 0.2.x 版本中将部分 LangChain 功能“图化”langgraph.prebuilt.tool_node将 LangChain 工具自动包装为图节点langgraph.prebuilt.chat_agent_executor基于 LangGraph 的通用聊天 Agent 框架内部已集成Retriever、Tool等 LangChain 组件langgraph.checkpoint.sqlite轻量级检查点适合边缘设备。这意味着LangChain 的角色正从“全能框架”转向“能力提供者”而 LangGraph 成为“统一编排平面”。对于新项目建议新项目以 LangGraph 为基座按需引入 LangChain 的tools、retrievers、llms模块老项目迁移逐步将 LangChain Chain 拆解为 LangGraph 节点保留原有工具和模型只重构流程。最后分享一个真实案例我们帮一家教育科技公司迁移其“AI 习题推荐”系统。原 LangChain 实现有 3 个 Chain知识点识别、难度评估、题目生成耦合严重。迁移到 LangGraph 后将每个 Chain 拆为 2-3 个节点加入validate_knowledge_point、fallback_to_easy_questions等新节点不仅支持了“学生说‘太难了’就自动降级”的需求还将平均响应时间从 2.1s 降至 1.3s——因为状态复用减少了重复的向量检索。我在实际项目中发现LangGraph 的学习曲线陡峭期大约在 3 天第一天困惑于“为什么非要写状态”第二天纠结于“边该怎么连”第三天突然顿悟“原来状态就是我的业务逻辑地图”。一旦跨过这个门槛你写的不再是 AI 应用而是可演进、可审计、可交付的 AI 业务流程。这或许就是它值得你投入时间的真正原因。
返回列表