一、AgentExecutor的角色1.1 什么是AgentExecutorAgentExecutor是Agent的运行时环境负责┌─────────────────────────────────────────────┐ │ AgentExecutor 职责 │ ├─────────────────────────────────────────────┤ │ 1. 协调Agent与工具的交互 │ │ 2. 管理执行流程和状态 │ │ 3. 处理错误和异常 │ │ 4. 控制迭代次数 │ │ 5. 管理记忆组件 │ └─────────────────────────────────────────────┘1.2 与Agent的关系Agent: 大脑决定做什么 ↓ AgentExecutor: 执行者负责实际执行二、AgentExecutor核心参数2.1 基本参数fromlangchain.agentsimportAgentExecutor agent_executorAgentExecutor(agentagent,# Agent实例toolstools,# 工具列表verboseTrue,# 详细输出max_iterations10,# 最大迭代次数max_execution_timeNone,# 最大执行时间early_stopping_methodforce,# 提前停止方式handle_parsing_errorsFalse,# 处理解析错误trim_intermediate_steps-1,# 裁剪中间步骤)2.2 参数详解参数说明默认值max_iterations最大工具调用次数10max_execution_time最大执行时间(秒)Noneearly_stopping_method停止方式“force”handle_parsing_errors解析错误处理False2.3 early_stopping_method选项# force: 强制停止返回当前结果agent_executorAgentExecutor(agentagent,toolstools,early_stopping_methodforce)# generate: 让LLM生成最终答案agent_executorAgentExecutor(agentagent,toolstools,early_stopping_methodgenerate)三、执行流程详解3.1 内部执行循环# 伪代码展示执行流程defexecute(agent,tools,input):iterations0whileiterationsmax_iterations:# 1. 调用Agent获取下一步actionagent.plan(input)# 2. 检查是否完成ifaction.is_final:returnaction.answer# 3. 执行工具tool_resulttools[action.name].invoke(action.args)# 4. 记录结果inputupdate_with_result(input,tool_result)iterations1# 达到最大迭代次数returnhandle_early_stop()3.2 状态管理fromlangchain.agentsimportAgentExecutorfromlangchain_openaiimportChatOpenAIfromlangchain.toolsimporttooltooldefsearch(query:str)-str:returnf结果:{query}tools[search]llmChatOpenAI(modelgpt-4o-mini)# 创建带verbose的执行器agent_executorAgentExecutor(agentagent,toolstools,verboseTrue# 查看执行过程)# 执行过程会显示resultagent_executor.invoke({input:测试})四、错误处理机制4.1 解析错误# 自动处理解析错误agent_executorAgentExecutor(agentagent,toolstools,handle_parsing_errorsTrue# 自动处理)# 或者自定义处理agent_executorAgentExecutor(agentagent,toolstools,handle_parsing_errors解析错误请重试)4.2 工具执行错误fromlangchain.toolsimportTooldefrisky_operation(param:str)-str:ifnotparam:raiseValueError(参数不能为空)returnf处理:{param}safe_toolTool(namesafe_op,funclambdax:risky_operation(x)ifxelse错误参数为空,description安全操作)4.3 超时处理agent_executorAgentExecutor(agentagent,toolstools,max_execution_time30# 30秒超时)五、与记忆组件集成5.1 添加记忆fromlangchain.memoryimportConversationBufferMemoryfromlangchain.agentsimportAgentExecutor memoryConversationBufferMemory(memory_keychat_history,return_messagesTrue)agent_executorAgentExecutor(agentagent,toolstools,memorymemory,verboseTrue)5.2 多轮对话# 第一轮result1agent_executor.invoke({input:我叫小明})# 第二轮能记住result2agent_executor.invoke({input:我叫什么})六、自定义执行器6.1 继承AgentExecutorfromlangchain.agentsimportAgentExecutorfromtypingimportDict,AnyclassCustomAgentExecutor(AgentExecutor):自定义执行器def_call(self,inputs:Dict[str,str])-Dict[str,Any]:# 自定义执行逻辑print(f执行输入:{inputs})resultsuper()._call(inputs)print(f执行结果:{result})returnresultdef_get_tool_response(self,tool_name,tool_input,excNone):# 自定义工具响应ifexc:returnf工具{tool_name}执行出错:{exc}returnsuper()._get_tool_response(tool_name,tool_input)6.2 使用Runnable接口fromlangchain_core.runnablesimportRunnableLambda# 将AgentExecutor作为Runnable使用defprocess_with_agent(input_text:str)-str:resultagent_executor.invoke({input:input_text})returnresult[output]# 组合到链中chain(RunnableLambda(lambdax:x.strip())|RunnableLambda(process_with_agent)|RunnableLambda(lambdax:x.upper()))resultchain.invoke(测试问题)七、性能优化7.1 限制中间步骤# 减少中间步骤存储agent_executorAgentExecutor(agentagent,toolstools,trim_intermediate_steps5# 只保留最近5步)7.2 流式输出# 使用回调处理流式输出fromlangchain.callbacks.baseimportBaseCallbackHandlerclassStreamHandler(BaseCallbackHandler):defon_llm_new_token(self,token,**kwargs):print(token,end,flushTrue)agent_executorAgentExecutor(agentagent,toolstools,callbacks[StreamHandler()])7.3 异步执行importasyncioasyncdefasync_run():resultawaitagent_executor.ainvoke({input:问题})returnresult# 并发执行多个任务asyncdefbatch_run(inputs):tasks[agent_executor.ainvoke({input:i})foriininputs]resultsawaitasyncio.gather(*tasks)returnresults八、调试与监控8.1 使用verboseagent_executorAgentExecutor(agentagent,toolstools,verboseTrue# 打印详细执行过程)8.2 回调监控fromlangchain.callbacks.baseimportBaseCallbackHandlerclassDebugHandler(BaseCallbackHandler):defon_chain_start(self,serialized,inputs,**kwargs):print(f开始执行:{inputs})defon_chain_end(self,outputs,**kwargs):print(f执行完成:{outputs})defon_tool_start(self,serialized,input_str,**kwargs):print(f调用工具:{input_str})defon_tool_end(self,output,**kwargs):print(f工具结果:{output})agent_executorAgentExecutor(agentagent,toolstools,callbacks[DebugHandler()])九、小结功能参数/方法限制迭代max_iterations超时控制max_execution_time错误处理handle_parsing_errors记忆管理memory调试输出verbose流式处理callbacksAgentExecutor核心职责协调Agent、工具、记忆的运行时提供错误处理和执行控制。下一篇文章26 智能助手实战整合记忆、工具与检索的完整案例