调用工具是agent的强大之处比如我们查询某地的天气时间等等但这只能完成简单任务对复杂问题的处理还有欠缺。因为复杂任务有对工具的多次调用及选择同时对agent的执行后的结果需要去判断有可能会出错…...为了agent能够解决复杂任务我们引入了AgentLoopLoop结束“循环”的意思相当于给定agent一个处理问题的执行过程。1.ReActReAct就是reasoningaction思考与行动。我们给定模型一个执行结果这里我画了一个大概的执行过程。state是“状态机”用于记录整个过程的信息包含queryAIMessagetoolcall等。同时它也会作为信息一同输入给每个状态。相当于“日记本”我们这里会有2种实现过程便于大家理解。1自实现DEFAULT_TASK 请检查 inbox把 a.txt 移动到 archive然后告诉我整理后的目录变化。这是我们的总任务我们定义了2个工具展示文件移动文件。tool def list_files(path: str .) - str: List files in the demo workspace. target workspace_path(path) if target.is_file(): return target.relative_to(WORKSPACE).as_posix() files sorted(item for item in target.rglob(*) if item.is_file()) return \n.join(f- {item.relative_to(WORKSPACE).as_posix()} for item in files) or (empty) tool def move_file(source: str, target: str) - str: Move a file in the demo workspace. source_path workspace_path(source) target_path workspace_path(target) if . not in target_path.name: target_path target_path / source_path.name target_path.parent.mkdir(parentsTrue, exist_okTrue) shutil.move(str(source_path), str(target_path)) return fmoved {source} - {target_path.relative_to(WORKSPACE).as_posix()}SYSTEM_PROMPT 你是一个文件整理助手。 你可以反复调用工具直到完成用户任务。 推荐顺序先 list_files(inbox)再 move_file(inbox/a.txt, archive/a.txt)再 list_files(.)最后总结。 每一轮最多调用一个工具。 .strip() def main() - None: task .join(sys.argv[1:]).strip() or DEFAULT_TASK reset_workspace() print( 01. 手写 while AgentLoop ) print(\n用户任务:) print(task) print(\n运行前 workspace:) print(show_workspace()) tools [list_files, move_file] tool_map {item.name: item for item in tools} 工具是函数所以我们要用函数体去处理参数 所以我们创建了一个tool_map字典 {工具名函数名:函数体} 同时下文的tool_call[name]得到的也是工具名所以要去对应到函数体本身 llm model.bind_tools(tools) messages [SystemMessage(contentSYSTEM_PROMPT), HumanMessage(contenttask)] # 创建一个状态机state for turn in range(1, 8): print(f\n--- 第 {turn} 轮模型思考 ---) response llm.invoke(messages) messages.append(response) # 模型思考是否要调用工具调用什么工具 if not response.tool_calls: print(\n最终回答:) print(response.content) break tool_call response.tool_calls[0] print(\n模型决定调用工具:) print(ftool_name {tool_call[name]}) print(ftool_args {tool_call[args]}) result tool_map[tool_call[name]].invoke(tool_call[args]) # 行动过程 print(\n工具返回:) print(result) messages.append( ToolMessage( contentstr(result), nametool_call[name], tool_call_idtool_call[id], ) ) # 追加到state里 print(\n运行后 workspace:) print(show_workspace())创建modelsystem_promptmessages状态机通过循环8次一方面设置整个执行过程的调用次数另一方面避免重复工具调用。输出结果可以看到agent执行后将a.txt移动到了archive文件下同时我的目录下也自动创建了相应的文件夹。2langgraph实现langgraph实现就是通过用langgraph内置的方式去实现。graph就是图包含节点边的指向。整个过程就是这样start是默认开始节点END默认结束结束条件仍然是tool_calls的存在。由于toolsprompt与前文的实现基本一致这里重点说明langgraph。def agent_node(state: AgentState) - AgentState: print(\n[agent] 模型思考) response llm.invoke([SystemMessage(contentSYSTEM_PROMPT), *state[messages]]) return {messages: [*state[messages], response]} def tools_node(state: AgentState) - AgentState: response state[messages][-1] new_messages list(state[messages]) for tool_call in response.tool_calls: print(\n[tools] 执行工具) print(ftool_name {tool_call[name]}) print(ftool_args {tool_call[args]}) result tool_map[tool_call[name]].invoke(tool_call[args]) print(result) new_messages.append( ToolMessage( contentstr(result), nametool_call[name], tool_call_idtool_call[id], ) ) return {messages: new_messages} def should_continue(state: AgentState) - str: last_message state[messages][-1] return tools if getattr(last_message, tool_calls, None) else END graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, tools_node) graph.add_edge(START, agent) graph.add_conditional_edges(agent, should_continue) 创建了一个虚拟默认结点start - agent然后有一个should_continue就是判断是否有tool_calls工具调用 有 - tools 无 - end graph.add_edge(tools, agent) result graph.compile().invoke( {messages: [HumanMessage(contenttask)]}, config{recursion_limit: 12}, )创建节点初始化graph用add_node添加节点add_edge连接节点关系。should_continue就是用于判断是否还要进行tool调用由state提供信息。state创建也有所不同class AgentState(TypedDict):messages: list[BaseMessage]用AgentState修饰包含消息列表的记录。2.ReflectionReflection在ReAct的基础上更进一步将tool的执行情况进行评价因为存在可能调用一次tool并未能处理好任务同时agent的调用可能存在错误。这里的reflection也是一个LLM用于评价操作所以一共有2个LLM。这里同样有state去保留整个执行情况的信息。class AgentState(TypedDict):messages: list[BaseMessage]reflection: str多一个reflection评价tool执行情况记录信息。这里的任务同样与前文一致我们主要说明reflection的重点。SYSTEM_PROMPT 你是一个 ReAct 文件整理助手。 目标检查 inbox移动 inbox/a.txt 到 archive/a.txt查看整理后的目录然后总结。 每一轮最多调用一个工具。 如果有 reflection note请优先参考它决定下一步。 .strip() REFLECTION_PROMPT 你是 reviewer。根据最近的工具结果给 agent 一句下一步建议。 目标检查 inbox - 移动 inbox/a.txt 到 archive/a.txt - 查看整理后的目录 - 总结。 如果已经看到 archive/a.txt请提醒 agent 停止调用工具并总结。 只输出一句 reflection note。 .strip() def main() - None: task .join(sys.argv[1:]).strip() or DEFAULT_TASK reset_workspace() print( 03. LangGraph ReAct Reflection Node ) print(\n用户任务:) print(task) print(\n运行前 workspace:) print(show_workspace()) tools [list_files, move_file] tool_map {item.name: item for item in tools} base_llm load_llm() tool_llm base_llm.bind_tools(tools) def agent_node(state: AgentState) - AgentState: print(\n[agent] 模型思考) prompt SYSTEM_PROMPT if state[reflection]: prompt f\n\nReflection note: {state[reflection]} response tool_llm.invoke([SystemMessage(contentprompt), *state[messages]]) return {messages: [*state[messages], response], reflection: state[reflection]} def tools_node(state: AgentState) - AgentState: response state[messages][-1] new_messages list(state[messages]) for tool_call in response.tool_calls: print(\n[tools] 执行工具) print(ftool_name {tool_call[name]}) print(ftool_args {tool_call[args]}) result tool_map[tool_call[name]].invoke(tool_call[args]) print(result) new_messages.append( ToolMessage( contentstr(result), nametool_call[name], tool_call_idtool_call[id], ) ) return {messages: new_messages, reflection: state[reflection]} def reflection_node(state: AgentState) - AgentState: print(\n[reflection] 复盘工具结果) transcript \n.join(str(message.content) for message in state[messages][-4:]) note base_llm.invoke( [ SystemMessage(contentREFLECTION_PROMPT), HumanMessage(contenttranscript), ] ).content print(note) return {messages: state[messages], reflection: str(note)} def should_continue(state: AgentState) - str: last_message state[messages][-1] return tools if getattr(last_message, tool_calls, None) else END graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, tools_node) graph.add_node(reflection, reflection_node) graph.add_edge(START, agent) graph.add_conditional_edges(agent, should_continue) graph.add_edge(tools, reflection) graph.add_edge(reflection, agent) result graph.compile().invoke( {messages: [HumanMessage(contenttask)], reflection: }, config{recursion_limit: 12}, ) # .compile() 理解为从“设计图纸”到“可运行程序”的“一键打包/组装”过程,前面是只创建了静态grap现在是得到了可运行程序 print(\n最终回答:) print(result[messages][-1].content) print(\n运行后 workspace:) print(show_workspace())由于有2个LLM所以我们配置了不同的prompt。同时我们在agent_node理能看到传入的reflection与prompt一同追加输入。3.PlanExecute前面的方式能很好地去让agent处理问题但由于问题越来越复杂模型执行越复杂上下文会逐渐拉长导致模型遗忘了最初的任务。所以agent需要有规划地去完成各个部分的任务。PlanExecute同样在Reflection的基础上加入了规划Todo。无非就是不断对之前的方法进行优化调整基本都是一样的。planner用于制定Todo规划很明显它也是一个LLM由此这里有3个LLM。state也有些变化。class AgentState(TypedDict):task:strplan:list[str]messages:list[BaseMessage]reflection:strtask就是我们的总任务plan是对总任务的执行规划。PLANNER_PROMPT 把用户任务拆成 3 个步骤。 必须覆盖检查 inbox、移动 a.txt 到 archive、查看整理后的目录。 每行一个步骤不要写额外解释。 .strip() SYSTEM_PROMPT 你是一个 ReAct executor。 你会收到计划请按计划用工具一步步完成任务。 每一轮最多调用一个工具。 如果有 reflection note请优先参考它决定下一步。 .strip() REFLECTION_PROMPT 你是 reviewer。根据计划和最近工具结果给 executor 一句下一步建议。 如果已经看到 archive/a.txt请提醒 executor 停止调用工具并总结。 只输出一句 reflection note。 .strip() def main() - None: task .join(sys.argv[1:]).strip() or DEFAULT_TASK reset_workspace() print( 04. LangGraph Plan ReAct Reflection ) print(\n用户任务:) print(task) print(\n运行前 workspace:) print(show_workspace()) tools [list_files, move_file] tool_map {item.name:item for item in tools} base_llm model tool_llm model.bind_tools(tools) def planner_node(state: AgentState) - AgentState: print(\n[planner] 生成计划) response base_llm.invoke( [SystemMessage(contentPLANNER_PROMPT), HumanMessage(contentstate[task])] ) plan parse_plan(str(response.content)) for index, step in enumerate(plan, start1): print(f{index}. {step}) plan_text \n.join(f{index}. {step} for index, step in enumerate(plan, start1)) return { **state, plan: plan, messages: [HumanMessage(contentf用户任务{state[task]}\n\n计划\n{plan_text})], } def agent_node(state: AgentState) - AgentState: print(\n[agent] 按计划执行下一步) prompt SYSTEM_PROMPT if state[reflection]: prompt f\n\nReflection note: {state[reflection]} response tool_llm.invoke([SystemMessage(contentprompt), *state[messages]]) return {**state, messages: [*state[messages], response]} def tools_node(state: AgentState) - AgentState: response state[messages][-1] new_messages list(state[messages]) for tool_call in response.tool_calls: print(\n[tools] 执行工具) print(ftool_name {tool_call[name]}) print(ftool_args {tool_call[args]}) result tool_map[tool_call[name]].invoke(tool_call[args]) print(result) new_messages.append( ToolMessage( contentstr(result), nametool_call[name], tool_call_idtool_call[id], ) ) return {**state, messages: new_messages} def reflection_node(state: AgentState) - AgentState: print(\n[reflection] 复盘计划和工具结果) plan_text \n.join(state[plan]) transcript \n.join(str(message.content) for message in state[messages][-4:]) note base_llm.invoke( [ SystemMessage(contentREFLECTION_PROMPT), HumanMessage(contentf计划\n{plan_text}\n\n最近记录\n{transcript}), ] ).content print(note) return {**state, reflection: str(note)} def should_continue(state: AgentState) - str: last_message state[messages][-1] return tools if getattr(last_message, tool_calls, None) else END graph StateGraph(AgentState) graph.add_node(planner, planner_node) graph.add_node(agent, agent_node) graph.add_node(tools, tools_node) graph.add_node(reflection, reflection_node) graph.add_edge(START, planner) graph.add_edge(planner, agent) graph.add_conditional_edges(agent, should_continue) graph.add_edge(tools, reflection) graph.add_edge(reflection, agent) result graph.compile().invoke( {task: task, plan: [], messages: [], reflection: }, config{recursion_limit: 50}, ) print(\n最终回答:) print(result[messages][-1].content) print(\n运行后 workspace:) print(show_workspace())这就是基本的AgentLoop实现代码来源于Wood-Q/MokioAgent: 从ToolCall到Claw全Agent指南我也是学习者有啥不对的大家多多指正共同学习