LangGraph reference 文档

作者：E的工程笔记

---

Example

1. from langgraph.prebuilt import create_react_agent
3. defcheck_weather(location:str)->str:'''Return the weather forecast for the specified location.'''returnf"It's always sunny in {location}"
5. graph = create_react_agent("anthropic:claude-3-7-sonnet-latest",
6.     tools=[check_weather],
7.     prompt="You are a helpful assistant",)
8. inputs ={"messages":[{"role":"user","content":"what is the weather in sf"}]}for chunk in graph.stream(inputs, stream_mode="updates"):print(chunk)

复制代码

---

ToolNode

基类：RunnableCallable
一个运行最后一个 AIMessage 中调用的工具的节点。
它可以在 StateGraph 中使用 “messages” 状态键（或通过 ToolNode 的 ‘messages_key’ 传递的自定义键）。
如果请求了多个工具调用，它们将并行运行。输出将是一个 ToolMessages 列表，每个工具调用一个。
工具调用也可以直接作为 ToolCall 字典列表传递。
参数：

名称	类型	描述	默认值
tools	Sequence[Union[BaseTool, Callable]]	ToolNode 可以调用的工具序列。	必需
name	str	图中 ToolNode 的名称。默认为 “tools”。	'tools'
tags	Optional[list[str]]	与节点关联的可选标签。默认为 None。	None
handle_tool_errors	Union[bool, str, Callable[..., str], tuple[type[Exception], ...]]	如何处理节点内工具引发的工具错误。默认为 True。必须是以下之一： * True：所有错误将被捕获，并返回带有默认错误消息（TOOL_CALL_ERROR_TEMPLATE）的 ToolMessage。 * str：所有错误将被捕获，并返回带有 ‘handle_tool_errors’ 字符串值的 ToolMessage。 * tuple[type[Exception], …]：元组中的异常将被捕获，并返回带有默认错误消息（TOOL_CALL_ERROR_TEMPLATE）的 ToolMessage。 * Callable[…, str]：从可调用签名的异常将被捕获，并返回带有 ‘handle_tool_errors’ 可调用结果字符串值的 ToolMessage。 * False：工具引发的所有错误都不会被捕获	True
messages_key	str	输入中包含消息列表的状态键。ToolNode 的输出也将使用相同的键。默认为 “messages”。	'messages'

---

ToolNode 大致相当于：

1. tools_by_name ={tool.name: tool for tool in tools}deftool_node(state:dict):
2.     result =[]for tool_call in state["messages"][-1].tool_calls:
3.         tool = tools_by_name[tool_call["name"]]
4.         observation = tool.invoke(tool_call["args"])
5.         result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))return{"messages": result}

复制代码

---

工具调用也可以直接传递给 ToolNode。这在使用 Send API 时非常有用，例如，在条件边中：

1. defexample_conditional_edge(state:dict)-> List[Send]:
2.     tool_calls = state["messages"][-1].tool_calls
3.     # If tools rely on state or store variables (whose values are not generated# directly by a model), you can inject them into the tool calls.
4.     tool_calls =[
5.         tool_node.inject_tool_args(call, state, store)for call in last_message.tool_calls
6.     ]return[Send("tools",[tool_call])for tool_call in tool_calls]

复制代码

---

LangGraph 监督器

https://langchain-ai.github.io/langgraph/reference/supervisor/
功能：

名称	描述
create_supervisor	创建多智能体监督器。

---

创建管理员

1. create_supervisor(*    agents:list[Pregel],**,*    model: LanguageModelLike,*    tools:(*list[BaseTool | Callable]| ToolNode |None*)=None,*    prompt: Prompt |None=None,*    response_format: Optional[*        Union[*            StructuredResponseSchema,*tuple[str, StructuredResponseSchema],*]*]=None,*    parallel_tool_calls:bool=False,*    state_schema: StateSchemaType = AgentState,*    config_schema: Type[Any]|None=None,*    output_mode: OutputMode ="last_message",*    add_handoff_messages:bool=True,*    handoff_tool_prefix: Optional[str]=None,*    add_handoff_back_messages: Optional[bool]=None,*    supervisor_name:str="supervisor",*    include_agent_name: AgentNameMode |None=None)-> StateGraph

复制代码

---

创建一个多智能体监督器。
参数：

名称	类型	描述	默认值
agents	list[Pregel]	要管理的代理列表。代理可以是 LangGraph CompiledStateGraph，功能 API workflow，或任何其他 Pregel 对象。	必需
model	LanguageModelLike	用于主管的语言模型	必需
tools	`list[BaseTool	Callable]	ToolNode | None`
prompt	Prompt | None	用于主管的可选提示。可以是以下之一： * str：这将被转换为 SystemMessage，并添加到 state[“messages”] 的消息列表的开头。 * SystemMessage：这将被添加到 state[“messages”] 的消息列表的开头。 * Callable：这个函数应该接收完整的图状态，其输出然后传递给语言模型。 * Runnable：这个可运行对象应该接收完整的图状态，其输出然后传递给语言模型。	None
response_format	Optional[Union[StructuredResponseSchema, tuple[str, StructuredResponseSchema]]]	最终主管输出的可选架构。 如果提供，输出将被格式化以匹配给定的架构，并在 ‘structured_response’ 状态键中返回。如果没有提供，输出状态中将不包含 structured_response。可以按以下方式传递： - OpenAI 功能/工具架构， - JSON 架构， - TypedDict 类， - 或 Pydantic 类。 - 元组 (prompt, schema)，其中 schema 是上述之一。 提示将与正在使用的模型一起用于生成结构化响应。 重要 response_format 要求模型支持 .with_structured_output 注意 response_format 要求状态架构中包含 structured_response 键。你可以使用预构建的 langgraph.prebuilt.chat_agent_executor.AgentStateWithStructuredResponse。	None
parallel_tool_calls	bool	是否允许主管 LLM 并行调用工具（仅限 OpenAI 和 Anthropic）。使用此选项控制主管是否可以同时交给多个代理。如果为 True，将启用并行工具调用。如果为 False，将禁用并行工具调用（默认）。 重要 目前仅支持 OpenAI 和 Anthropic 模型。要控制其他提供商的并行工具调用，请在系统提示中添加显式的工具使用说明。	False
state_schema	StateSchemaType	用于主管图的状态架构。	AgentState
config_schema	Type[Any] | None	配置的可选架构。使用此选项可通过 supervisor.config_specs 暴露可配置参数。	None
output_mode	OutputMode	在多代理工作流程中将托管代理的输出添加到消息历史记录的模式。可以是以下之一： * full_history：添加整个代理消息历史记录 * last_message：仅添加最后一条消息（默认）	'last_message'
add_handoff_messages	bool	当发生交接时，是否将一对 (AIMessage, ToolMessage) 添加到消息历史记录中。	True
handoff_tool_prefix	Optional[str]	交接工具的可选前缀（例如，“delegate_to_” 或 “transfer_to_”）。如果提供，交接工具将被命名为 handoff_tool_prefix_agent_name。如果没有提供，交接工具将被命名为 transfer_to_agent_name。	None
add_handoff_back_messages	Optional[bool]	当控制权返回给主管以指示发生了交接时，是否将一对 (AIMessage, ToolMessage) 添加到消息历史记录中。	None
supervisor_name	str	主管节点的名称。	'supervisor'
include_agent_name	AgentNameMode | None	用于指定如何将代理名称暴露给底层主管 LLM。 * None：依赖于 LLM 提供者使用 AI 消息上的名称属性。目前，仅 OpenAI 支持。 * "inline"：使用 XML 风格的标签将代理名称直接添加到 AI 消息的内容字段中。 例如：“How can I help you” -> "<name>agent_name</name><content>How can I help you?</content>"	None

---

Example

1. from langchain_openai import ChatOpenAI
3. from langgraph_supervisor import create_supervisor
4. from langgraph.prebuilt import create_react_agent
6. # Create specialized agentsdefadd(a:float, b:float)->float:*'''Add two numbers.'''*return a + b
8. defweb_search(query:str)->str:*'''Search the web for information.'''*return'Here are the headcounts for each of the FAANG companies in 2024...'
10. math_agent = create_react_agent(*    model="openai:gpt-4o",*    tools=[add],*    name="math_expert",)
12. research_agent = create_react_agent(*    model="openai:gpt-4o",*    tools=[web_search],*    name="research_expert",)# Create supervisor workflow
13. workflow = create_supervisor(*[research_agent, math_agent],*    model=ChatOpenAI(model="gpt-4o"),)# Compile and run
14. app = workflow.compile()
15. result = app.invoke({*"messages":[*{*"role":"user",*"content":"what's the combined headcount of the FAANG companies in 2024?"*}*]})

复制代码

---

函数：

名称	描述
create_handoff_tool	创建一个可以将控制权移交给请求代理的工具。
create_forward_message_tool	创建一个监督者可以使用该工具按名称转发工人消息的工具。

---

create_handoff_tool

1. create_handoff_tool(**,*    agent_name:str,*    name:str|None=None,*    description:str|None=None,*    add_handoff_messages:bool=True)-> BaseTool

复制代码

---

创建一个工具，可以将控制权移交给请求的代理。
参数：

名称	类型	描述	默认值
agent_name	str	要移交控制权的代理的名称，即多代理图中代理节点的名称。代理名称应简单、清晰且唯一，最好使用snake_case，尽管你只能使用LangGraph节点接受的名称以及LLM提供商接受的工具名称（工具名称将如下所示：transfer_to_<agent_name>）。	必填
name	str | None	可选，用于移交的工具名称。如果未提供，工具名称将为 transfer_to_<agent_name>。	None
description	str | None	可选，移交工具的描述。如果未提供，描述将为 Ask agent <agent_name> for help。	None
add_handoff_messages	bool	是否将移交消息添加到消息历史中。如果为False，则移交消息将不会出现在消息历史中。	True

---

create_forward_message_tool

1. create_forward_message_tool(*    supervisor_name:str="supervisor",)-> BaseTool

复制代码

---

创建一个供监督者使用的工具，用于按名称转发工作节点的消息。
这有助于避免监督者向用户重写工作节点查询时造成信息丢失，同时还能节省部分token消耗。
参数：

名称	类型	说明	默认值
supervisor_name	str	监督节点的名称（用于工具命名空间）。	'supervisor'

返回值：

名称	类型	说明
BaseTool	BaseTool	'forward_message’工具实例。

---

LangGraph 集群

https://langchain-ai.github.io/langgraph/reference/swarm/
类：

名称	描述
SwarmState	多智能体集群的状态模式。

函数：

名称	描述
create_swarm	创建一个多智能体集群。
add_active_agent_router	将当前活跃智能体的路由器添加到状态图中。

---

群集状态 (SwarmState)

基类：MessagesState
用于多智能体群集的状态模式。

---

创建集群

1. create_swarm(*    agents:list[Pregel],**,*    default_active_agent:str,*    state_schema: StateSchemaType = SwarmState,*    config_schema: Type[Any]|None=None)-> StateGraph

复制代码

---

创建一个多智能体集群。
参数：

名称	类型	描述	默认值
agents	list[Pregel]	要加入集群的智能体列表。智能体可以是LangGraph的CompiledStateGraph、函数式API的workflow，或其他任何Pregel对象。	必填
default_active_agent	str	默认路由的目标智能体名称（当没有活跃智能体时使用）。	必填
state_schema	StateSchemaType	用于多智能体图的状态模式。	SwarmState
config_schema	Type[Any] | None	可选的配置模式。通过swarm.config_specs暴露可配置参数时使用。	None

返回值：

类型	描述
StateGraph	一个多智能体集群的StateGraph。

示例

1. from langgraph.checkpoint.memory import InMemorySaver
2. from langgraph.prebuilt import create_react_agent
3. from langgraph_swarm import create_handoff_tool, create_swarm
5. defadd(a:int, b:int)->int:*'''Add two numbers'''*return a + b
7. alice = create_react_agent(*"openai:gpt-4o",*[add, create_handoff_tool(agent_name="Bob")],*    prompt="You are Alice, an addition expert.",*    name="Alice",)
9. bob = create_react_agent(*"openai:gpt-4o",*[create_handoff_tool(agent_name="Alice", description="Transfer to Alice, she can help with math")],*    prompt="You are Bob, you speak like a pirate.",*    name="Bob",)
11. checkpointer = InMemorySaver()
12. workflow = create_swarm(*[alice, bob],*    default_active_agent="Alice")
13. app = workflow.compile(checkpointer=checkpointer)
15. config ={"configurable":{"thread_id":"1"}}
16. turn_1 = app.invoke(*{"messages":[{"role":"user","content":"i'd like to speak to Bob"}]},*    config,)
17. turn_2 = app.invoke(*{"messages":[{"role":"user","content":"what's 5 + 7?"}]},*    config,)

复制代码

---

add_active_agent_router

1. add_active_agent_router(*    builder: StateGraph,**,*    route_to:list[str],*    default_active_agent:str)-> StateGraph

复制代码

---

向当前活跃的代理添加路由器到状态图(StateGraph)。
参数：

名称	类型	描述	默认值
builder	StateGraph	要添加路由器的图构建器(StateGraph)。	必填
route_to	list[str]	要路由到的代理(节点)名称列表。	必填
default_active_agent	str	默认路由到的代理名称(当没有代理处于活跃状态时)。	必填

返回值：

类型	描述
StateGraph	添加了路由器的StateGraph。

示例

1. from langgraph.checkpoint.memory import InMemorySaver
2. from langgraph.prebuilt import create_react_agent
3. from langgraph.graph import StateGraph
4. from langgraph_swarm import SwarmState, create_handoff_tool, add_active_agent_router
6. defadd(a:int, b:int)->int:*'''Add two numbers'''*return a + b
8. alice = create_react_agent(*"openai:gpt-4o",*[add, create_handoff_tool(agent_name="Bob")],*    prompt="You are Alice, an addition expert.",*    name="Alice",)
10. bob = create_react_agent(*"openai:gpt-4o",*[create_handoff_tool(agent_name="Alice", description="Transfer to Alice, she can help with math")],*    prompt="You are Bob, you speak like a pirate.",*    name="Bob",)
12. checkpointer = InMemorySaver()
13. workflow =(*    StateGraph(SwarmState)*.add_node(alice, destinations=("Bob",))*.add_node(bob, destinations=("Alice",)))# this is the router that enables us to keep track of the last active agent
14. workflow = add_active_agent_router(*    builder=workflow,*    route_to=["Alice","Bob"],*    default_active_agent="Alice",)# compile the workflow
15. app = workflow.compile(checkpointer=checkpointer)
17. config ={"configurable":{"thread_id":"1"}}
18. turn_1 = app.invoke(*{"messages":[{"role":"user","content":"i'd like to speak to Bob"}]},*    config,)
19. turn_2 = app.invoke(*{"messages":[{"role":"user","content":"what's 5 + 7?"}]},*    config,)

复制代码

---

功能：

名称	描述
create_handoff_tool	创建一个可以将控制权移交给指定代理的工具。

---

创建交接工具

1. create_handoff_tool(**,*    agent_name:str,*    name:str|None=None,*    description:str|None=None)-> BaseTool

复制代码

---

创建一个能够将控制权交接给指定代理的工具。
参数：

名称	类型	描述	默认值
agent_name	str	要交接控制权的代理名称，即多代理图中代理节点的名称。代理名称应简洁、清晰且唯一，建议使用snake_case命名法，但需符合LangGraph节点命名规则以及LLM提供商接受的工具名称规范（工具名称将显示为transfer_to_<agent_name>格式）。	必填
name	str | None	可选参数，指定交接工具的名称。如未提供，工具名称将默认为transfer_to_<agent_name>。	None
description	str | None	可选参数，为交接工具提供描述信息。如未提供，工具描述将默认为向代理<agent_name>请求帮助。	None

---

LangChain 模型上下文协议 (MCP) 适配器

https://langchain-ai.github.io/langgraph/reference/mcp/

---

类说明：

名称	描述
MultiServerMCPClient	用于连接多个 MCP 服务器，并从中加载与 LangChain 兼容的工具、提示词和资源的客户端。

---

MultiServerMCPClient

用于连接多个MCP服务器并从中加载与LangChain兼容的工具、提示和资源的客户端。
方法：

名称	描述
__init__	初始化一个带有MCP服务器连接的MultiServerMCPClient。
session	连接到MCP服务器并初始化会话。
get_tools	从所有连接的服务器获取工具列表。
get_prompt	从指定的MCP服务器获取提示。
get_resources	从指定的MCP服务器获取资源。

---

__init__

1. __init__(*    connections:dict[str, Connection]|None=None,)->None

复制代码

---

初始化一个带有 MCP 服务器连接的多服务器 MCP 客户端。
参数：

名称	类型	描述	默认值
connections	dict[str, Connection] | None	将服务器名称映射到连接配置的字典。如果为 None，则不建立初始连接。	None

示例：基本用法（每次工具调用时启动新会话）

1. from langchain_mcp_adapters.client import MultiServerMCPClient
3. client = MultiServerMCPClient(*{*"math":{*"command":"python",*# Make sure to update to the full absolute path to your math_server.py file*"args":["/path/to/math_server.py"],*"transport":"stdio",*},*"weather":{*# make sure you start your weather server on port 8000*"url":"http://localhost:8000/mcp",*"transport":"streamable_http",*}*})
4. all_tools =await client.get_tools()

复制代码

---

示例：显式启动会话

1. from langchain_mcp_adapters.client import MultiServerMCPClient
2. from langchain_mcp_adapters.tools import load_mcp_tools
4. client = MultiServerMCPClient({...})asyncwith client.session("math")as session:*    tools =await load_mcp_tools(session)

复制代码

---

会话 async

1. session(*    server_name:str,*, auto_initialize:bool=True)-> AsyncIterator[ClientSession]

复制代码

---

连接到 MCP 服务器并初始化会话。
参数：

名称	类型	描述	默认值
server_name	str	用于标识此服务器连接的名称	必填
auto_initialize	bool	是否自动初始化会话	True

可能抛出的异常：

类型	描述
ValueError	如果在连接中找不到服务器名称

生成：

类型	描述
AsyncIterator[ClientSession]	一个已初始化的 ClientSession

---

get_tools async

1. get_tools(**, server_name:str|None=None)->list[BaseTool]

复制代码

---

获取所有连接服务器上的工具列表。
参数：

名称	类型	描述	默认值
server_name	str | None	可选参数，指定要获取工具的服务器名称。如果为None，则返回所有服务器上的所有工具（默认行为）。	None

注意：每次工具调用都会创建一个新会话
返回：

类型	描述
list[BaseTool]	LangChain工具列表

---

get_prompt async

1. get_prompt(*    server_name:str,*    prompt_name:str,**,*    arguments:dict[str, Any]|None=None)->list[HumanMessage | AIMessage]

复制代码

---

get_resources async

1. get_resources(*    server_name:str,*, uris:str|list[str]|None=None)->list[Blob]

复制代码

---

从指定的 MCP 服务器获取资源。
参数：

名称	类型	描述	默认值
server_name	str	要获取资源的服务器名称	必填
uris	`str	list[str] | None`	可选资源 URI 或 URI 列表。如果未提供，将加载所有资源。

返回值：

类型	描述
list[Blob]	LangChain Blob 对象列表

函数：

名称	描述
load_mcp_tools	加载所有可用的 MCP 工具并将其转换为 LangChain 工具。

---

加载MCP工具 async

1. load_mcp_tools(*    session: ClientSession |None,**,*    connection: Connection |None=None)->list[BaseTool]

复制代码

加载所有可用的 MCP 工具并将其转换为 LangChain 工具。
参数：

名称	类型	描述	默认值
session	ClientSession | None	MCP 客户端会话	必填
connection	Connection | None	当未提供 session 时，用于创建新会话的可选连接配置	None

---

返回：

类型	描述
list[BaseTool]	LangChain 工具列表

函数：

名称	描述
load_mcp_prompt	加载 MCP 提示并转换为 LangChain 消息。

1. load_mcp_prompt(*    session: ClientSession,*    name:str,**,*    arguments:dict[str, Any]|None=None)->list[HumanMessage | AIMessage]

复制代码

加载 MCP 提示并转换为 LangChain 消息。
功能列表：

名称	描述
load_mcp_resources	加载 MCP 资源并将其转换为 LangChain Blobs。

---

load_mcp_resources async

1. load_mcp_resources(*    session: ClientSession,**,*    uris:str|list[str]|None=None)->list[Blob]

复制代码

加载 MCP 资源并将其转换为 LangChain Blobs。
参数：

名称	类型	描述	默认值
session	ClientSession	MCP 客户端会话	必填
uris	`str	list[str] | None`	要加载的 URI 列表。如果为 None，将加载所有资源。注意：如果指定为 None，则不会加载动态资源，因为它们需要提供参数，且会被 MCP SDK 的 session.list_resources() 方法忽略。

返回：

类型	描述
list[Blob]	LangChain Blobs 列表

---

2025-05-25®



原文地址：https://blog.csdn.net/lovechris00/article/details/148201008