在人工智能领域，智能代理（agent）的设计愈发重要。尤其在面对复杂任务时，规划（planning）技术成为智能代理的核心能力之一。Langgraph 是一个专为构建智能代理而设计的框架，结合了自然语言处理（NLP）与规划技术，旨在帮助开发者快速构建具有语言理解能力和决策能力的智能代理。本文将深入探讨 Langgraph 中的 planning agent，包括其设计理念、核心功能、实际应用及详细的代码示例，帮助开发者全面了解这一强大的工具。
目录
1. Langgraph 简介
1.1 什么是 Langgraph
1.1.1 与其他智能代理框架的比较
1.2 Langgraph 的核心组件
1.2.1 语言理解模块
代码示例：基本的语言理解模型配置
1.2.2 知识库
代码示例：如何创建和查询知识库
1.2.3 规划模块
代码示例：简单的计划生成算法
2. Planning Agent 概述
2.1 什么是 Planning Agent
2.2 Planning Agent 的工作流程
2.2.1 目标识别
代码示例：用户输入的处理
2.2.2 环境感知
代码示例：获取和处理环境数据
2.2.3 计划生成
代码示例：计划生成算法的实现
2.2.4 执行与反馈
代码示例：动态调整计划的实现
3. Langgraph 中的 Planning Agent
3.1 规划模块设计
3.1.1 状态表示
3.1.2 计划生成的核心算法
代码示例：计划生成算法的实现
3.2 规划模块的实现
4. 实际应用案例
4.1 任务管理助手
用户输入示例
代理处理过程
代码示例：任务管理助手
5.完整代码示例
5.1 项目架构
5.2 .env文件配置
5.3 agent.py
5.4 graph.py
5.5 plan.py
5.6 运行
6. 未来发展方向
结论
1. Langgraph 简介
1.1 什么是 Langgraph
Langgraph 是一个集成化的框架，旨在简化智能代理的开发过程。它通过提供语言理解、知识管理和规划能力，使开发者能够创建能够与用户进行自然语言交流的智能代理。Langgraph 采用模块化设计，使得每个功能组件都可以独立配置和扩展。
1.1.1 与其他智能代理框架的比较
与其他框架（如 Rasa、Dialogflow）相比，Langgraph 强调规划能力的集成。这意味着在 Langgraph 中，开发者不仅可以处理用户的输入，还能为代理生成复杂的行动计划。例如，在 Rasa 中，虽然可以处理对话流，但在执行多个任务时，往往需要额外的逻辑支持。而 Langgraph 的规划模块能够动态生成行动计划，使得智能代理在面对复杂场景时表现得更为灵活。
1.2 Langgraph 的核心组件
Langgraph 的核心组件包括语言理解模块、知识库和规划模块。
1.2.1 语言理解模块
语言理解模块负责解析用户的自然语言输入，将其转化为可供计算机处理的结构化信息。这一模块使用先进的自然语言处理技术，如意图识别和实体抽取，帮助代理理解用户的需求。
功能与实现方式
- 意图识别：识别用户希望完成的任务，例如“预定酒店”或“查询天气”。
- 实体抽取：从用户的输入中提取关键信息，例如时间、地点、数量等。
代码示例：基本的语言理解模型配置
from langgraph.nlp import LanguageUnderstandingModel
# 创建语言理解模型
lu_model = LanguageUnderstandingModel()
# 配置模型参数
lu_model.set_intent_recognition(['book_hotel', 'check_weather'])
lu_model.set_entity_extraction(['location', 'date', 'time'])
# 示例输入
user_input = "我想在明天预定一家酒店"
parsed_result = lu_model.parse(user_input)
print(parsed_result) # 输出：{'intent': 'book_hotel', 'entities': {'date': '明天'}}
1.2.2 知识库
知识库用于存储和管理智能代理所需的知识信息。通过知识图谱的形式，代理能够快速查询相关信息并为用户提供准确的答案。
知识库的结构与管理
- 知识存储：以图结构存储信息，使得关系查询更加高效。
- 查询接口：提供灵活的查询接口，支持复杂的知识检索。
代码示例：如何创建和查询知识库
from langgraph.knowledge import KnowledgeBase
# 创建知识库
knowledge_base = KnowledgeBase()
# 添加知识
knowledge_base.add_fact('hotel_in_paris', {'name': '巴黎酒店', 'location': '巴黎', 'rating': 5})
knowledge_base.add_fact('weather_in_paris', {'condition': '晴', 'temperature': '25°C'})
# 查询知识
hotel_info = knowledge_base.query('hotel_in_paris')
print(hotel_info) # 输出：{'name': '巴黎酒店', 'location': '巴黎', 'rating': 5}
1.2.3 规划模块
规划模块是 Langgraph 的核心，负责根据用户的目标生成相应的行动计划。它结合了多种规划算法，使代理能够根据环境变化实时调整计划。
规划的基本原理
- 状态表示：定义环境状态和目标的表示方式。
- 计划生成：根据当前状态和目标生成行动计划。
代码示例：简单的计划生成算法
from langgraph.planning import PlanningAgent
# 创建规划代理
planning_agent = PlanningAgent()
# 设置初始状态和目标
initial_state = {'user_intent': 'book_hotel', 'available_hotels': ['巴黎酒店', '伦敦酒店']}
goal_state = {'action': 'book', 'hotel': '巴黎酒店'}
# 生成计划
plan = planning_agent.generate_plan(initial_state, goal_state)
print(plan) # 输出：['查询酒店', '确认预定', '发送确认信息']
2. Planning Agent 概述
2.1 什么是 Planning Agent
Planning agent 是一种智能代理，能够根据特定目标生成可执行的行动计划。与传统的基于规则的代理相比，planning agent 具备更高的灵活性和适应性，能够处理复杂的决策和动态环境。
2.2 Planning Agent 的工作流程
Planning agent 的工作流程可以分为以下几个步骤：
- 目标识别：解析用户输入，识别需要达成的目标。
- 环境感知：收集当前环境状态的信息，包括可用资源和限制条件。
- 计划生成：基于目标和环境状态，生成可行的行动计划。
- 执行与反馈：执行计划并根据执行结果进行调整。
2.2.1 目标识别
在目标识别阶段，planning agent 解析用户输入，确定用户的意图和期望的目标。这一过程依赖于语言理解模块的准确性。
代码示例：用户输入的处理
user_input = "我想在周末去巴黎旅行"
parsed_input = lu_model.parse(user_input)
# 从解析结果中提取意图
user_intent = parsed_input['intent']
print(user_intent) # 输出：'book_travel'
2.2.2 环境感知
环境感知阶段，planning agent 收集当前的环境状态信息。这包括可用的资源、外部条件和限制。
代码示例：获取和处理环境数据
# 假设环境数据包含可用酒店和天气信息
environment_data = {
'available_hotels': ['巴黎酒店', '伦敦酒店'],
'weather': {'condition': '晴', 'temperature': '25°C'}
}
# 代理收集环境状态
planning_agent.set_environment(environment_data)
2.2.3 计划生成
计划生成是 planning agent 的核心任务。基于目标和当前环境状态，生成一系列可行的行动计划。
代码示例：计划生成算法的实现
# 生成计划
goal_state = {'action': 'book', 'destination': '巴黎'}
plan = planning_agent.generate_plan(environment_data, goal_state)
print(plan) # 输出：['查询可用酒店', '选择酒店', '确认预定']
2.2.4 执行与反馈
执行阶段，planning agent 根据生成的计划逐步执行，并根据反馈调整计划。例如，如果用户选择了不同的酒店，代理需要实时更新计划。
代码示例：动态调整计划的实现
# 执行第一个行动
action = plan[0] # '查询可用酒店'
response = execute_action(action) # 假设这是执行行动的函数
# 根据用户的反馈调整计划
if response == '用户选择了不同的酒店':
plan = planning_agent.update_plan(new_hotel_choice='伦敦酒店')
3. Langgraph 中的 Planning Agent
3.1 规划模块设计
Langgraph 的规划模块使用多种算法来生成和优化计划。以下是其主要设计思路。
3.1.1 状态表示
规划模块首先需要定义状态表示。状态可以是一个复杂的数据结构，包含当前任务的所有相关信息，例如用户输入、当前环境状态等。
3.1.2 计划生成的核心算法
计划生成采用启发式搜索和图搜索等方法，使得生成的计划更加高效。启发式算法能够通过评估可能的行动，优先考虑最有希望的路径，而图搜索则能够全面探索状态空间。
代码示例：计划生成算法的实现
以下是一个简单的计划生成算法的代码示例：
class PlanningAgent:
def __init__(self, initial_state, goal_state):
self.initial_state = initial_state
self.goal_state = goal_state
def heuristic(self, state):
# 计算当前状态与目标状态的差异
return abs(state - self.goal_state)
def a_star_search(self):
open_set = {self.initial_state}
came_from = {}
g_score = {self.initial_state: 0}
f_score = {self.initial_state: self.heuristic(self.initial_state)}
while open_set:
current = min(open_set, key=lambda x: f_score.get(x, float('inf')))
if current == self.goal_state:
return self.reconstruct_path(came_from, current)
open_set.remove(current)
for neighbor in self.get_neighbors(current):
tentative_g_score = g_score[current] + self.cost(current, neighbor)
if tentative_g_score < g_score.get(neighbor, float('inf')):
came_from[neighbor] = current
g_score[neighbor] = tentative_g_score
f_score[neighbor] = g_score[neighbor] + self.heuristic(neighbor)
open_set.add(neighbor)
return None # 如果没有找到路径
def get_neighbors(self, state):
# 返回当前状态的邻居状态
# 这部分逻辑依赖于具体应用
pass
def cost(self, current, neighbor):
# 计算从当前状态到邻居状态的成本
return 1 # 简化示例，假设每个步骤成本相同
def reconstruct_path(self, came_from, current):
total_path = [current]
while current in came_from:
current = came_from[current]
total_path.append(current)
return total_path[::-1] # 反转路径
3.2 规划模块的实现
规划模块的实现包含多个步骤，从定义状态到执行计划。开发者可以根据具体需求，定制状态表示、成本函数和邻居生成逻辑。
4. 实际应用案例
4.1 任务管理助手
一个常见的应用场景是任务管理助手。用户可以通过自然语言与代理交互，要求代理为他们安排会议、提醒待办事项等。
用户输入示例
- “帮我安排一个明天的会议。”
- “提醒我下午三点开会。”
代理处理过程
- 目标识别：解析用户请求，识别出“安排会议”和“设置提醒”。
- 环境感知：查询用户日历，确定可用时间。
- 计划生成：根据用户的需求生成行动计划，例如在日历中添加事件。
- 执行与反馈：执行计划，并向用户反馈结果。
代码示例：任务管理助手
以下是任务管理助手的简单实现：
class TaskManagerAgent(PlanningAgent):
def __init__(self, initial_state, goal_state, calendar):
super().__init__(initial_state, goal_state)
self.calendar = calendar # 用户的日历
def identify_goal(self, user_input):
# 识别用户的目标
if "安排会议" in user_input:
return "schedule_meeting"
elif "提醒我" in user_input:
return "set_reminder"
return None
def execute_task(self, task):
if task == "schedule_meeting":
# 具体的安排逻辑
print("会议已安排在：", self.goal_state)
elif task == "set_reminder":
print("提醒已设置在：", self.goal_state)
# 示例用法
calendar = {} # 用户的日历
task_manager = TaskManagerAgent(initial_state=None, goal_state="2024-09-30 10:00", calendar=calendar)
task_manager.execute_task(task_manager.identify_goal("帮我安排一个明天的会议。"))
5.完整代码示例
根据官方博客：Plan-and-Execute (langchain-ai.github.io)
以及GitHub代码地址：主要的 langgraph/docs/docs/tutorials/plan-and-execute/plan-and-execute.ipynb ·langchain-ai/语言图 ·GitHub的
完整的运行文件如下：
5.1 项目架构
5.2 .env文件配置
OPENAI_API_KEY=sk-proj-tfV9dwAWhKhtI7pjT2U3T3BlbkFJvnYXOW8AHhh3cVgO370I
TAVILY_API_KEY=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
LANGSMITH_API_KEY=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其中第二个是Tavily搜索的api key，需要创建账号并自己申请；第三个是检控LangChain运行的api key，课查看官方博客了解具体作用。
5.3 agent.py
import os # 导入操作系统模块
from langchain_core.messages import HumanMessage # 从 langchain_core 导入 HumanMessage 类
# 设置 API 密钥
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
from langchain import hub # 从 langchain 导入 hub 模块
from langchain_openai import ChatOpenAI # 从 langchain_openai 导入 ChatOpenAI 类
from langgraph.prebuilt import create_react_agent # 从 langgraph.prebuilt 导入 create_react_agent 函数
from langchain_community.tools.tavily_search import TavilySearchResults # 从 langchain_community.tools 导入 TavilySearchResults 类
# 创建 LLM 实例
llm = ChatOpenAI(model="gpt-4-turbo-preview") # 实例化 ChatOpenAI，使用 gpt-4-turbo-preview 模型
# 定义工具，使用 Tavily 搜索工具
tools = [TavilySearchResults(max_results=3)] # 设置最大结果为 3
search_agent_executor = create_react_agent(llm, tools, state_modifier="你是一个有用的Agent,可以在线搜索") # 创建代理执行器
# 可选：调用代理并打印响应
# response = search_agent_executor.invoke({"messages": [("user", "who is the winner of the us open")]}) # 调用代理，询问 US Open 胜者
# print(response) # 打印响应
5.4 graph.py
import asyncio
from typing import Literal
from agent import search_agent_executor
from plan import PlanExecute, replanner, Response, planner
from langgraph.graph import StateGraph, START
# 设置 API 密钥
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
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"""For the following plan:
{plan_str}\n\nYou are tasked with executing step {1}, {task}.""" # 构造执行任务的提示
agent_response = await search_agent_executor.ainvoke(
{"messages": [("user", task_formatted)]} # 发送消息给代理执行任务
)
return {
"past_steps": [(task, agent_response["messages"][-1].content)], # 返回过去的步骤
}
async def plan_step(state: PlanExecute):
plan = await planner.ainvoke({"messages": [("user", state["input"])]}) # 调用 planner 获取计划
return {"plan": plan.steps} # 返回计划步骤
async def replan_step(state: PlanExecute):
output = await replanner.ainvoke(state) # 调用 replanner 进行重新规划
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" # 否则继续代理
# 创建状态图
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) # 添加条件边，判断是否结束
# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
app = workflow.compile()
# 可视化状态图
img_path = "graph.png"
graph_image = app.get_graph(xray=True).draw_mermaid_png() # 直接获取图像数据
with open(img_path, "wb") as img_file:
img_file.write(graph_image) # 保存为文件
from IPython.display import Image, display
display(Image(img_path)) # 读取并显示文件
# 运行应用程序
config = {"recursion_limit": 50}
inputs = {"input": "what is the hometown of the mens 2024 Australia open winner?"}
# 处理事件流
async def main():
async for event in app.astream(inputs, config=config):
for k, v in event.items():
if k != "__end__":
print(v)
# 运行主程序
if __name__ == "__main__":
asyncio.run(main())
5.5 plan.py
# 导入所需的库
import operator
import os
from typing import Annotated, List, Tuple, TypedDict
from langchain_openai import ChatOpenAI # 导入OpenAI的聊天模型
# 设置 API 密钥
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
# 定义状态类
class PlanExecute(TypedDict):
input: str # 用户输入
plan: List[str] # 计划的步骤列表
past_steps: Annotated[List[Tuple], operator.add] # 已执行的步骤
response: str # 最终响应
from pydantic import BaseModel, Field # 导入Pydantic用于数据模型
# 定义计划模型
class Plan(BaseModel):
"""Plan to follow in future""" # 用于未来执行的计划
steps: List[str] = Field(
description="different steps to follow, should be in sorted order" # 描述计划的步骤，应该是有序的
)
from langchain_core.prompts import ChatPromptTemplate # 导入提示模板
# 定义规划提示模板
planner_prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"""For the given objective, come up with a simple step by step plan. \
This plan should involve individual tasks, that if executed correctly will yield the correct answer. Do not add any superfluous steps. \
The result of the final step should be the final answer. Make sure that each step has all the information needed - do not skip steps.""",
),
("placeholder", "{messages}"),
]
)
# 创建规划实例
planner = planner_prompt | ChatOpenAI(
model="gpt-4o", temperature=0 # 使用OpenAI的GPT模型，设置温度为0以确保一致性
).with_structured_output(Plan)
# 调用规划器生成计划
planner.invoke(
{
"messages": [
("user", "what is the hometown of the current Australia open winner?") # 用户询问的问题
]
}
)
# 重新规划部分
from typing import Union # 导入Union类型
# 定义响应类
class Response(BaseModel):
"""Response to user.""" # 响应用户的类
response: str # 响应内容
# 定义动作类
class Act(BaseModel):
"""Action to perform.""" # 要执行的动作类
action: Union[Response, Plan] = Field(
description="Action to perform. If you want to respond to user, use Response. "
"If you need to further use tools to get the answer, use Plan." # 描述动作的类型
)
# 定义重新规划的提示模板
replanner_prompt = ChatPromptTemplate.from_template(
"""For the given objective, come up with a simple step by step plan. \
This plan should involve individual tasks, that if executed correctly will yield the correct answer. Do not add any superfluous steps. \
The result of the final step should be the final answer. Make sure that each step has all the information needed - do not skip steps.
Your objective was this:
{input}
Your original plan was this:
{plan}
You have currently done the follow steps:
{past_steps}
Update your plan accordingly. If no more steps are needed and you can return to the user, then respond with that. Otherwise, fill out the plan. Only add steps to the plan that still NEED to be done. Do not return previously done steps as part of the plan."""
)
replanner = replanner_prompt | ChatOpenAI(
model="gpt-4o", temperature=0
).with_structured_output(Act)5.6 运行
直接运行graph.py文件就行啦，更改输入也是在graph.py文件中~
6. 未来发展方向
Langgraph 提供了一个强大的框架，用于构建具有规划能力的智能代理。通过灵活的设计和强大的功能，开发者能够快速实现复杂的任务规划。本文详细探讨了 planning agent 的概念、构建过程和实际应用，旨在帮助开发者深入理解这一技术。随着人工智能的持续发展，Langgraph 和 planning agent 也将迎来更多可能，值得持续关注和研究。
结论
Langgraph 的 planning agent 提供了一种强大的方式来构建智能代理。通过有效的规划和动态调整，智能代理能够在复杂的任务中表现出色。本文详细介绍了 Langgraph 的核心组件、规划模块的设计以及实际应用案例，为开发者在构建智能代理时提供了实用的参考。
