LangGraph：构建具有状态的多智能体工作流框架

1. 概述

LangGraph 是一个用于构建具有状态、多参与者应用程序的大语言模型（LLM）的库，专门用于创建智能体（Agent）和多智能体的工作流程。与传统的基于链式调用或无状态调用的 LLM 框架相比，LangGraph 提供了更强大的控制能力，使其能够处理复杂的业务逻辑。

from typing import Annotated, Literal, TypedDict from langchain_core.messages import HumanMessage from langchain_anthropic import ChatAnthropic from langchain_core.tools import tool from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import END, StateGraph, MessagesState from langgraph.prebuilt import ToolNode # 为智能体定义工具 @tool def search(query: str): """调用以浏览网络。""" # 这是一个占位符，模拟搜索行为 if "sf" in query.lower() or "san francisco" in query.lower(): return "现在是 60 度，有雾。" return "现在是 90 度，晴天。" tools = [search] tool_node = ToolNode(tools) # 初始化模型 model = ChatAnthropic(model="claude-3-5-sonnet-20240620", temperature=0).bind_tools(tools) # 定义决定是否继续的函数 def should_continue(state: MessagesState) -> Literal["tools", END]: messages = state['messages'] last_message = messages[-1] # 如果 LLM 进行工具调用，那么我们将路径设置为 "tools" 节点 if last_message.tool_calls: return "tools" # 否则，我们停止（回复用户） return END # 定义调用模型的函数 def call_model(state: MessagesState): messages = state['messages'] response = model.invoke(messages) # 我们返回一个列表，因为这将被添加到现有列表中 return {"messages": [response]} # 定义一个新图 workflow = StateGraph(MessagesState) # 定义我们将循环的两个节点 workflow.add_node("agent", call_model) workflow.add_node("tools", tool_node) # 将入口点设置为 `agent` # 这意味着这个节点是首先被调用的 workflow.set_entry_point("agent") # 我们现在添加一个条件边 workflow.add_conditional_edges( # 首先，我们定义起始节点。我们使用 `agent`。 # 这意味着这些边是在调用 `agent` 节点后执行的。 "agent", # 接下来，我们传入决定下一个被调用节点的函数。 should_continue, ) # 我们现在从 `tools` 到 `agent` 添加一条普通边。 # 这意味着在 `tools` 被调用后，`agent` 节点会接着被调用。 workflow.add_edge("tools", 'agent') # 初始化内存以在图运行之间保存状态 checkpointer = MemorySaver() # 最后，我们编译它！ # 这将其编译为一个 LangChain 可运行体， # 这意味着你可以像使用其他可运行体一样使用它。 # 请注意，我们在编译图时（可选地）传递了内存 app = workflow.compile(checkpointer=checkpointer)

LangGraph：构建具有状态的多智能体工作流框架