当前位置：首页 > news >正文

LangGraph高级教程：构建规划执行型智能体

news 2025/6/13 16:23:12

LangGraph高级教程：构建规划执行型智能体

在现代AI应用开发中，我们通常需要构建能够自主规划和执行任务的智能体系统。本文将详细介绍如何使用LangGraph框架构建一个具有规划能力的AI智能体，该智能体能够自动生成解决问题的步骤计划，并逐步执行这些计划以得出最终答案。

1. 什么是规划执行型智能体？

规划执行型智能体是一种先计划后执行的智能系统。它首先分析问题，制定解决方案的步骤，然后按照计划逐步执行，如有必要还可以重新规划。这种模式非常适合解决需要多步骤推理的复杂问题。

本文构建的智能体具有以下三个主要组件：

规划器(Planner)：负责生成解决问题的步骤计划
执行器(Agent)：负责执行计划中的具体步骤
重新规划器(Replanner)：根据执行结果，决定是继续执行、重新规划，还是直接回答

2. 环境准备与依赖导入

首先，我们需要导入必要的库和设置环境：

import os
from typing import TypedDict, Annotated, List, Tuplefrom dotenv import load_dotenv
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplateload_dotenv()
tools = [TavilySearchResults(max_length=1, tavily_api_key=os.getenv("TAVILY_API_KEY"))]
import operator
from langchain import hub
from langchain_openai import ChatOpenAI
import asyncio
from langgraph.prebuilt import create_react_agent
from langsmith import Client

3. 设置基础智能体

接下来，我们设置一个基础的ReAct智能体，用于执行具体任务：

client = Client(api_key=os.getenv("SMITH_KEY"))
prompt_template = ChatPromptTemplate.from_messages([("system", "你是一个人工智能助手，回答用户的问题，使用中文回答"),("user", "{question}"),]
)llm = ChatOpenAI(base_url="https://ark.cn-beijing.volces.com/api/v3", api_key=os.getenv("OB_OPENAI_KEY"),model="doubao-1-5-thinking-pro-250415")agent_executor = create_react_agent(llm, tools)

4. 定义状态和数据模型

我们需要定义数据结构来存储智能体的状态信息：

class PlanExecute(TypedDict):input: str  # 输入plan: List[str]  # 计划past_steps: Annotated[List[Tuple], operator.add]  # 步骤response: str

定义计划结构：

from langchain_core.pydantic_v1 import BaseModel, Fieldclass Plan(BaseModel):steps: List[str] = Field(description="需要执行的不同步骤，应该按照顺序排列")

5. 构建规划器(Planner)

规划器的作用是分析用户输入，生成解决问题的步骤计划：

planner_prompt = ChatPromptTemplate.from_messages([("system","""对于给定的目标，提出一个简单的逐步计划。这个计划应该包含独立的任务，如果正确执行将得出正确答案。不要添加任何多余的步骤。最后一步的结果应该是最终答案。确保每一步都有所有必要的信息 -不要跳过步骤"""),("placeholder", "{messages}"),]
)
planner = planner_prompt | llm.with_structured_output(Plan)

6. 构建重新规划器(Replanner)

重新规划器会根据已执行的步骤，决定是继续执行、重新规划，还是直接回答：

class Response(BaseModel):"""用户响应"""response: strclass Act(BaseModel):"""要执行的行为"""action: Union[Response, Plan] = Field(description="要执行的行为。如果要回应用户，使用Response。如果需要进一步使用工具获取答案，使用Plan。")replanner_prompt = ChatPromptTemplate.from_template("""对于给定的目标，提出一个简单的逐步计划。这个计划应该包含独立的任务，如果正确执行将得出正确的答案。不要添加任何多余的步骤。最后一步的结果应该是最终答案。确保每一步都有所有必要的信息 - 不要跳过步骤。你的目标是：
{input}你的原计划是：
{plan}你目前已完成的步骤是：
{past_steps}相应地更新你的计划。如果不需要更多步骤并且可以返回给用户，那么就这样回应。如果需要，填写计划。只添加仍然需要完成的步骤。不要返回已完成的步骤作为计划的一部分。"""
)replanner = replanner_prompt | llm.with_structured_output(Act)

7. 构建工作流图

现在，我们可以构建工作流图，定义智能体的工作流程：

async def main():# 定义规划步骤函数async def plan_step(state: PlanExecute):plan = await planner.ainvoke({"messages": [("user", state["input"])]})return {"plan": plan.steps}# 定义执行步骤函数async def execute_step(state: PlanExecute):plan = state["plan"]plan_str = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(plan))task = plan[0]task_formatted = f"""对于以下计划：
{plan_str}\n\n你的任务是执行第{1}步，{task}。"""agent_response = await agent_executor.ainvoke({"messages": [("user", task_formatted)]})return {"past_steps": state["past_steps"] + [(task, agent_response["messages"][-1].content)],}# 定义重新规划步骤函数async def replan_step(state: PlanExecute):output = await replanner.ainvoke(state)if isinstance(output.action, Response):return {"response": output.action.response}else:return {"plan": output.action.steps}# 定义判断结束条件的函数def should_end(state: PlanExecute) -> Literal["agent", "__end__"]:if "response" in state and state["response"]:return "__end__"else:return "agent"# 创建状态图from langgraph.graph import StateGraph, STARTworkflow = StateGraph(PlanExecute)# 添加节点workflow.add_node("planner", plan_step)workflow.add_node("agent", execute_step)workflow.add_node("replan", replan_step)# 设置边workflow.add_edge(START, "planner")workflow.add_edge("planner", "agent")workflow.add_edge("agent", "replan")workflow.add_conditional_edges("replan",should_end,)# 编译状态图app = workflow.compile()# 将图可视化并保存graph_png = app.get_graph().draw_mermaid_png()with open("agent_workflow.png", "wb") as f:f.write(graph_png)# 执行工作流config = {"recursion_limit": 50}inputs = {"input": "2024年巴黎奥运会100米自由泳决赛冠军的家乡是哪里?请用中文答复"}async for event in app.astream(inputs, config=config):for k, v in event.items():if k != "__end__":print(v)# 运行主函数
asyncio.run(main())

8. 工作流程解析

让我们详细解析这个规划执行型智能体的工作流程：

流程概述

初始规划：接收用户输入，生成解决问题的步骤计划
步骤执行：执行计划中的第一个步骤
重新规划：根据执行结果，决定接下来的操作：
- 如果问题已解决，返回最终答案
- 如果需要继续执行步骤，更新计划并执行下一步

节点功能详解

planner节点：
- 输入：用户的查询问题
- 处理：分析问题，生成步骤计划
- 输出：一系列有序的步骤
agent节点：
- 输入：当前计划和要执行的步骤
- 处理：使用ReAct代理执行当前步骤
- 输出：步骤执行的结果
replan节点：
- 输入：原始计划、已执行的步骤和结果
- 处理：决定是继续执行、重新规划，还是直接回答
- 输出：更新的计划或最终答案

状态管理

整个过程中，PlanExecute状态被用来管理：

input：用户的原始输入
plan：当前的步骤计划
past_steps：已执行的步骤和结果
response：最终的回答（如果有）

9. 示例解析

以"2024年巴黎奥运会100米自由泳决赛冠军的家乡是哪里?"为例，智能体的处理流程如下：

规划阶段：
- 步骤1：查询2024年巴黎奥运会100米自由泳决赛冠军是谁
- 步骤2：查询该冠军的家乡信息
执行第一步：
- 智能体使用Tavily搜索工具查询2024年巴黎奥运会100米自由泳冠军
- 获取结果：例如"潘浩东是2024年巴黎奥运会100米自由泳决赛冠军"
重新规划：
- 根据第一步的结果，更新计划：查询潘浩东的家乡信息
执行第二步：
- 智能体再次使用搜索工具查询潘浩东的家乡
- 获取结果：例如"潘浩东的家乡是浙江省温州市"
再次重新规划：
- 判断所有必要信息已获取，生成最终回答
生成最终回答：
- “2024年巴黎奥运会100米自由泳决赛冠军潘浩东的家乡是浙江省温州市。”