AI创想

标题: LangGraph reference 文档 [打印本页]
作者: 米落枫    时间: 昨天 19:09
标题: 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 字典列表传递。
参数：
名称类型描述默认值
toolsSequence[Union[BaseTool, Callable]]ToolNode 可以调用的工具序列。必需
namestr图中 ToolNode 的名称。默认为 “tools”。'tools'
tagsOptional[list[str]]与节点关联的可选标签。默认为 None。None
handle_tool_errorsUnion[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_keystr输入中包含消息列表的状态键。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
复制代码
创建一个多智能体监督器。
参数：
名称类型描述默认值
agentslist[Pregel]要管理的代理列表。代理可以是 LangGraph CompiledStateGraph，功能 API workflow，或任何其他 Pregel 对象。必需
modelLanguageModelLike用于主管的语言模型必需
tools`list[BaseToolCallable]ToolNode | None`
promptPrompt | None用于主管的可选提示。可以是以下之一： * str：这将被转换为 SystemMessage，并添加到 state[“messages”] 的消息列表的开头。 * SystemMessage：这将被添加到 state[“messages”] 的消息列表的开头。 * Callable：这个函数应该接收完整的图状态，其输出然后传递给语言模型。 * Runnable：这个可运行对象应该接收完整的图状态，其输出然后传递给语言模型。None
response_formatOptional[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_callsbool是否允许主管 LLM 并行调用工具（仅限 OpenAI 和 Anthropic）。使用此选项控制主管是否可以同时交给多个代理。如果为 True，将启用并行工具调用。如果为 False，将禁用并行工具调用（默认）。 重要 目前仅支持 OpenAI 和 Anthropic 模型。要控制其他提供商的并行工具调用，请在系统提示中添加显式的工具使用说明。False
state_schemaStateSchemaType用于主管图的状态架构。AgentState
config_schemaType[Any] | None配置的可选架构。使用此选项可通过 supervisor.config_specs 暴露可配置参数。None
output_modeOutputMode在多代理工作流程中将托管代理的输出添加到消息历史记录的模式。可以是以下之一： * full_history：添加整个代理消息历史记录 * last_message：仅添加最后一条消息（默认）'last_message'
add_handoff_messagesbool当发生交接时，是否将一对 (AIMessage, ToolMessage) 添加到消息历史记录中。True
handoff_tool_prefixOptional[str]交接工具的可选前缀（例如，“delegate_to_” 或 “transfer_to_”）。如果提供，交接工具将被命名为 handoff_tool_prefix_agent_name。如果没有提供，交接工具将被命名为 transfer_to_agent_name。None
add_handoff_back_messagesOptional[bool]当控制权返回给主管以指示发生了交接时，是否将一对 (AIMessage, ToolMessage) 添加到消息历史记录中。None
supervisor_namestr主管节点的名称。'supervisor'
include_agent_nameAgentNameMode | 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_namestr要移交控制权的代理的名称，即多代理图中代理节点的名称。代理名称应简单、清晰且唯一，最好使用snake_case，尽管你只能使用LangGraph节点接受的名称以及LLM提供商接受的工具名称（工具名称将如下所示：transfer_to_<agent_name>）。必填
namestr | None可选，用于移交的工具名称。如果未提供，工具名称将为 transfer_to_<agent_name>。None
descriptionstr | None可选，移交工具的描述。如果未提供，描述将为 Ask agent <agent_name> for help。None
add_handoff_messagesbool是否将移交消息添加到消息历史中。如果为False，则移交消息将不会出现在消息历史中。True
create_forward_message_tool
  1. create_forward_message_tool(*    supervisor_name:str="supervisor",)-> BaseTool
复制代码
创建一个供监督者使用的工具，用于按名称转发工作节点的消息。
这有助于避免监督者向用户重写工作节点查询时造成信息丢失，同时还能节省部分token消耗。
参数：
名称类型说明默认值
supervisor_namestr监督节点的名称（用于工具命名空间）。'supervisor'
返回值：
名称类型说明
BaseToolBaseTool'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
复制代码
创建一个多智能体集群。
参数：
名称类型描述默认值
agentslist[Pregel]要加入集群的智能体列表。智能体可以是LangGraph的CompiledStateGraph、函数式API的workflow，或其他任何Pregel对象。必填
default_active_agentstr默认路由的目标智能体名称（当没有活跃智能体时使用）。必填
state_schemaStateSchemaType用于多智能体图的状态模式。SwarmState
config_schemaType[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)。
参数：
名称类型描述默认值
builderStateGraph要添加路由器的图构建器(StateGraph)。必填
route_tolist[str]要路由到的代理(节点)名称列表。必填
default_active_agentstr默认路由到的代理名称(当没有代理处于活跃状态时)。必填
返回值：
类型描述
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_namestr要交接控制权的代理名称，即多代理图中代理节点的名称。代理名称应简洁、清晰且唯一，建议使用snake_case命名法，但需符合LangGraph节点命名规则以及LLM提供商接受的工具名称规范（工具名称将显示为transfer_to_<agent_name>格式）。必填
namestr | None可选参数，指定交接工具的名称。如未提供，工具名称将默认为transfer_to_<agent_name>。None
descriptionstr | 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 客户端。
参数：
名称类型描述默认值
connectionsdict[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_namestr用于标识此服务器连接的名称必填
auto_initializebool是否自动初始化会话True
可能抛出的异常：
类型描述
ValueError如果在连接中找不到服务器名称
生成：
类型描述
AsyncIterator[ClientSession]一个已初始化的 ClientSession
get_tools async
  1. get_tools(**, server_name:str|None=None)->list[BaseTool]
复制代码
获取所有连接服务器上的工具列表。
参数：
名称类型描述默认值
server_namestr | 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_namestr要获取资源的服务器名称必填
uris`strlist[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 工具。
参数：
名称类型描述默认值
sessionClientSession | NoneMCP 客户端会话必填
connectionConnection | 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。
参数：
名称类型描述默认值
sessionClientSessionMCP 客户端会话必填
uris`strlist[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



欢迎光临 AI创想 (https://llms-ai.com/) Powered by Discuz! X3.4