LATSLanguage Agent Tree Search语言智能体树搜索ICML 2024 论文用 LangGraph 搭建一棵思考树每一步都生成多个备选方案像 DFS 一样探索所有可能路径让大模型当裁判投票挑最好的那条思考路径最后用这条最优路径输出答案不像普通 ReAct 一条路走到黑ToT 是把思考拆成多分支树每一步生成多个候选方案像深度优先搜索一样探索多条路径再投票选出最好的一条路线superfluoussuːˈpɜːfluəsadj. 多余的过剩的奢侈的外部参数max_candidates重规划下一个 step 时只需要前 x 个候选 stepmax_paths最大路径超过最大路径后只能跳 vote 决策否则会回到 run 节点继续生成当前步骤答案-决策出下一步候选步骤列表planning agentfrompydanticimportBaseModel,Fieldfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_google_genaiimportChatGoogleGenerativeAIclassPlan(BaseModel):Plan to follow in futuresteps:list[str]Field(descriptiondifferent steps to follow, should be in sorted order)system_prompt_template(# come up with a step by step plan制定一个分步计划# final step 的 result 应该是最终 answerFor the given task, come up with a step by step plan.\nThis plan should involve individual tasks, that if executed correctly will yield the correct answer. Do not add any superfluous steps.\nThe result of the final step should be the final answer. Make sure that each step has all the information needed - do not skip steps.)planner_promptChatPromptTemplate.from_messages([(system,system_prompt_template),(user,Prepare a plan how to solve the following task:\n{task}\n)])plannerplanner_prompt|ChatGoogleGenerativeAI(modelgemini-2.5-flash,temperature1.0).with_structured_output(Plan)构建 agent 用于执行 plan 中的一个 stepfromlangchain_community.agent_toolkits.load_toolsimportload_toolsfromlanggraph.prebuiltimportcreate_react_agent llmChatGoogleGenerativeAI(modelgemini-2.5-flash)# DDG DuckDuckGoddg-search 就是 DuckDuckGo 网页搜索工具MCP 工具# DuckDuckGo 是主打隐私保护的英文搜索引擎官网[duckduckgo.com](https://duckduckgo.com)# 适合 Agent 回答时效性问题新闻、文档、产品信息弥补大模型知识截止的缺陷toolsload_tools(tool_names[ddg-search,arxiv,wikipedia],llmllm)system_prompt(# 给定一个用于解决一个 task 的通用 plan和一个特定 stepwork on this step进行这一步# 使用 tools 来二确自己Youre a smart assistant that carefully helps to solve complex tasks.\n Given a general plan to solve a task and a specific step, work on this step. Dont assume anything, keep in minds things might change and always try to use tools to double-check yourself.\nUse Search to gather information about common facts, fresh events and news, use Arxiv to get ideas on recent research and use Wikipedia for common knowledge.)step_template(Given the task and the plan, try to execute on a specific step of the plan.\nTASK:\n{task}\n\nPLAN:\n{previous_steps}\n\nSTEP TO EXECUTE:\n{step}\n)prompt_templateChatPromptTemplate.from_messages([(system,system_prompt),(user,step_template),])execution_agentprompt_template|create_react_agent(modelllm,toolstools)定义一个单独的结构来跟踪将要探索的选项树fromtypingimportOptionalclassTreeNode:def__init__(self,node_id:int,# step计划 / 动作 / 思考要做什么# step 是 当前步骤的名字# step_output观测、返回结果做完之后拿到了什么step:str,step_output:Optional[str]None,parent:Optional[TreeNode]None,):self.node_idnode_id self.stepstep self.step_outputstep_output self.parentparent self.children[]self.final_responseNonedef__repr__(self):parent_idself.parent.node_idifself.parentelseNonereturnfNode_id:{self.node_id}, parent:{parent_id},{len(self.children)}children.defget_full_plan(self)-str:Returns formatted plan with step numbers and past results.steps[]nodeselfwhilenode.parent:steps.append((node.step,node.step_output))nodenode.parent full_plan[]fori,(step,result)inenumerate(steps[::-1]):ifresult:full_plan.append(f#{i1}. Planned step:{step}\nResult:{result}\n)return\n.join(full_plan)定义额外的状态probableˈprɒbəb(ə)ladj. 很可能发生或存在等的大概的n. 可能获胜的人或动物可能入选者importoperatorfromcollectionsimportdequefromtypingimportAnnotated,TypedDictclassPlanEvaluation(BaseModel):is_final:boolField(descriptionWhether the task is solved or not. Plan is not final if additional steps are required.,defaultFalse)score:floatField(descriptionA score from 0. to 1.0 that evaluates the quality of the plan and how probable it is to solve the task with this plan.,defaultFalse)classPlanState(TypedDict):task:strroot:TreeNode queue:deque[TreeNode]current_node:TreeNode next_node:TreeNode is_current_node_final:boolpaths_explored:Annotated[int,operator.add]visited_ids:set[int]max_id:intcandidates:Annotated[list[str],operator.add]best_candidate:str创建一个replanning agent考虑之前的steps和适当调节plancreate a replanning agent that takes into account the previous steps and adjusts the plan accordingly:fromlangchain_core.promptsimportChatPromptTemplate# 作用在执行完一步之后重新规划输出多个候选下一步动作# 不是只输出 1 个方案会生成多个可选路径这就是树的分支classReplanStep(BaseModel):Replanned next step in the plan.steps:list[str]Field(descriptiondifferent options of the proposed next step)llm_replannerllm.with_structured_output(ReplanStep)replanner_prompt_template(Suggest next action in the plan. Do not add any superfluous steps.\nIf you think no actions are needed, just return an empty list of steps. TASK: {task}\n PREVIOUS STEPS WITH OUTPUTS: {current_plan})replanner_promptChatPromptTemplate.from_messages([(system,Youre a helpful assistant. You goal is to help with planning actions to solve the task. Do not solve the task itself.),(user,replanner_prompt_template)])replannerreplanner_prompt|llm_replannerfinal step投票选出最好的 candidateWe also need a final step that includes a major voting for the best candidate:fromlangchain_core.promptsimportPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser prompt_votingPromptTemplate.from_template(Pick the best solution for a given task. \nTASK:{task}\n\nSOLUTIONS:\n{candidates}\n)# 模型投票选出最优分支# 输入多条不同路径的候选方案# LLM 充当评委从多个候选里挑选最优的下一步方案# 结构化输出返回选中的最佳候选编号# ToT 的核心亮点同一个步骤多条路线并行探索用 LLM 做评判、剪枝抛弃差的分支而不是一条路硬走def_vote_for_the_best_option(state):candidatesstate.get(candidates,[])ifnotcandidates:return{best_response:None}all_candidates[]fori,candidateinenumerate(candidates):all_candidates.append(fOPTION{i1}:{candidate})response_schema{type:STRING,# 枚举大模型只能从这些数字中选enum:[str(i1)foriinrange(len(all_candidates))],}llm_enumChatGoogleGenerativeAI(modelgemini-2.5-flash,).with_structured_output(response_schema)result(prompt_voting|llm_enum).invoke({candidates:\n.join(all_candidates),task:state[task]})return{best_candidate:candidates[int(result)-1]}_vote_for_the_best_option({candidates:[1,5,4],task:How much is 22?})汇总上述逻辑构建 agent 图fromtypingimportLiteralfromlanggraph.graphimportStateGraph,START,ENDfromlangchain_core.runnablesimportRunnableConfigfromlangchain_core.output_parsersimportStrOutputParserfromlanggraph.typesimportCommand# 不是当前单一步骤的答案是把【整条路径全部步骤 每一步执行结果】一次性交给 LLM生成针对原始 task 的最终完整答案final_promptPromptTemplate.from_template(Youre a helpful assistant that has executed on a plan.Given the results of the execution, prepare the final response.\nDont assume anything\nTASK:\n{task}\n\nPLAN WITH RESUlTS:\n{plan}\nFINAL RESPONSE:\n)responderfinal_prompt|llm|StrOutputParser()# 这是一条单链链式树也就是链表结构不是多分支树一条从头到尾串联起来的父子链条# queueA - B - Casyncdef_build_initial_plan(state:PlanState)-PlanState:planawaitplanner.ainvoke(state[task])queuedeque()rootTreeNode(stepplan.steps[0],node_id1)queue.append(root)current_rootrootfori,stepinenumerate(plan.steps[1:]):childTreeNode(node_idi2,stepstep,parentcurrent_root)current_root.children.append(child)queue.append(child)current_rootchild# max_id 记录当前已经使用过最大的 node_id。后面新增子节点时拿这个值 1保证每个 TreeNode 的node_id全局唯一不会重复return{root:root,queue:queue,max_id:i2}asyncdef_run_node(state:PlanState,config:RunnableConfig):# 用 next_node 显式指定下一个要执行的节点nodestate.get(next_node)visited_idsstate.get(visited_ids,set())queuestate[queue]ifnodeisNone:whilequeueandnotnode:nodestate[queue].popleft()ifnode.node_idinvisited_ids:nodeNoneifnotnode:returnCommand(gotovote,update{})stepawaitexecution_agent.ainvoke({previous_steps:node.get_full_plan(),step:node.step,task:state[task]})node.step_outputstep[messages][-1].content visited_ids.add(node.node_id)return{current_node:node,queue:queue,visited_ids:visited_ids,next_node:None}asyncdef_plan_next(state:PlanState,config:RunnableConfig)-PlanState:# max_candidates重规划时只需要前 x 个候选max_candidatesconfig[configurable].get(max_candidates,1)nodestate[current_node]next_stepawaitreplanner.ainvoke({task:state[task],current_plan:node.get_full_plan()})ifnotnext_step.steps:return{is_current_node_final:True}# max_id 记录当前已经使用过最大的 node_id。后面新增子节点时拿这个值 1保证每个 TreeNode 的node_id全局唯一不会重复max_idstate[max_id]forstepinnext_step.steps[:max_candidates]:childTreeNode(node_idmax_id1,stepstep,parentnode)max_id1node.children.append(child)state[queue].append(child)return{is_current_node_final:False,next_node:child,max_id:max_id}# candidates存放本次探索路径得到的候选答案注意是最终交可能会选定给到用户的答案asyncdef_get_final_response(state:PlanState)-PlanState:nodestate[current_node]final_responseawaitresponder.ainvoke({task:state[task],plan:node.get_full_plan()})node.final_responsefinal_responsereturn{paths_explored:1,candidates:[final_response]}def_should_create_final_response(state:PlanState)-Literal[run,generate_response]:returngenerate_responseifstate[is_current_node_final]elserundef_should_continue(state:PlanState,config:RunnableConfig)-Literal[run,vote]:max_pathsconfig[configurable].get(max_paths,30)# max_paths 不能太深ifstate.get(paths_explored,1)max_paths:returnvoteifstate[queue]orstate.get(next_node):returnrunreturnvotebuilderStateGraph(PlanState)builder.add_node(initial_plan,_build_initial_plan)builder.add_node(run,_run_node)builder.add_node(plan_next,_plan_next)builder.add_node(generate_response,_get_final_response)builder.add_node(vote,_vote_for_the_best_option)builder.add_edge(START,initial_plan)builder.add_edge(initial_plan,run)builder.add_edge(run,plan_next)# _should_create_final_response 跳转 run 或者 generate_responsebuilder.add_conditional_edges(plan_next,_should_create_final_response)# _should_continue 跳转 run 或者 votebuilder.add_conditional_edges(generate_response,_should_continue)builder.add_edge(vote,END)graphbuilder.compile()fromIPython.displayimportImage,display display(Image(graph.get_graph().draw_mermaid_png()))ToT 和普通 ReAct 的区别