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

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

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

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

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

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

2. 环境准备与依赖导入

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

import os
from typing import TypedDict, Annotated, List, Tuple

from dotenv import load_dotenv
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplate

load_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, Field

class 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: str

class 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, START

    workflow = 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. 工作流程解析

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

流程概述

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

节点功能详解

1. planner节点：

   • 输入：用户的查询问题
   • 处理：分析问题，生成步骤计划
   • 输出：一系列有序的步骤

2. agent节点：

   • 输入：当前计划和要执行的步骤
   • 处理：使用ReAct代理执行当前步骤
   • 输出：步骤执行的结果

3. replan节点：

   • 输入：原始计划、已执行的步骤和结果
   • 处理：决定是继续执行、重新规划，还是直接回答
   • 输出：更新的计划或最终答案

状态管理

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

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

9. 示例解析

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

1. 规划阶段：

   • 步骤1：查询2024年巴黎奥运会100米自由泳决赛冠军是谁
   • 步骤2：查询该冠军的家乡信息

2. 执行第一步：

   • 智能体使用Tavily搜索工具查询2024年巴黎奥运会100米自由泳冠军
   • 获取结果：例如"潘浩东是2024年巴黎奥运会100米自由泳决赛冠军"

3. 重新规划：

   • 根据第一步的结果，更新计划：查询潘浩东的家乡信息

4. 执行第二步：

   • 智能体再次使用搜索工具查询潘浩东的家乡
   • 获取结果：例如"潘浩东的家乡是浙江省温州市"

5. 再次重新规划：

   • 判断所有必要信息已获取，生成最终回答

6. 生成最终回答：

   • “2024年巴黎奥运会100米自由泳决赛冠军潘浩东的家乡是浙江省温州市。”

10. LangGraph的优势分析

在构建规划执行型智能体时，LangGraph框架展现了诸多优势：

1. 状态管理：提供了方便的状态管理机制，便于在各步骤间传递信息
2. 模块化设计：将规划、执行、重新规划等功能分解为独立节点，提高了代码的可维护性
3. 流程控制：通过条件边实现了动态决策，使智能体能够根据情况调整工作流程
4. 异步支持：支持异步操作，提高了处理效率
5. 可视化：可以生成工作流程图，便于理解和调试系统

11. 结语

本文介绍了如何使用LangGraph框架构建一个具有规划执行能力的智能体系统。这种模式非常适合处理需要多步骤分析和推理的复杂问题。通过将问题分解为可管理的步骤，智能体可以有条不紊地解决问题，并且能够根据执行过程中获取的新信息动态调整计划。

规划执行型智能体展示了LLM应用的一个重要发展趋势：从简单的问答模式逐渐向具有复杂推理、规划和执行能力的智能系统演进。随着技术的发展，我们可以期待这类智能体在更复杂的场景下发挥作用。

您可以根据自己的需求，扩展本文的示例代码，添加更多工具、调整规划策略，或者优化执行流程，构建更强大的AI应用。