1. 从LangChain到DeepAgents为什么我们需要更复杂的智能体框架如果你最近在折腾AI应用开发尤其是想搞点能自主决策、执行复杂任务的“智能体”那你肯定绕不开LangChain。LangChain的Agent模块可以说是让LLM大语言模型从“聊天机器人”迈向“任务执行者”的第一步。它通过一个简单的“思考-行动-观察”循环让LLM能够调用你定义的工具比如搜索、计算、API去完成一个目标。这听起来很酷对吧我刚开始用的时候也觉得打开了新世界的大门用几行代码就能让GPT-4去查天气、算数学题成就感满满。但很快现实就给了我一记重拳。当我试图构建一个稍微复杂点的流程比如一个能根据用户需求自动调研、撰写报告、并发送邮件的智能体时用原生的LangChain Agent就开始捉襟见肘了。问题接踵而至状态管理混乱一个多轮对话下来上下文Context里塞满了各种中间结果LLM自己都搞不清哪是哪流程控制僵硬想实现“如果A条件成立就执行B否则执行C”这种分支逻辑得写一堆丑陋的条件判断和回调函数错误处理更是噩梦工具调用失败后智能体要么卡死要么开始胡言乱语。更别提多人协作、长期记忆、子任务分解这些高级需求了。你会发现LangChain Agent更像是一个精巧的“单步执行器”而不是一个能驾驭复杂工作流的“大脑”。这就是DeepAgents或者说以LangGraph为代表的下一代智能体框架登场的背景。它们要解决的正是LangChain Agent在复杂场景下的“力不从心”。DeepAgents不是一个具体的、官方的库名它更像是一个概念指的是那些构建在LangChain等基础组件之上采用更先进架构如图计算、状态机来实现深度、复杂、可编排智能体的方法和框架。而LangGraph作为LangChain官方推出的“亲儿子”是目前实现这一理念最成熟、最受关注的选择。它用“图”Graph的思维来建模智能体的工作流节点Node代表一个执行步骤如调用LLM、运行工具边Edge代表步骤之间的流转条件。这听起来有点抽象但简单理解就是它把智能体从一个线性的“流水线工人”升级成了一个有明确分工和协作规则的“项目团队”每个成员节点只负责自己那部分通过清晰的规则边来传递任务和结果。所以当你看到“langchain deepagents-深度研究实战”这个标题时它指向的绝不仅仅是学会调用几个API。它是一场思维模式的升级从编写线性的、脆弱的脚本到设计健壮的、可扩展的、可视化的智能体工作流。接下来我会结合我近期的实战项目带你深入LangGraph的核心看看我们是如何用它来构建一个真正能打的“深度智能体”的。2. LangGraph核心架构拆解State、Node与Edge是如何协同工作的要玩转LangGraph你必须吃透它的三个核心概念State状态、Node节点和Edge边。这构成了它整个运行时的骨架。很多人一开始会被这些术语吓到其实我们可以用一个非常生活化的场景来类比一个智能体就像一家餐厅的后厨。State状态就是后厨里那张唯一的、共享的订单白板。上面记录了当前所有订单的信息客人点了什么菜用户输入哪些菜正在做执行中任务哪些菜已经做好了已完成的结果以及后厨现有的食材上下文记忆。这个白板State是一个字典Dict结构所有厨师Node都只能读写这块白板来协作而不是互相私下传递纸条。这保证了信息的一致性和可追溯性。Node节点就是后厨里的各个工位或厨师。比如有专门看订单的“理解员”LLM调用节点有负责切菜的“预处理员”工具调用节点有掌勺的“主厨”另一个LLM节点还有负责摆盘的“装饰员”结果格式化节点。每个节点都是一个Python函数它的唯一职责就是从State白板上读取它需要的信息进行处理然后把结果写回State白板。它很“单纯”只关心自己的输入和输出。Edge边就是连接工位之间的传送带和规则。它决定了订单State的流向。比如规则可以是“如果订单里有‘牛排’就传给主厨节点如果是‘沙拉’就传给预处理员节点”。或者更简单“无论上一步做了什么下一步都传给理解员节点”。这些规则就是边它们可以是条件判断conditional_edge也可以是无条件流转Edge。让我们用代码来具象化这个餐厅后厨。假设我们要构建一个智能体它能理解用户问题并决定是直接回答简单问题还是去网上搜索后回答复杂问题。首先我们定义State。在LangGraph中我们通常用一个TypedDict来明确定义State里都有哪些“字段”。from typing import TypedDict, Annotated from langgraph.graph.message import add_messages import operator class State(TypedDict): # 消息历史这是一个由LangGraph管理的特殊列表用于记录对话 messages: Annotated[list, add_messages] # 用户当前的问题 user_query: str # 判断结果是‘direct_answer’还是‘need_search’ next_step: str # 从网络搜索到的内容如果需要的话 search_results: str # 最终给用户的答案 final_answer: str这里Annotated[list, add_messages]是LangGraph提供的一个语法糖它自动帮我们管理消息列表的追加非常方便。其他字段就是普通的字符串。接下来我们创建节点。每个节点都是一个函数接收一个State字典更新它然后返回更新后的字典。def understand_query(state: State) - State: 节点1理解用户问题并判断下一步动作 from langchain_openai import ChatOpenAI llm ChatOpenAI(model“gpt-4”) # 从State中获取用户问题 query state[“user_query”] # 让LLM判断问题类型 prompt f””” 请判断以下用户问题是否需要联网搜索才能准确回答。 问题{query} 如果问题涉及实时信息、最新事件、特定数据查询请回答‘need_search’。 如果问题是一般性知识、解释概念、无需最新信息请回答‘direct_answer’。 只输出‘need_search’或‘direct_answer’。 “”” decision llm.invoke(prompt).content.strip() # 将判断结果写回State state[“next_step”] decision return state def direct_answer_node(state: State) - State: 节点2直接回答问题 from langchain_openai import ChatOpenAI llm ChatOpenAI(model“gpt-4”) query state[“user_query”] prompt f”请回答以下问题{query}” answer llm.invoke(prompt).content state[“final_answer”] answer return state def search_and_answer_node(state: State) - State: 节点3搜索后回答问题 # 假设我们有一个搜索工具 from my_tools import web_search query state[“user_query”] # 1. 执行搜索 results web_search(query) state[“search_results”] results # 2. 基于搜索结果生成答案 from langchain_openai import ChatOpenAI llm ChatOpenAI(model“gpt-4”) prompt f””” 基于以下搜索结果为用户的问题生成一个全面、准确的答案。 用户问题{query} 搜索结果{results} “”” answer llm.invoke(prompt).content state[“final_answer”] answer return state现在我们有三个厨师节点了。怎么把他们组织起来呢这就需要定义边构建图Graph。from langgraph.graph import StateGraph, END # 创建一个图并指定它处理的是我们定义的State类型 workflow StateGraph(State) # 添加节点 workflow.add_node(“understand”, understand_query) workflow.add_node(“direct_answer”, direct_answer_node) workflow.add_node(“search_answer”, search_and_answer_node) # 设置入口点所有流程都从‘understand’节点开始 workflow.set_entry_point(“understand”) # 添加条件边从‘understand’节点出来后根据State里的‘next_step’字段决定去哪 workflow.add_conditional_edges( “understand” # 这是一个路由函数它读取State返回下一个节点的名字 lambda state: state[“next_step”] # 一个映射将路由函数返回的值‘direct_answer’或‘need_search’映射到对应的节点名 { “direct_answer”: “direct_answer” “need_search”: “search_answer” } ) # 添加普通边从‘direct_answer’和‘search_answer’节点出来后都直接结束 workflow.add_edge(“direct_answer”, END) workflow.add_edge(“search_answer”, END) # 编译图得到可执行的对象 app workflow.compile()至此一个最简单的LangGraph智能体就构建完成了。它的执行流程是输入用户问题 -understand节点判断 - 根据判断结果路由到direct_answer或search_answer节点 - 生成答案 - 结束。整个过程中State就像那个流动的订单白板承载了所有中间数据和最终结果。注意在实际开发中LLM调用和工具调用如web_search往往是性能瓶颈和错误来源。务必为它们添加完善的超时timeout和重试retry机制。例如使用tenacity库进行装饰或者利用LangChain内置的RunnableRetry。否则网络抖动或API限流会导致整个智能体流程卡死。3. 实战构建一个具备长期记忆与工具调用的研究型智能体理解了基础架构我们来点真格的。我将还原一个近期为内部知识库构建的“研究型智能体”项目。它的目标是用户输入一个复杂的研究主题例如“对比LangGraph和CrewAI在复杂工作流设计上的优劣”智能体能够自动进行多轮、深度的资料搜集、信息整理与对比分析并最终生成一份结构化的报告。这远非一次搜索就能解决它需要长期记忆来避免重复搜索需要灵活的工具调用来获取不同维度的信息还需要子图来管理复杂的分析子任务。3.1 状态设计为复杂任务预留空间对于复杂智能体State的设计是重中之重它决定了系统的扩展性和数据流是否清晰。我们为研究型智能体设计了如下Statefrom typing import TypedDict, Annotated, List, Optional from langgraph.graph.message import add_messages import datetime class ResearchState(TypedDict): # 核心对话与指令 messages: Annotated[list, add_messages] research_topic: str # 研究主题 user_requirements: str # 用户的额外要求如“侧重工程实践” # 执行过程 current_subtask: Optional[str] # 当前正在执行的子任务名 subtask_results: Annotated[List[str], operator.add] # 所有子任务的结果列表 # 工具与记忆 search_queries: Annotated[List[str], operator.add] # 历史搜索查询用于去重 visited_urls: Annotated[List[str], operator.add] # 已访问的URL避免重复抓取 collected_data: Annotated[List[dict], operator.add] # 收集到的原始数据片段列表 # 输出 outline: Optional[str] # 报告大纲 report_content: Optional[str] # 最终报告内容 status: str # 状态如 ‘planning’ ‘collecting’ ‘analyzing’ ‘writing’ ‘finished’这里有几个关键点Annotated[List[str], operator.add]这是LangGraph的“缩减器”Reducer语法。它告诉框架当多个节点并行或先后修改这个字段时不是覆盖而是用operator.add即列表的操作来合并结果。这对于收集碎片化信息如多次搜索的结果至关重要。current_subtask和status这是控制流程的“指针”和“信号灯”。通过它们我们可以让智能体在不同模式间切换。分离的collected_data和report_content原始数据和分析后的报告分开存储符合数据处理流程也便于调试和溯源。3.2 工具集成与子图实现模块化分工智能体需要调用多种工具。我们使用LangChain的tool装饰器来定义from langchain.tools import tool from duckduckgo_search import DDGS import asyncio tool async def web_search_tool(query: str, max_results: int 5) - str: 使用DuckDuckGo进行网络搜索。输入应为搜索关键词。 try: with DDGS() as ddgs: results [] # 使用异步避免阻塞注意DDGS可能需要同步调用这里用线程池包装 loop asyncio.get_event_loop() r await loop.run_in_executor(None, lambda: [r for r in ddgs.text(query, max_resultsmax_results)]) for res in r[:max_results]: results.append(f”标题{res[‘title’]}\n摘要{res[‘body’]}\n链接{res[‘href’]}\n{‘-’*40}”) return “\n”.join(results) if results else “未找到相关结果。” except Exception as e: return f”搜索过程出错{str(e)}” tool def summarize_text_tool(long_text: str) - str: 总结长文本的核心观点。 # 这里可以调用一个专门的总结LLM或者使用gpt-3.5-turbo等 from langchain_openai import ChatOpenAI llm ChatOpenAI(model“gpt-3.5-turbo-16k” temperature0) prompt f”请用不超过200字总结以下文本的核心内容和观点\n\n{long_text}” return llm.invoke(prompt).content # 将工具绑定到LLM from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor llm ChatOpenAI(model“gpt-4” temperature0.1) tools [web_search_tool, summarize_text_tool] agent create_tool_calling_agent(llm, tools, prompt) # prompt需自定义 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue)现在我们有了执行单一任务的“工人”agent_executor。但研究任务需要规划、收集、分析、写作等多个阶段。我们可以为每个阶段创建一个子图Subgraph。子图本身也是一个完整的图可以被主图调用这实现了功能的模块化和复用。例如创建一个资料收集子图from langgraph.graph import StateGraph, START, END def create_collection_subgraph(): “””创建资料收集子图负责执行一轮搜索、去重、保存结果。”“” from typing import TypedDict class CollectionState(TypedDict): topic: str search_queries: list visited_urls: list collected_data: list def plan_query(state: CollectionState): “””根据主题规划出几个搜索关键词。”“” # 调用LLM生成3-5个相关搜索词 planned_queries [f”{state[‘topic’]} 最新发展” f”{state[‘topic’]} 优缺点” f”{state[‘topic’]} 实战案例”] state[‘search_queries’].extend(planned_queries) return state def execute_search(state: CollectionState): “””执行一个尚未执行的搜索查询。”“” # 简化逻辑取第一个未执行的查询 for query in state[‘search_queries’]: if query not in state.get(‘_executed_queries’ []): results web_search_tool.invoke(query) # 解析结果提取链接和内容去重后存入collected_data # … (具体解析去重逻辑) state[‘collected_data’].append({‘query’: query ‘results’: parsed_results}) state.setdefault(‘_executed_queries’ []).append(query) break return state def check_completion(state: CollectionState): “””检查是否收集了足够的数据。”“” # 简单策略如果已执行查询数3或者collected_data条目数5则认为完成 if len(state.get(‘collected_data’ [])) 5: return “enough_data” elif len(state.get(‘_executed_queries’ [])) 3: return “enough_queries” else: return “continue_search” # 构建子图 workflow StateGraph(CollectionState) workflow.add_node(“plan” plan_query) workflow.add_node(“search” execute_search) workflow.add_edge(START “plan”) workflow.add_edge(“plan” “search”) workflow.add_conditional_edges( “search” check_completion {“enough_data”: END “enough_queries”: END “continue_search”: “search”} ) return workflow.compile()在主图中我们可以像调用一个超级节点一样调用这个子图。这使主图逻辑变得非常清晰规划阶段 - 调用收集子图 - 分析阶段 - 调用写作子图。3.3 编译与运行可视化与调试将所有的节点和子图通过边连接起来就形成了主图。使用workflow.compile()进行编译。# 假设我们已经定义了 plan_node analysis_node write_node 等主图节点 main_workflow StateGraph(ResearchState) main_workflow.add_node(“plan” plan_node) main_workflow.add_node(“collect” collection_subgraph) # 这里传入的是编译好的子图 main_workflow.add_node(“analyze” analysis_node) main_workflow.add_node(“write” write_node) main_workflow.set_entry_point(“plan”) main_workflow.add_edge(“plan” “collect”) main_workflow.add_edge(“collect” “analyze”) main_workflow.add_edge(“analyze” “write”) main_workflow.add_edge(“write” END) research_agent main_workflow.compile()LangGraph一个强大的功能是可视化。你可以直接将图导出为PNG。from IPython.display import Image display try: display(Image(research_agent.get_graph().draw_mermaid_png())) except: # 如果环境不支持可以输出Mermaid文本到支持渲染的编辑器查看 print(research_agent.get_graph().draw_mermaid())这张图是你智能体的“架构图”对于理解流程、排查问题有巨大帮助。运行智能体非常简单initial_state { “messages”: [] “research_topic”: “对比LangGraph和CrewAI在复杂工作流设计上的优劣” “user_requirements”: “请侧重从开发体验、状态管理、错误处理三个方面进行对比。” “status”: “planning” “search_queries”: [] “visited_urls”: [] “collected_data”: [] “subtask_results”: [] } # 以流式事件的方式运行可以实时看到执行到了哪个节点 final_state None for event in research_agent.stream(initial_state stream_mode“values”): node_name list(event.keys())[0] print(f”\n 进入节点[{node_name}] ) # 可以在这里打印或记录state的特定字段 if node_name “__end__”: final_state event[node_name] print(“\n 任务执行完毕 ”) print(f”最终报告状态{final_state.get(‘status’)}”) # 可以保存或输出final_state[‘report_content’]在流式运行中你会看到智能体按照你设计的蓝图一步步推进规划搜索词、触发收集子图进行多轮搜索、分析收集到的数据、最后撰写报告。整个过程状态清晰可控性强。4. 性能调优与生产级部署的避坑指南构建出可运行的智能体只是第一步。要让它在真实环境中稳定、高效地工作还有一大堆“坑”要填。以下是我在项目中积累的关键经验。4.1 状态管理的性能陷阱与优化State是共享内存频繁读写大对象比如很长的collected_data列表会影响性能尤其是在多线程或异步环境下。优化策略分而治之不要把所有数据都塞进一个State。对于大型、独立的数据块如一篇完整的爬取文章可以考虑存储一个引用如ID或路径而将数据本身存入外部数据库如Redis、PostgreSQL或向量数据库如Chroma、Weaviate。State里只保留元数据和索引。使用Annotated进行高效更新如前所述对于列表类数据使用Annotated[List, operator.add]让LangGraph在后台进行合并比在节点函数中手动state[‘list’].append(item); return state更高效、更安全。定期清理对于长期运行的智能体如聊天机器人State中的messages历史会无限增长最终导致LLM上下文超长响应变慢且成本激增。必须在图中设计一个“清理节点”定期将旧消息总结压缩Summary或转移到长期记忆存储中。4.2 工具调用的稳定性保障工具调用尤其是网络请求是智能体最脆弱的环节。必须做到超时设置为每一个工具调用、每一个LLM请求设置明确的超时。使用asyncio.wait_for或httpx.Timeout。import asyncio from langchain_core.runnables import RunnableConfig async def safe_tool_call(tool, input_args): try: # 设置5秒超时 result await asyncio.wait_for(tool.ainvoke(input_args) timeout5.0) return result except asyncio.TimeoutError: return “工具调用超时请重试或跳过此步骤。”指数退避重试对于可能因瞬时网络问题或API限流导致的失败实施重试机制。LangChain的RunnableRetry可以很方便地包装你的工具或LLM。from langchain_core.runnables import RunnableRetry retry_policy RunnableRetry( retry_if_exception_type(Exception) # 重试所有异常 wait_exponential_jitterTrue # 指数退避抖动 max_attempt_number3 ) robust_llm retry_policy | llm # 将重试策略应用到LLM上优雅降级当核心工具如搜索失败时智能体不应该崩溃。可以在State中设计一个fallback_mode标志当工具连续失败时切换到一个降级模式例如仅基于已有知识库回答并告知用户能力受限。4.3 错误处理与流程中断在复杂的图中一个节点的失败不应导致整个流程崩溃。LangGraph提供了interrupt机制。节点级Try-Catch在每个节点函数内部进行细致的异常捕获并将错误信息写入State例如state[‘last_error’] str(e)然后让流程根据错误信息路由到专门的“错误处理节点”。使用StateGraph的interrupt你可以定义一些“中断”点。当state[‘status’] ‘error’时通过条件边将流程导向一个human_review_node人工审核节点或者一个retry_node重试节点。这给了你从错误中恢复或引入人工干预的机会。超时中断整个图对于长时间运行的任务可以在调用app.invoke()或app.stream()时设置超时或者在外层用asyncio.wait_for包装。4.4 记忆Memory的持久化与加载智能体的价值在于其连续性。你需要将一次运行结束后的State保存下来并在下次同会话中加载。LangGraph的Checkpointer抽象就是干这个的。from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 # 1. 创建一个SQLite检查点存储器 conn sqlite3.connect(“checkpoints.db” check_same_threadFalse) checkpointer SqliteSaver(conn) # 2. 在编译图时传入checkpointer app workflow.compile(checkpointercheckpointer) # 3. 运行智能体时指定一个thread_id会话ID config {“configurable”: {“thread_id”: “user_123_session_1”}} initial_state {…} result app.invoke(initial_state configconfig) # 4. 下次运行时可以从检查点恢复状态 # 通过get_state获取最新状态或直接invoke时会自动从上次中断处继续 latest_state checkpointer.get_tuple(config).next使用检查点你可以实现智能体的“暂停/继续”以及跨对话的长期记忆。这对于构建客服机器人、个人助理等场景至关重要。4.5 监控与可观测性在生产环境你必须要知道你的智能体在干什么、性能如何。结构化日志在每个节点的入口和出口记录关键信息节点名、输入State摘要、输出State摘要、耗时、错误。使用structlog或loggingJSON格式输出方便接入ELK或Datadog。追踪Tracing利用LangSmithLangChain官方平台或OpenTelemetry进行深度追踪。这能让你清晰地看到每一次LLM调用、工具调用的输入输出、延迟和token消耗是进行成本分析和性能优化的黄金标准。自定义监控指标在State中埋点统计工具调用成功率、各节点平均耗时、最终输出质量评分可通过另一个LLM评估等并暴露给Prometheus等监控系统。5. 超越基础探索LangGraph的高级模式与生态当你掌握了上述核心内容后可以探索一些更高级的模式让智能体更强大。5.1 多智能体协作与竞争LangGraph的图模型天然支持多智能体。你可以创建多个“智能体节点”每个节点背后是一个独立的LLM甚至可以是不同模型的LLM比如一个GPT-4负责创意一个Claude负责严谨分析并让它们通过共享的State进行通信和协作。例如一个“辩论图”可以让两个智能体节点就一个议题进行多轮辩论将各自观点写入State最后由一个“裁判”节点进行总结。这通过条件边和循环很容易实现。5.2 与FastAPI集成构建API服务将编译好的app即你的智能体封装成FastAPI应用是提供服务的标准做法。from fastapi import FastAPI, HTTPException from pydantic import BaseModel import uvicorn app_fastapi FastAPI(title“Research Agent API”) class ResearchRequest(BaseModel): topic: str requirements: str “” app_fastapi.post(“/research”) async def start_research(request: ResearchRequest): try: initial_state { “research_topic”: request.topic “user_requirements”: request.requirements “status”: “planning” # … 其他初始字段 } config {“configurable”: {“thread_id”: f”req_{uuid.uuid4().hex}”}} # 异步流式响应Server-Sent Events async def event_generator(): async for event in research_agent.astream(initial_state configconfig stream_mode“values”): node list(event.keys())[0] if node ! “__end__”: yield {“event”: “node_update” “node”: node “data”: event[node].get(‘status’)} else: final_report event[node].get(‘report_content’) yield {“event”: “complete” “report”: final_report} return EventSourceResponse(event_generator()) except Exception as e: raise HTTPException(status_code500 detailstr(e))这样前端就可以通过SSEServer-Sent Events实时看到智能体的执行进度并在完成后获取报告。5.3 与Dify、Coze等平台的区别与选型现在市面上有很多智能体平台如Dify、Coze、阿里的魔搭ModelScope Agent。它们和基于LangGraph自研有什么区别Dify/Coze等平台优点是开箱即用可视化编排集成了大量预置工具、知识库和模型非常适合快速原型验证、无代码/低代码构建简单智能体应用。你不需要关心部署和运维。缺点是灵活性受限深度定制复杂比如实现我们上面那种复杂的状态管理和子图且可能产生平台绑定和成本。LangGraph自研优点是完全自主可控灵活性极高可以实现任何你能想象到的复杂逻辑和架构。能与现有技术栈深度集成便于进行性能优化和成本控制。缺点是门槛高需要较强的工程能力并且要自行处理部署、监控、运维等所有生产问题。选型建议如果你的需求是快速做一个概念验证PoC或者智能体逻辑相对标准聊天、简单问答、基于文档的检索优先考虑Dify等平台。如果你的智能体需要复杂的业务逻辑、独特的状态流、与内部系统的深度集成、或对性能和成本有极致要求那么LangGraph自研是更合适的选择。它更像是一个“智能体框架的框架”为你提供了构建强大引擎的所有零件。从我个人的实战经验来看LangGraph的学习曲线确实比直接使用LangChain Agent要陡峭但带来的能力提升是数量级的。它迫使你以更工程化、更结构化的方式思考智能体这种思维模式的价值远超学会一个工具本身。当你成功用它搭建起第一个能稳定处理复杂任务的智能体时那种对流程的掌控感和成就感是使用现成平台无法比拟的。 LangGraph深度实战：从状态管理到复杂智能体架构设计
1. 从LangChain到DeepAgents为什么我们需要更复杂的智能体框架如果你最近在折腾AI应用开发尤其是想搞点能自主决策、执行复杂任务的“智能体”那你肯定绕不开LangChain。LangChain的Agent模块可以说是让LLM大语言模型从“聊天机器人”迈向“任务执行者”的第一步。它通过一个简单的“思考-行动-观察”循环让LLM能够调用你定义的工具比如搜索、计算、API去完成一个目标。这听起来很酷对吧我刚开始用的时候也觉得打开了新世界的大门用几行代码就能让GPT-4去查天气、算数学题成就感满满。但很快现实就给了我一记重拳。当我试图构建一个稍微复杂点的流程比如一个能根据用户需求自动调研、撰写报告、并发送邮件的智能体时用原生的LangChain Agent就开始捉襟见肘了。问题接踵而至状态管理混乱一个多轮对话下来上下文Context里塞满了各种中间结果LLM自己都搞不清哪是哪流程控制僵硬想实现“如果A条件成立就执行B否则执行C”这种分支逻辑得写一堆丑陋的条件判断和回调函数错误处理更是噩梦工具调用失败后智能体要么卡死要么开始胡言乱语。更别提多人协作、长期记忆、子任务分解这些高级需求了。你会发现LangChain Agent更像是一个精巧的“单步执行器”而不是一个能驾驭复杂工作流的“大脑”。这就是DeepAgents或者说以LangGraph为代表的下一代智能体框架登场的背景。它们要解决的正是LangChain Agent在复杂场景下的“力不从心”。DeepAgents不是一个具体的、官方的库名它更像是一个概念指的是那些构建在LangChain等基础组件之上采用更先进架构如图计算、状态机来实现深度、复杂、可编排智能体的方法和框架。而LangGraph作为LangChain官方推出的“亲儿子”是目前实现这一理念最成熟、最受关注的选择。它用“图”Graph的思维来建模智能体的工作流节点Node代表一个执行步骤如调用LLM、运行工具边Edge代表步骤之间的流转条件。这听起来有点抽象但简单理解就是它把智能体从一个线性的“流水线工人”升级成了一个有明确分工和协作规则的“项目团队”每个成员节点只负责自己那部分通过清晰的规则边来传递任务和结果。所以当你看到“langchain deepagents-深度研究实战”这个标题时它指向的绝不仅仅是学会调用几个API。它是一场思维模式的升级从编写线性的、脆弱的脚本到设计健壮的、可扩展的、可视化的智能体工作流。接下来我会结合我近期的实战项目带你深入LangGraph的核心看看我们是如何用它来构建一个真正能打的“深度智能体”的。2. LangGraph核心架构拆解State、Node与Edge是如何协同工作的要玩转LangGraph你必须吃透它的三个核心概念State状态、Node节点和Edge边。这构成了它整个运行时的骨架。很多人一开始会被这些术语吓到其实我们可以用一个非常生活化的场景来类比一个智能体就像一家餐厅的后厨。State状态就是后厨里那张唯一的、共享的订单白板。上面记录了当前所有订单的信息客人点了什么菜用户输入哪些菜正在做执行中任务哪些菜已经做好了已完成的结果以及后厨现有的食材上下文记忆。这个白板State是一个字典Dict结构所有厨师Node都只能读写这块白板来协作而不是互相私下传递纸条。这保证了信息的一致性和可追溯性。Node节点就是后厨里的各个工位或厨师。比如有专门看订单的“理解员”LLM调用节点有负责切菜的“预处理员”工具调用节点有掌勺的“主厨”另一个LLM节点还有负责摆盘的“装饰员”结果格式化节点。每个节点都是一个Python函数它的唯一职责就是从State白板上读取它需要的信息进行处理然后把结果写回State白板。它很“单纯”只关心自己的输入和输出。Edge边就是连接工位之间的传送带和规则。它决定了订单State的流向。比如规则可以是“如果订单里有‘牛排’就传给主厨节点如果是‘沙拉’就传给预处理员节点”。或者更简单“无论上一步做了什么下一步都传给理解员节点”。这些规则就是边它们可以是条件判断conditional_edge也可以是无条件流转Edge。让我们用代码来具象化这个餐厅后厨。假设我们要构建一个智能体它能理解用户问题并决定是直接回答简单问题还是去网上搜索后回答复杂问题。首先我们定义State。在LangGraph中我们通常用一个TypedDict来明确定义State里都有哪些“字段”。from typing import TypedDict, Annotated from langgraph.graph.message import add_messages import operator class State(TypedDict): # 消息历史这是一个由LangGraph管理的特殊列表用于记录对话 messages: Annotated[list, add_messages] # 用户当前的问题 user_query: str # 判断结果是‘direct_answer’还是‘need_search’ next_step: str # 从网络搜索到的内容如果需要的话 search_results: str # 最终给用户的答案 final_answer: str这里Annotated[list, add_messages]是LangGraph提供的一个语法糖它自动帮我们管理消息列表的追加非常方便。其他字段就是普通的字符串。接下来我们创建节点。每个节点都是一个函数接收一个State字典更新它然后返回更新后的字典。def understand_query(state: State) - State: 节点1理解用户问题并判断下一步动作 from langchain_openai import ChatOpenAI llm ChatOpenAI(model“gpt-4”) # 从State中获取用户问题 query state[“user_query”] # 让LLM判断问题类型 prompt f””” 请判断以下用户问题是否需要联网搜索才能准确回答。 问题{query} 如果问题涉及实时信息、最新事件、特定数据查询请回答‘need_search’。 如果问题是一般性知识、解释概念、无需最新信息请回答‘direct_answer’。 只输出‘need_search’或‘direct_answer’。 “”” decision llm.invoke(prompt).content.strip() # 将判断结果写回State state[“next_step”] decision return state def direct_answer_node(state: State) - State: 节点2直接回答问题 from langchain_openai import ChatOpenAI llm ChatOpenAI(model“gpt-4”) query state[“user_query”] prompt f”请回答以下问题{query}” answer llm.invoke(prompt).content state[“final_answer”] answer return state def search_and_answer_node(state: State) - State: 节点3搜索后回答问题 # 假设我们有一个搜索工具 from my_tools import web_search query state[“user_query”] # 1. 执行搜索 results web_search(query) state[“search_results”] results # 2. 基于搜索结果生成答案 from langchain_openai import ChatOpenAI llm ChatOpenAI(model“gpt-4”) prompt f””” 基于以下搜索结果为用户的问题生成一个全面、准确的答案。 用户问题{query} 搜索结果{results} “”” answer llm.invoke(prompt).content state[“final_answer”] answer return state现在我们有三个厨师节点了。怎么把他们组织起来呢这就需要定义边构建图Graph。from langgraph.graph import StateGraph, END # 创建一个图并指定它处理的是我们定义的State类型 workflow StateGraph(State) # 添加节点 workflow.add_node(“understand”, understand_query) workflow.add_node(“direct_answer”, direct_answer_node) workflow.add_node(“search_answer”, search_and_answer_node) # 设置入口点所有流程都从‘understand’节点开始 workflow.set_entry_point(“understand”) # 添加条件边从‘understand’节点出来后根据State里的‘next_step’字段决定去哪 workflow.add_conditional_edges( “understand” # 这是一个路由函数它读取State返回下一个节点的名字 lambda state: state[“next_step”] # 一个映射将路由函数返回的值‘direct_answer’或‘need_search’映射到对应的节点名 { “direct_answer”: “direct_answer” “need_search”: “search_answer” } ) # 添加普通边从‘direct_answer’和‘search_answer’节点出来后都直接结束 workflow.add_edge(“direct_answer”, END) workflow.add_edge(“search_answer”, END) # 编译图得到可执行的对象 app workflow.compile()至此一个最简单的LangGraph智能体就构建完成了。它的执行流程是输入用户问题 -understand节点判断 - 根据判断结果路由到direct_answer或search_answer节点 - 生成答案 - 结束。整个过程中State就像那个流动的订单白板承载了所有中间数据和最终结果。注意在实际开发中LLM调用和工具调用如web_search往往是性能瓶颈和错误来源。务必为它们添加完善的超时timeout和重试retry机制。例如使用tenacity库进行装饰或者利用LangChain内置的RunnableRetry。否则网络抖动或API限流会导致整个智能体流程卡死。3. 实战构建一个具备长期记忆与工具调用的研究型智能体理解了基础架构我们来点真格的。我将还原一个近期为内部知识库构建的“研究型智能体”项目。它的目标是用户输入一个复杂的研究主题例如“对比LangGraph和CrewAI在复杂工作流设计上的优劣”智能体能够自动进行多轮、深度的资料搜集、信息整理与对比分析并最终生成一份结构化的报告。这远非一次搜索就能解决它需要长期记忆来避免重复搜索需要灵活的工具调用来获取不同维度的信息还需要子图来管理复杂的分析子任务。3.1 状态设计为复杂任务预留空间对于复杂智能体State的设计是重中之重它决定了系统的扩展性和数据流是否清晰。我们为研究型智能体设计了如下Statefrom typing import TypedDict, Annotated, List, Optional from langgraph.graph.message import add_messages import datetime class ResearchState(TypedDict): # 核心对话与指令 messages: Annotated[list, add_messages] research_topic: str # 研究主题 user_requirements: str # 用户的额外要求如“侧重工程实践” # 执行过程 current_subtask: Optional[str] # 当前正在执行的子任务名 subtask_results: Annotated[List[str], operator.add] # 所有子任务的结果列表 # 工具与记忆 search_queries: Annotated[List[str], operator.add] # 历史搜索查询用于去重 visited_urls: Annotated[List[str], operator.add] # 已访问的URL避免重复抓取 collected_data: Annotated[List[dict], operator.add] # 收集到的原始数据片段列表 # 输出 outline: Optional[str] # 报告大纲 report_content: Optional[str] # 最终报告内容 status: str # 状态如 ‘planning’ ‘collecting’ ‘analyzing’ ‘writing’ ‘finished’这里有几个关键点Annotated[List[str], operator.add]这是LangGraph的“缩减器”Reducer语法。它告诉框架当多个节点并行或先后修改这个字段时不是覆盖而是用operator.add即列表的操作来合并结果。这对于收集碎片化信息如多次搜索的结果至关重要。current_subtask和status这是控制流程的“指针”和“信号灯”。通过它们我们可以让智能体在不同模式间切换。分离的collected_data和report_content原始数据和分析后的报告分开存储符合数据处理流程也便于调试和溯源。3.2 工具集成与子图实现模块化分工智能体需要调用多种工具。我们使用LangChain的tool装饰器来定义from langchain.tools import tool from duckduckgo_search import DDGS import asyncio tool async def web_search_tool(query: str, max_results: int 5) - str: 使用DuckDuckGo进行网络搜索。输入应为搜索关键词。 try: with DDGS() as ddgs: results [] # 使用异步避免阻塞注意DDGS可能需要同步调用这里用线程池包装 loop asyncio.get_event_loop() r await loop.run_in_executor(None, lambda: [r for r in ddgs.text(query, max_resultsmax_results)]) for res in r[:max_results]: results.append(f”标题{res[‘title’]}\n摘要{res[‘body’]}\n链接{res[‘href’]}\n{‘-’*40}”) return “\n”.join(results) if results else “未找到相关结果。” except Exception as e: return f”搜索过程出错{str(e)}” tool def summarize_text_tool(long_text: str) - str: 总结长文本的核心观点。 # 这里可以调用一个专门的总结LLM或者使用gpt-3.5-turbo等 from langchain_openai import ChatOpenAI llm ChatOpenAI(model“gpt-3.5-turbo-16k” temperature0) prompt f”请用不超过200字总结以下文本的核心内容和观点\n\n{long_text}” return llm.invoke(prompt).content # 将工具绑定到LLM from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor llm ChatOpenAI(model“gpt-4” temperature0.1) tools [web_search_tool, summarize_text_tool] agent create_tool_calling_agent(llm, tools, prompt) # prompt需自定义 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue)现在我们有了执行单一任务的“工人”agent_executor。但研究任务需要规划、收集、分析、写作等多个阶段。我们可以为每个阶段创建一个子图Subgraph。子图本身也是一个完整的图可以被主图调用这实现了功能的模块化和复用。例如创建一个资料收集子图from langgraph.graph import StateGraph, START, END def create_collection_subgraph(): “””创建资料收集子图负责执行一轮搜索、去重、保存结果。”“” from typing import TypedDict class CollectionState(TypedDict): topic: str search_queries: list visited_urls: list collected_data: list def plan_query(state: CollectionState): “””根据主题规划出几个搜索关键词。”“” # 调用LLM生成3-5个相关搜索词 planned_queries [f”{state[‘topic’]} 最新发展” f”{state[‘topic’]} 优缺点” f”{state[‘topic’]} 实战案例”] state[‘search_queries’].extend(planned_queries) return state def execute_search(state: CollectionState): “””执行一个尚未执行的搜索查询。”“” # 简化逻辑取第一个未执行的查询 for query in state[‘search_queries’]: if query not in state.get(‘_executed_queries’ []): results web_search_tool.invoke(query) # 解析结果提取链接和内容去重后存入collected_data # … (具体解析去重逻辑) state[‘collected_data’].append({‘query’: query ‘results’: parsed_results}) state.setdefault(‘_executed_queries’ []).append(query) break return state def check_completion(state: CollectionState): “””检查是否收集了足够的数据。”“” # 简单策略如果已执行查询数3或者collected_data条目数5则认为完成 if len(state.get(‘collected_data’ [])) 5: return “enough_data” elif len(state.get(‘_executed_queries’ [])) 3: return “enough_queries” else: return “continue_search” # 构建子图 workflow StateGraph(CollectionState) workflow.add_node(“plan” plan_query) workflow.add_node(“search” execute_search) workflow.add_edge(START “plan”) workflow.add_edge(“plan” “search”) workflow.add_conditional_edges( “search” check_completion {“enough_data”: END “enough_queries”: END “continue_search”: “search”} ) return workflow.compile()在主图中我们可以像调用一个超级节点一样调用这个子图。这使主图逻辑变得非常清晰规划阶段 - 调用收集子图 - 分析阶段 - 调用写作子图。3.3 编译与运行可视化与调试将所有的节点和子图通过边连接起来就形成了主图。使用workflow.compile()进行编译。# 假设我们已经定义了 plan_node analysis_node write_node 等主图节点 main_workflow StateGraph(ResearchState) main_workflow.add_node(“plan” plan_node) main_workflow.add_node(“collect” collection_subgraph) # 这里传入的是编译好的子图 main_workflow.add_node(“analyze” analysis_node) main_workflow.add_node(“write” write_node) main_workflow.set_entry_point(“plan”) main_workflow.add_edge(“plan” “collect”) main_workflow.add_edge(“collect” “analyze”) main_workflow.add_edge(“analyze” “write”) main_workflow.add_edge(“write” END) research_agent main_workflow.compile()LangGraph一个强大的功能是可视化。你可以直接将图导出为PNG。from IPython.display import Image display try: display(Image(research_agent.get_graph().draw_mermaid_png())) except: # 如果环境不支持可以输出Mermaid文本到支持渲染的编辑器查看 print(research_agent.get_graph().draw_mermaid())这张图是你智能体的“架构图”对于理解流程、排查问题有巨大帮助。运行智能体非常简单initial_state { “messages”: [] “research_topic”: “对比LangGraph和CrewAI在复杂工作流设计上的优劣” “user_requirements”: “请侧重从开发体验、状态管理、错误处理三个方面进行对比。” “status”: “planning” “search_queries”: [] “visited_urls”: [] “collected_data”: [] “subtask_results”: [] } # 以流式事件的方式运行可以实时看到执行到了哪个节点 final_state None for event in research_agent.stream(initial_state stream_mode“values”): node_name list(event.keys())[0] print(f”\n 进入节点[{node_name}] ) # 可以在这里打印或记录state的特定字段 if node_name “__end__”: final_state event[node_name] print(“\n 任务执行完毕 ”) print(f”最终报告状态{final_state.get(‘status’)}”) # 可以保存或输出final_state[‘report_content’]在流式运行中你会看到智能体按照你设计的蓝图一步步推进规划搜索词、触发收集子图进行多轮搜索、分析收集到的数据、最后撰写报告。整个过程状态清晰可控性强。4. 性能调优与生产级部署的避坑指南构建出可运行的智能体只是第一步。要让它在真实环境中稳定、高效地工作还有一大堆“坑”要填。以下是我在项目中积累的关键经验。4.1 状态管理的性能陷阱与优化State是共享内存频繁读写大对象比如很长的collected_data列表会影响性能尤其是在多线程或异步环境下。优化策略分而治之不要把所有数据都塞进一个State。对于大型、独立的数据块如一篇完整的爬取文章可以考虑存储一个引用如ID或路径而将数据本身存入外部数据库如Redis、PostgreSQL或向量数据库如Chroma、Weaviate。State里只保留元数据和索引。使用Annotated进行高效更新如前所述对于列表类数据使用Annotated[List, operator.add]让LangGraph在后台进行合并比在节点函数中手动state[‘list’].append(item); return state更高效、更安全。定期清理对于长期运行的智能体如聊天机器人State中的messages历史会无限增长最终导致LLM上下文超长响应变慢且成本激增。必须在图中设计一个“清理节点”定期将旧消息总结压缩Summary或转移到长期记忆存储中。4.2 工具调用的稳定性保障工具调用尤其是网络请求是智能体最脆弱的环节。必须做到超时设置为每一个工具调用、每一个LLM请求设置明确的超时。使用asyncio.wait_for或httpx.Timeout。import asyncio from langchain_core.runnables import RunnableConfig async def safe_tool_call(tool, input_args): try: # 设置5秒超时 result await asyncio.wait_for(tool.ainvoke(input_args) timeout5.0) return result except asyncio.TimeoutError: return “工具调用超时请重试或跳过此步骤。”指数退避重试对于可能因瞬时网络问题或API限流导致的失败实施重试机制。LangChain的RunnableRetry可以很方便地包装你的工具或LLM。from langchain_core.runnables import RunnableRetry retry_policy RunnableRetry( retry_if_exception_type(Exception) # 重试所有异常 wait_exponential_jitterTrue # 指数退避抖动 max_attempt_number3 ) robust_llm retry_policy | llm # 将重试策略应用到LLM上优雅降级当核心工具如搜索失败时智能体不应该崩溃。可以在State中设计一个fallback_mode标志当工具连续失败时切换到一个降级模式例如仅基于已有知识库回答并告知用户能力受限。4.3 错误处理与流程中断在复杂的图中一个节点的失败不应导致整个流程崩溃。LangGraph提供了interrupt机制。节点级Try-Catch在每个节点函数内部进行细致的异常捕获并将错误信息写入State例如state[‘last_error’] str(e)然后让流程根据错误信息路由到专门的“错误处理节点”。使用StateGraph的interrupt你可以定义一些“中断”点。当state[‘status’] ‘error’时通过条件边将流程导向一个human_review_node人工审核节点或者一个retry_node重试节点。这给了你从错误中恢复或引入人工干预的机会。超时中断整个图对于长时间运行的任务可以在调用app.invoke()或app.stream()时设置超时或者在外层用asyncio.wait_for包装。4.4 记忆Memory的持久化与加载智能体的价值在于其连续性。你需要将一次运行结束后的State保存下来并在下次同会话中加载。LangGraph的Checkpointer抽象就是干这个的。from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 # 1. 创建一个SQLite检查点存储器 conn sqlite3.connect(“checkpoints.db” check_same_threadFalse) checkpointer SqliteSaver(conn) # 2. 在编译图时传入checkpointer app workflow.compile(checkpointercheckpointer) # 3. 运行智能体时指定一个thread_id会话ID config {“configurable”: {“thread_id”: “user_123_session_1”}} initial_state {…} result app.invoke(initial_state configconfig) # 4. 下次运行时可以从检查点恢复状态 # 通过get_state获取最新状态或直接invoke时会自动从上次中断处继续 latest_state checkpointer.get_tuple(config).next使用检查点你可以实现智能体的“暂停/继续”以及跨对话的长期记忆。这对于构建客服机器人、个人助理等场景至关重要。4.5 监控与可观测性在生产环境你必须要知道你的智能体在干什么、性能如何。结构化日志在每个节点的入口和出口记录关键信息节点名、输入State摘要、输出State摘要、耗时、错误。使用structlog或loggingJSON格式输出方便接入ELK或Datadog。追踪Tracing利用LangSmithLangChain官方平台或OpenTelemetry进行深度追踪。这能让你清晰地看到每一次LLM调用、工具调用的输入输出、延迟和token消耗是进行成本分析和性能优化的黄金标准。自定义监控指标在State中埋点统计工具调用成功率、各节点平均耗时、最终输出质量评分可通过另一个LLM评估等并暴露给Prometheus等监控系统。5. 超越基础探索LangGraph的高级模式与生态当你掌握了上述核心内容后可以探索一些更高级的模式让智能体更强大。5.1 多智能体协作与竞争LangGraph的图模型天然支持多智能体。你可以创建多个“智能体节点”每个节点背后是一个独立的LLM甚至可以是不同模型的LLM比如一个GPT-4负责创意一个Claude负责严谨分析并让它们通过共享的State进行通信和协作。例如一个“辩论图”可以让两个智能体节点就一个议题进行多轮辩论将各自观点写入State最后由一个“裁判”节点进行总结。这通过条件边和循环很容易实现。5.2 与FastAPI集成构建API服务将编译好的app即你的智能体封装成FastAPI应用是提供服务的标准做法。from fastapi import FastAPI, HTTPException from pydantic import BaseModel import uvicorn app_fastapi FastAPI(title“Research Agent API”) class ResearchRequest(BaseModel): topic: str requirements: str “” app_fastapi.post(“/research”) async def start_research(request: ResearchRequest): try: initial_state { “research_topic”: request.topic “user_requirements”: request.requirements “status”: “planning” # … 其他初始字段 } config {“configurable”: {“thread_id”: f”req_{uuid.uuid4().hex}”}} # 异步流式响应Server-Sent Events async def event_generator(): async for event in research_agent.astream(initial_state configconfig stream_mode“values”): node list(event.keys())[0] if node ! “__end__”: yield {“event”: “node_update” “node”: node “data”: event[node].get(‘status’)} else: final_report event[node].get(‘report_content’) yield {“event”: “complete” “report”: final_report} return EventSourceResponse(event_generator()) except Exception as e: raise HTTPException(status_code500 detailstr(e))这样前端就可以通过SSEServer-Sent Events实时看到智能体的执行进度并在完成后获取报告。5.3 与Dify、Coze等平台的区别与选型现在市面上有很多智能体平台如Dify、Coze、阿里的魔搭ModelScope Agent。它们和基于LangGraph自研有什么区别Dify/Coze等平台优点是开箱即用可视化编排集成了大量预置工具、知识库和模型非常适合快速原型验证、无代码/低代码构建简单智能体应用。你不需要关心部署和运维。缺点是灵活性受限深度定制复杂比如实现我们上面那种复杂的状态管理和子图且可能产生平台绑定和成本。LangGraph自研优点是完全自主可控灵活性极高可以实现任何你能想象到的复杂逻辑和架构。能与现有技术栈深度集成便于进行性能优化和成本控制。缺点是门槛高需要较强的工程能力并且要自行处理部署、监控、运维等所有生产问题。选型建议如果你的需求是快速做一个概念验证PoC或者智能体逻辑相对标准聊天、简单问答、基于文档的检索优先考虑Dify等平台。如果你的智能体需要复杂的业务逻辑、独特的状态流、与内部系统的深度集成、或对性能和成本有极致要求那么LangGraph自研是更合适的选择。它更像是一个“智能体框架的框架”为你提供了构建强大引擎的所有零件。从我个人的实战经验来看LangGraph的学习曲线确实比直接使用LangChain Agent要陡峭但带来的能力提升是数量级的。它迫使你以更工程化、更结构化的方式思考智能体这种思维模式的价值远超学会一个工具本身。当你成功用它搭建起第一个能稳定处理复杂任务的智能体时那种对流程的掌控感和成就感是使用现成平台无法比拟的。