LangChain LangGraph 企业级 Agent 开发实战从 ReAct 到多智能体协作一、引言2026年LangChain已经从一个LLM应用开发框架演进为AI Agent基础设施。配合LangGraph的图状态管理能力开发者可以构建具有复杂决策逻辑、多轮自修正能力的企业级Agent系统。本文将深入LangChain LangGraph的核心机制通过完整的代码实战展示从ReAct模式到多智能体协作的全流程。二、Agent的核心思想智能体Agent是一种能够感知环境、进行推理、决策并执行动作以实现目标的系统。在企业级应用中智能体通常需要具备以下能力长期记忆存储和检索历史信息跨会话保持上下文工具调用利用外部工具API、数据库、搜索引擎获取信息或执行操作规划与推理分解复杂任务制定执行步骤根据中间结果动态调整计划协作多个智能体之间或智能体与人之间的交互三、LangChain基础构建第一个Agent3.1 环境准备pipinstalllangchain langchain-openai langchain-community pipinstalllanggraph duckduckgo-search pipinstallchromadb tiktoken3.2 ReAct模式AgentReActReasoning Acting是Agent开发中最经典的模式fromlangchainimporthubfromlangchain.agentsimportAgentExecutor,create_react_agentfromlangchain_openaiimportChatOpenAIfromlangchain_community.toolsimportDuckDuckGoSearchResultsfromlangchain.toolsimportToolfromlangchain.memoryimportConversationBufferMemoryimportjson# 初始化LLMllmChatOpenAI(modelgpt-4o,temperature0.3,max_tokens4096,)# 定义工具search_toolDuckDuckGoSearchResults(max_results5,output_formatlist)defcalculator(expression:str)-str:安全计算数学表达式try:# 使用安全的eval替代方案importastimportoperator allowed_ops{ast.Add:operator.add,ast.Sub:operator.sub,ast.Mult:operator.mul,ast.Div:operator.truediv,ast.Pow:operator.pow,}defsafe_eval(node):ifisinstance(node,ast.Expression):returnsafe_eval(node.body)elifisinstance(node,ast.BinOp):leftsafe_eval(node.left)rightsafe_eval(node.right)returnallowed_ops[type(node.op)](left,right)elifisinstance(node,ast.Constant):returnnode.valueelse:raiseValueError(fUnsupported operation:{type(node)})treeast.parse(expression,modeeval)resultsafe_eval(tree)returnf计算结果{result}exceptExceptionase:returnf计算错误{str(e)}tools[Tool(nameweb_search,funcsearch_tool.run,description在互联网上搜索最新信息。输入搜索关键词。),Tool(namecalculator,funccalculator,description执行数学计算。输入数学表达式如2 3 * 4。),]# 获取ReAct提示模板prompthub.pull(hwchase17/react)# 创建Agentagentcreate_react_agent(llm,tools,prompt)# 创建Agent执行器agent_executorAgentExecutor(agentagent,toolstools,memoryConversationBufferMemory(memory_keychat_history),verboseTrue,max_iterations10,handle_parsing_errorsTrue,)# 执行查询resultagent_executor.invoke({input:2026年量子计算的最新进展是什么})print(result[output])3.3 自定义工具开发fromlangchain.toolsimportBaseToolfrompydanticimportBaseModel,FieldfromtypingimportType,OptionalimportaiohttpimportasyncioclassSQLQueryInput(BaseModel):query:strField(descriptionSQL查询语句)limit:intField(default100,description返回行数限制)classSQLQueryTool(BaseTool):name:strsql_querydescription:str执行SQL查询并返回结果。用于查询数据库中的业务数据。args_schema:Type[BaseModel]SQLQueryInputdef_run(self,query:str,limit:int100)-str:同步执行SQL查询importsqlite3 connsqlite3.connect(business.db)cursorconn.cursor()try:# 安全限制只允许SELECT语句query_upperquery.strip().upper()ifnotquery_upper.startswith(SELECT):return错误只允许执行SELECT查询# 添加LIMIT子句ifLIMITnotinquery_upper:queryf{query}LIMIT{limit}cursor.execute(query)columns[desc[0]fordescincursor.description]rowscursor.fetchall()result[]forrowinrows:result.append(dict(zip(columns,row)))returnjson.dumps(result,ensure_asciiFalse,defaultstr)exceptExceptionase:returnf查询错误{str(e)}finally:conn.close()asyncdef_arun(self,query:str,limit:int100)-str:异步执行SQL查询returnself._run(query,limit)四、LangGraph图状态管理与多步推理4.1 LangGraph核心概念LangGraph基于图论用于构建具有状态管理和循环的智能体系统。核心概念包括节点Node处理函数接收状态并返回更新边Edge定义节点间的转移逻辑状态State在所有节点间共享的数据结构条件边Conditional Edge根据状态决定下一个节点4.2 构建多步推理AgentfromtypingimportTypedDict,Annotated,List,Literalfromlanggraph.graphimportStateGraph,ENDfromlanggraph.checkpoint.memoryimportMemorySaverfromlangchain_core.messagesimportHumanMessage,AIMessage,ToolMessageimportoperator# 定义Agent状态classAgentState(TypedDict):messages:Annotated[List,operator.add]task:strplan:List[str]current_step:inttool_results:dictfinal_answer:strerror_count:intmax_retries:int# 创建状态图workflowStateGraph(AgentState)# 定义节点函数defplanner(state:AgentState)-AgentState:规划节点分解任务并制定执行计划llmChatOpenAI(modelgpt-4o,temperature0.2)promptf你是一个任务规划专家。请将以下任务分解为具体的执行步骤 任务{state[task]}请以JSON格式返回执行计划每个步骤包含 - step_id: 步骤编号 - description: 步骤描述 - tool_needed: 需要的工具名称如果没有则为null - expected_output: 期望的输出结果 只返回JSON数组不要包含其他内容。responsellm.invoke(prompt)planjson.loads(response.content)return{plan:plan,current_step:0,tool_results:{},error_count:0,max_retries:state.get(max_retries,3),}defexecutor(state:AgentState)-AgentState:执行节点执行当前步骤currentstate[plan][state[current_step]]step_idcurrent[step_id]ifcurrent[tool_needed]:tooltools_map.get(current[tool_needed])iftool:resulttool.run(current[description])state[tool_results][step_id]resultreturn{current_step:state[current_step]1,tool_results:state[tool_results],}defsynthesizer(state:AgentState)-AgentState:综合节点汇总所有步骤结果生成最终答案llmChatOpenAI(modelgpt-4o,temperature0.3)contextf 任务{state[task]}执行计划的各步骤结果{json.dumps(state[tool_results],ensure_asciiFalse,indent2)}请基于以上信息生成一个全面、准确的最终答案。responsellm.invoke(context)return{final_answer:response.content}deferror_handler(state:AgentState)-AgentState:错误处理节点state[error_count]1ifstate[error_count]state[max_retries]:# 重试当前步骤return{current_step:max(0,state[current_step]-1),error_count:state[error_count],}else:# 超过最大重试次数跳过当前步骤return{current_step:state[current_step]1,final_answer:部分步骤执行失败已生成基于可用信息的答案。,}# 条件路由函数defshould_continue(state:AgentState)-Literal[execute,synthesize]:ifstate[current_step]len(state[plan]):returnexecutereturnsynthesize# 添加节点workflow.add_node(plan,planner)workflow.add_node(execute,executor)workflow.add_node(synthesize,synthesizer)workflow.add_node(error_handler,error_handler)# 设置边workflow.set_entry_point(plan)workflow.add_edge(plan,execute)workflow.add_conditional_edges(execute,should_continue,{execute:execute,synthesize:synthesize,})workflow.add_edge(synthesize,END)# 编译图memoryMemorySaver()appworkflow.compile(checkpointermemory)# 执行图config{configurable:{thread_id:session-001}}resultapp.invoke({task:分析2026年新能源汽车市场趋势包括销量数据、主要品牌表现和技术创新},config)print(result[final_answer])五、多智能体协作系统5.1 多Agent架构设计fromlanggraph.graphimportStateGraph,ENDfromtypingimportTypedDict,List,Dict,AnyclassMultiAgentState(TypedDict):task:strsub_tasks:Dict[str,str]agent_results:Dict[str,str]coordinator_notes:List[str]final_output:strstatus:str# planning, executing, reviewing, done, errorclassAgentTeam:def__init__(self):self.agents{}self.workflowStateGraph(MultiAgentState)defregister_agent(self,name:str,role:str,expertise:List[str]):self.agents[name]{role:role,expertise:expertise,llm:ChatOpenAI(modelgpt-4o,temperature0.3,system_messagef你是{role}专家擅长{, .join(expertise)}。)}defbuild_workflow(self):# 协调者节点defcoordinator(state:MultiAgentState)-MultiAgentState:llmChatOpenAI(modelgpt-4o,temperature0.2)promptf你是多智能体系统的协调者。任务如下{state[task]}可用的Agent及其专业领域{json.dumps({k:v[role]fork,vinself.agents.items()},ensure_asciiFalse)}请将任务分解为子任务并分配给最合适的Agent。返回JSON格式 {{sub_tasks: {{agent_name: sub_task_description, ...}}}}responsellm.invoke(prompt)sub_tasksjson.loads(response.content)[sub_tasks]return{sub_tasks:sub_tasks,status:executing}# 执行节点defexecute_tasks(state:MultiAgentState)-MultiAgentState:resultsforagent_name,task_descinstate[sub_tasks].items():agentself.agents[agent_name]promptf作为{agent[role]}专家请完成以下子任务{task_desc}请提供详细、专业的分析和建议。responseagent[llm].invoke(prompt)results[agent_name]response.contentreturn{agent_results:results,status:reviewing}# 审查节点defreviewer(state:MultiAgentState)-MultiAgentState:llmChatOpenAI(modelgpt-4o,temperature0.3)promptf你是审核专家。请审查以下各Agent的输出结果检查一致性、完整性和准确性 原始任务{state[task]}各Agent的结果{json.dumps(state[agent_results],ensure_asciiFalse,indent2)}如果有不一致或需要补充的地方请指出。如果全部通过请汇总生成最终输出。responsellm.invoke(prompt)return{final_output:response.content,status:done}# 添加节点self.workflow.add_node(coordinator,coordinator)self.workflow.add_node(executor,execute_tasks)self.workflow.add_node(reviewer,reviewer)# 设置流程self.workflow.set_entry_point(coordinator)self.workflow.add_edge(coordinator,executor)self.workflow.add_edge(executor,reviewer)self.workflow.add_edge(reviewer,END)returnself.workflow.compile()# 使用示例teamAgentTeam()team.register_agent(market_analyst,市场分析,[行业趋势,竞争格局,市场规模预测])team.register_agent(tech_expert,技术,[技术创新,技术路线,技术可行性评估])team.register_agent(financial_advisor,财务,[成本分析,ROI计算,融资策略])appteam.build_workflow()resultapp.invoke({task:评估公司在2026年进入AI Agent市场的可行性包括市场机会、技术方案和财务预测。})六、生产环境部署6.1 LangServe部署# server.pyfromfastapiimportFastAPIfromlangserveimportadd_routesfromlangchain_core.runnablesimportRunnableLambda appFastAPI(titleEnterprise Agent API)# 添加Agent路由add_routes(app,agent_executor,path/agent,input_typeAgentInput,output_typeAgentOutput,)# 添加健康检查app.get(/health)asyncdefhealth_check():return{status:healthy,version:1.0.0}# 添加监控端点app.get(/metrics)asyncdefmetrics():return{total_requests:request_counter,avg_latency_ms:avg_latency,error_rate:error_rate,token_usage:token_usage,}6.2 Docker化部署FROM python:3.12-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD [uvicorn, server:app, --host, 0.0.0.0, --port, 8000]七、总结LangChain LangGraph的组合为2026年的企业级Agent开发提供了坚实的基础设施。LangChain简化了LLM调用、工具集成和记忆管理而LangGraph则解决了复杂工作流的状态管理和多步推理问题。两者的结合使得开发者可以构建从简单的ReAct Agent到复杂的多智能体协作系统。关键实践建议使用ReAct模式作为Agent的起点逐步引入LangGraph的图状态管理工具函数必须包含完整的错误处理和重试逻辑多Agent系统中协调者Agent的设计至关重要生产环境必须配备完整的监控、日志和成本追踪体系定期进行红队测试评估Agent在异常输入下的表现