如何从旧版 LangChain Agents 迁移到 LangGraph

先决条件

本指南假定您熟悉以下概念

• Agents
• LangGraph
• 工具调用

本文重点介绍如何从旧版 LangChain Agents 迁移到更灵活的 LangGraph Agents。LangChain Agents（特别是 AgentExecutor）具有多个配置参数。在本笔记本中，我们将展示如何使用 create_react_agent 预构建帮助方法将这些参数映射到 LangGraph react agent executor。

注意

在 LangGraph 中，图取代了 LangChain 的 agent executor。它管理 agent 的循环并在其状态中将 scratchpad 作为消息进行跟踪。LangChain “agent” 对应于您提供的提示和 LLM。

先决条件

本操作指南使用 OpenAI 作为 LLM。安装依赖项以运行。

%%capture --no-stderr
%pip install -U langgraph langchain langchain-openai

然后，设置您的 OpenAI API 密钥。

import getpass
import os

if "OPENAI_API_KEY" not in os.environ:
    os.environ["OPENAI_API_KEY"] = getpass.getpass("OpenAI API key:\n")

基本用法

对于工具调用 ReAct 风格 agent 的基本创建和用法，功能是相同的。首先，让我们定义一个模型和工具，然后我们将使用它们来创建一个 agent。

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")


@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    return input + 2


tools = [magic_function]


query = "what is the value of magic_function(3)?"

API 参考：tool | ChatOpenAI

对于 LangChain AgentExecutor，我们定义一个提示，其中包含 agent scratchpad 的占位符。可以按如下方式调用 agent

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant"),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)


agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

agent_executor.invoke({"input": query})

API 参考：AgentExecutor | create_tool_calling_agent | ChatPromptTemplate

{'input': 'what is the value of magic_function(3)?',
 'output': 'The value of `magic_function(3)` is 5.'}

LangGraph 的 react agent executor 管理由消息列表定义的状态。它将继续处理列表，直到 agent 的输出中没有工具调用。为了启动它，我们输入一个消息列表。输出将包含图的整个状态——在本例中为对话历史记录。

from langgraph.prebuilt import create_react_agent

langgraph_agent_executor = create_react_agent(model, tools)


messages = langgraph_agent_executor.invoke({"messages": [("human", query)]})
{
    "input": query,
    "output": messages["messages"][-1].content,
}

API 参考：create_react_agent

{'input': 'what is the value of magic_function(3)?',
 'output': 'The value of `magic_function(3)` is 5.'}

message_history = messages["messages"]

new_query = "Pardon?"

messages = langgraph_agent_executor.invoke(
    {"messages": message_history + [("human", new_query)]}
)
{
    "input": new_query,
    "output": messages["messages"][-1].content,
}

{'input': 'Pardon?',
 'output': 'The result of applying `magic_function` to the input value 3 is 5.'}

提示模板

使用旧版 LangChain Agents，您必须传入提示模板。您可以使用它来控制 agent。

使用 LangGraph react agent executor，默认情况下没有提示。您可以通过以下几种方式实现对 agent 的类似控制

1. 将系统消息作为输入传入
2. 使用系统消息初始化 agent
3. 使用函数初始化 agent，以在传递给模型之前转换图状态中的消息。
4. 使用 Runnable 初始化 agent，以在传递给模型之前转换图状态中的消息。这包括传递提示模板。

让我们看看下面的所有这些。我们将传入自定义指令，以使 agent 以西班牙语回复。

首先，使用 AgentExecutor

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant. Respond only in Spanish."),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)


agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

agent_executor.invoke({"input": query})

{'input': 'what is the value of magic_function(3)?',
 'output': 'El valor de magic_function(3) es 5.'}

现在，让我们将自定义系统消息传递给 react agent executor。

LangGraph 的预构建 create_react_agent 不直接将提示模板作为参数，而是采用 prompt 参数。这会在调用 llm 之前修改图状态，并且可以是以下四个值之一

• 一个 SystemMessage，它被添加到消息列表的开头。
• 一个 string，它被转换为 SystemMessage 并添加到消息列表的开头。
• 一个 Callable，它应该接收完整的图状态。然后将输出传递给语言模型。
• 或一个 Runnable，它应该接收完整的图状态。然后将输出传递给语言模型。

以下是它的实际效果

from langchain_core.messages import SystemMessage
from langgraph.prebuilt import create_react_agent

system_message = "You are a helpful assistant. Respond only in Spanish."
# This could also be a SystemMessage object
# system_message = SystemMessage(content="You are a helpful assistant. Respond only in Spanish.")

langgraph_agent_executor = create_react_agent(model, tools, prompt=system_message)


messages = langgraph_agent_executor.invoke({"messages": [("user", query)]})

API 参考：SystemMessage | create_react_agent

我们还可以传入任意函数或 runnable。此函数/runnable 应接收图状态并输出消息列表。我们可以在此处对消息进行各种类型的任意格式化。在这种情况下，让我们在消息列表的开头添加一个 SystemMessage，并在末尾附加另一个用户消息。

from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.prebuilt import create_react_agent
from langgraph.prebuilt.chat_agent_executor import AgentState

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant. Respond only in Spanish."),
        ("placeholder", "{messages}"),
        ("user", "Also say 'Pandamonium!' after the answer."),
    ]
)

# alternatively, this can be passed as a function, e.g.
# def prompt(state: AgentState):
#     return (
#         [SystemMessage(content="You are a helpful assistant. Respond only in Spanish.")] +
#         state["messages"] +
#         [HumanMessage(content="Also say 'Pandamonium!' after the answer.")]
#     )


langgraph_agent_executor = create_react_agent(model, tools, prompt=prompt)


messages = langgraph_agent_executor.invoke({"messages": [("human", query)]})
print(
    {
        "input": query,
        "output": messages["messages"][-1].content,
    }
)

API 参考：HumanMessage | SystemMessage | create_react_agent

{'input': 'what is the value of magic_function(3)?', 'output': 'El valor de magic_function(3) es 5. ¡Pandamonium!'}

内存

在 LangChain 中

使用 LangChain 的 AgentExecutor，您可以添加聊天 内存，以便它可以进行多轮对话。

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")
memory = InMemoryChatMessageHistory(session_id="test-session")
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        # First put the history
        ("placeholder", "{chat_history}"),
        # Then the new input
        ("human", "{input}"),
        # Finally the scratchpad
        ("placeholder", "{agent_scratchpad}"),
    ]
)


@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    return input + 2


tools = [magic_function]


agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

agent_with_chat_history = RunnableWithMessageHistory(
    agent_executor,
    # This is needed because in most real world scenarios, a session id is needed
    # It isn't really used here because we are using a simple in memory ChatMessageHistory
    lambda session_id: memory,
    input_messages_key="input",
    history_messages_key="chat_history",
)

config = {"configurable": {"session_id": "test-session"}}
print(
    agent_with_chat_history.invoke(
        {"input": "Hi, I'm polly! What's the output of magic_function of 3?"}, config
    )["output"]
)
print("---")
print(agent_with_chat_history.invoke({"input": "Remember my name?"}, config)["output"])
print("---")
print(
    agent_with_chat_history.invoke({"input": "what was that output again?"}, config)[
        "output"
    ]
)

API 参考：AgentExecutor | create_tool_calling_agent | InMemoryChatMessageHistory | ChatPromptTemplate | RunnableWithMessageHistory | tool | ChatOpenAI

The output of the magic function when the input is 3 is 5.
---
Yes, you mentioned your name is Polly.
---
The output of the magic function when the input is 3 is 5.

在 LangGraph 中

内存只是 持久性，也称为 检查点。

向 agent 添加 checkpointer，您就可以免费获得聊天内存。

from langgraph.checkpoint.memory import MemorySaver  # an in-memory checkpointer
from langgraph.prebuilt import create_react_agent

system_message = "You are a helpful assistant."
# This could also be a SystemMessage object
# system_message = SystemMessage(content="You are a helpful assistant. Respond only in Spanish.")

memory = MemorySaver()
langgraph_agent_executor = create_react_agent(
    model, tools, prompt=system_message, checkpointer=memory
)

config = {"configurable": {"thread_id": "test-thread"}}
print(
    langgraph_agent_executor.invoke(
        {
            "messages": [
                ("user", "Hi, I'm polly! What's the output of magic_function of 3?")
            ]
        },
        config,
    )["messages"][-1].content
)
print("---")
print(
    langgraph_agent_executor.invoke(
        {"messages": [("user", "Remember my name?")]}, config
    )["messages"][-1].content
)
print("---")
print(
    langgraph_agent_executor.invoke(
        {"messages": [("user", "what was that output again?")]}, config
    )["messages"][-1].content
)

API 参考：MemorySaver | create_react_agent

The output of the magic function for the input 3 is 5.
---
Yes, you mentioned that your name is Polly.
---
The output of the magic function for the input 3 was 5.

迭代步骤

在 LangChain 中

使用 LangChain 的 AgentExecutor，您可以使用 stream（或异步 astream）方法或 iter 方法迭代步骤。LangGraph 使用 stream 支持逐步迭代

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")


prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)


@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    return input + 2


tools = [magic_function]

agent = create_tool_calling_agent(model, tools, prompt=prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

for step in agent_executor.stream({"input": query}):
    print(step)

API 参考：AgentExecutor | create_tool_calling_agent | ChatPromptTemplate | tool | ChatOpenAI

{'actions': [ToolAgentAction(tool='magic_function', tool_input={'input': 3}, log="\nInvoking: `magic_function` with `{'input': 3}`\n\n\n", message_log=[AIMessageChunk(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls', 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7'}, id='run-7a3a5ada-52ec-4df0-bf7d-81e5051b01b4', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'type': 'tool_call'}], tool_call_chunks=[{'name': 'magic_function', 'args': '{"input":3}', 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'index': 0, 'type': 'tool_call_chunk'}])], tool_call_id='call_yyetzabaDBRX9Ml2KyqfKzZM')], 'messages': [AIMessageChunk(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls', 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7'}, id='run-7a3a5ada-52ec-4df0-bf7d-81e5051b01b4', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'type': 'tool_call'}], tool_call_chunks=[{'name': 'magic_function', 'args': '{"input":3}', 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'index': 0, 'type': 'tool_call_chunk'}])]}
{'steps': [AgentStep(action=ToolAgentAction(tool='magic_function', tool_input={'input': 3}, log="\nInvoking: `magic_function` with `{'input': 3}`\n\n\n", message_log=[AIMessageChunk(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls', 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7'}, id='run-7a3a5ada-52ec-4df0-bf7d-81e5051b01b4', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'type': 'tool_call'}], tool_call_chunks=[{'name': 'magic_function', 'args': '{"input":3}', 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'index': 0, 'type': 'tool_call_chunk'}])], tool_call_id='call_yyetzabaDBRX9Ml2KyqfKzZM'), observation=5)], 'messages': [FunctionMessage(content='5', additional_kwargs={}, response_metadata={}, name='magic_function')]}
{'output': 'The value of `magic_function(3)` is 5.', 'messages': [AIMessage(content='The value of `magic_function(3)` is 5.', additional_kwargs={}, response_metadata={})]}

在 LangGraph 中

在 LangGraph 中，使用 stream 或异步 astream 方法本地处理事情。

from langgraph.prebuilt import create_react_agent
from langgraph.prebuilt.chat_agent_executor import AgentState

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("placeholder", "{messages}"),
    ]
)

langgraph_agent_executor = create_react_agent(model, tools, prompt=prompt)

for step in langgraph_agent_executor.stream(
    {"messages": [("human", query)]}, stream_mode="updates"
):
    print(step)

API 参考：create_react_agent

{'agent': {'messages': [AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_IHTMrjvIHn8gFOX42FstIpr9', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 61, 'total_tokens': 75, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-1a6970da-163a-4e4d-b9b7-7e73b1057f42-0', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_IHTMrjvIHn8gFOX42FstIpr9', 'type': 'tool_call'}], usage_metadata={'input_tokens': 61, 'output_tokens': 14, 'total_tokens': 75, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}})]}}
{'tools': {'messages': [ToolMessage(content='5', name='magic_function', id='51a9d3e4-734d-426f-a5a1-c6597e4efe25', tool_call_id='call_IHTMrjvIHn8gFOX42FstIpr9')]}}
{'agent': {'messages': [AIMessage(content='The value of `magic_function(3)` is 5.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 84, 'total_tokens': 98, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a20a4ee344', 'finish_reason': 'stop', 'logprobs': None}, id='run-73001576-a3dc-4552-8d81-c9ce8aec05b3-0', usage_metadata={'input_tokens': 84, 'output_tokens': 14, 'total_tokens': 98, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}})]}}

return_intermediate_steps

在 LangChain 中

在 AgentExecutor 上设置此参数允许用户访问 intermediate_steps，它将 agent 操作（例如，工具调用）与其结果配对。

agent_executor = AgentExecutor(agent=agent, tools=tools, return_intermediate_steps=True)
result = agent_executor.invoke({"input": query})
print(result["intermediate_steps"])

[(ToolAgentAction(tool='magic_function', tool_input={'input': 3}, log="\nInvoking: `magic_function` with `{'input': 3}`\n\n\n", message_log=[AIMessageChunk(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_njTvl2RsVf4q1aMUxoYnJuK1', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls', 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7'}, id='run-c9dfe3ab-2db6-4592-851e-89e056aeab32', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_njTvl2RsVf4q1aMUxoYnJuK1', 'type': 'tool_call'}], tool_call_chunks=[{'name': 'magic_function', 'args': '{"input":3}', 'id': 'call_njTvl2RsVf4q1aMUxoYnJuK1', 'index': 0, 'type': 'tool_call_chunk'}])], tool_call_id='call_njTvl2RsVf4q1aMUxoYnJuK1'), 5)]

在 LangGraph 中

默认情况下，LangGraph 中的 react agent executor 将所有消息附加到中心状态。因此，只需查看完整状态即可轻松查看任何中间步骤。

from langgraph.prebuilt import create_react_agent

langgraph_agent_executor = create_react_agent(model, tools=tools)

messages = langgraph_agent_executor.invoke({"messages": [("human", query)]})

messages

API 参考：create_react_agent

{'messages': [HumanMessage(content='what is the value of magic_function(3)?', additional_kwargs={}, response_metadata={}, id='1abb52c2-4bc2-4d82-bd32-5a24c3976b0f'),
  AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_XfQD6C7rAalcmicQubkhJVFq', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 55, 'total_tokens': 69, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a20a4ee344', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-34f02786-5b5c-4bb1-bd9e-406c81944a24-0', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_XfQD6C7rAalcmicQubkhJVFq', 'type': 'tool_call'}], usage_metadata={'input_tokens': 55, 'output_tokens': 14, 'total_tokens': 69, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}}),
  ToolMessage(content='5', name='magic_function', id='cbc9fadf-1962-4ed7-b476-348c774652be', tool_call_id='call_XfQD6C7rAalcmicQubkhJVFq'),
  AIMessage(content='The value of `magic_function(3)` is 5.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 78, 'total_tokens': 92, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'stop', 'logprobs': None}, id='run-547e03d2-872d-4008-a38d-b7f739a77df5-0', usage_metadata={'input_tokens': 78, 'output_tokens': 14, 'total_tokens': 92, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}})]}

max_iterations

在 LangChain 中

AgentExecutor 实现了 max_iterations 参数，允许用户中止运行超过指定迭代次数的运行。

@tool
def magic_function(input: str) -> str:
    """Applies a magic function to an input."""
    return "Sorry, there was an error. Please try again."


tools = [magic_function]

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant. Respond only in Spanish."),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)

agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=3,
)

agent_executor.invoke({"input": query})

> Entering new AgentExecutor chain...
Lo siento, no puedo decirte directamente el valor de `magic_function(3)`. Si deseas, puedo usar la función mágica para calcularlo. ¿Te gustaría que lo hiciera?

> Finished chain.

{'input': 'what is the value of magic_function(3)?',
 'output': 'Lo siento, no puedo decirte directamente el valor de `magic_function(3)`. Si deseas, puedo usar la función mágica para calcularlo. ¿Te gustaría que lo hiciera?'}

在 LangGraph 中

在 LangGraph 中，这通过 recursion_limit 配置参数控制。

请注意，在 AgentExecutor 中，“迭代”包括工具调用和执行的完整回合。在 LangGraph 中，每个步骤都会影响递归限制，因此我们需要乘以 2（并加 1）才能获得等效结果。

如果达到递归限制，LangGraph 会引发特定的异常类型，我们可以捕获和管理它，类似于 AgentExecutor。

from langgraph.errors import GraphRecursionError
from langgraph.prebuilt import create_react_agent

RECURSION_LIMIT = 2 * 3 + 1

langgraph_agent_executor = create_react_agent(model, tools=tools)

try:
    for chunk in langgraph_agent_executor.stream(
        {"messages": [("human", query)]},
        {"recursion_limit": RECURSION_LIMIT},
        stream_mode="values",
    ):
        print(chunk["messages"][-1])
except GraphRecursionError:
    print({"input": query, "output": "Agent stopped due to max iterations."})

API 参考：create_react_agent

content='what is the value of magic_function(3)?' additional_kwargs={} response_metadata={} id='c2489fe8-e69c-4163-876d-3cce26b28521'
content='' additional_kwargs={'tool_calls': [{'id': 'call_OyNTcO6SDAvZcBlIEknPRrTR', 'function': {'arguments': '{"input":"3"}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None} response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 55, 'total_tokens': 69, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'tool_calls', 'logprobs': None} id='run-b65504bb-fa23-4f8a-8d6c-7edb6d16e7ff-0' tool_calls=[{'name': 'magic_function', 'args': {'input': '3'}, 'id': 'call_OyNTcO6SDAvZcBlIEknPRrTR', 'type': 'tool_call'}] usage_metadata={'input_tokens': 55, 'output_tokens': 14, 'total_tokens': 69, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}}
content='Sorry, there was an error. Please try again.' name='magic_function' id='f00e0bff-54fe-4726-a1a7-127a59d8f7ed' tool_call_id='call_OyNTcO6SDAvZcBlIEknPRrTR'
content="It seems there was an error when trying to compute the value of the magic function with input 3. Let's try again." additional_kwargs={'tool_calls': [{'id': 'call_Q020rQoJh4cnh8WglIMnDm4z', 'function': {'arguments': '{"input":"3"}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None} response_metadata={'token_usage': {'completion_tokens': 40, 'prompt_tokens': 88, 'total_tokens': 128, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'tool_calls', 'logprobs': None} id='run-556d8cb2-b47a-4826-b17d-b520982c2475-0' tool_calls=[{'name': 'magic_function', 'args': {'input': '3'}, 'id': 'call_Q020rQoJh4cnh8WglIMnDm4z', 'type': 'tool_call'}] usage_metadata={'input_tokens': 88, 'output_tokens': 40, 'total_tokens': 128, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}}
content='Sorry, there was an error. Please try again.' name='magic_function' id='777212cd-8381-44db-9762-3f81951ea73e' tool_call_id='call_Q020rQoJh4cnh8WglIMnDm4z'
content="It seems there is a persistent issue in computing the value of the magic function with the input 3. Unfortunately, I can't provide the value at this time. If you have any other questions or need further assistance, feel free to ask!" additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 49, 'prompt_tokens': 150, 'total_tokens': 199, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'stop', 'logprobs': None} id='run-92ec0b90-bc8e-4851-9139-f1d976145ab7-0' usage_metadata={'input_tokens': 150, 'output_tokens': 49, 'total_tokens': 199, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}}

max_execution_time

在 LangChain 中

AgentExecutor 实现了 max_execution_time 参数，允许用户中止运行超过总时间限制的运行。

import time


@tool
def magic_function(input: str) -> str:
    """Applies a magic function to an input."""
    time.sleep(2.5)
    return "Sorry, there was an error. Please try again."


tools = [magic_function]

agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_execution_time=2,
    verbose=True,
)

agent_executor.invoke({"input": query})

> Entering new AgentExecutor chain...
Lo siento, no tengo la capacidad de evaluar directamente una función llamada "magic_function" con el valor 3. Sin embargo, si me proporcionas más detalles sobre qué hace la función o cómo está definida, podría intentar ayudarte a comprender su comportamiento o resolverlo de otra manera.

> Finished chain.

{'input': 'what is the value of magic_function(3)?',
 'output': 'Lo siento, no tengo la capacidad de evaluar directamente una función llamada "magic_function" con el valor 3. Sin embargo, si me proporcionas más detalles sobre qué hace la función o cómo está definida, podría intentar ayudarte a comprender su comportamiento o resolverlo de otra manera.'}

在 LangGraph 中

使用 LangGraph 的 react agent，您可以在两个级别控制超时。

您可以设置 step_timeout 以限制每个步骤

from langgraph.prebuilt import create_react_agent

langgraph_agent_executor = create_react_agent(model, tools=tools)
# Set the max timeout for each step here
langgraph_agent_executor.step_timeout = 2

try:
    for chunk in langgraph_agent_executor.stream({"messages": [("human", query)]}):
        print(chunk)
        print("------")
except TimeoutError:
    print({"input": query, "output": "Agent stopped due to a step timeout."})

API 参考：create_react_agent

{'agent': {'messages': [AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_UuxSgpGaqzX84sNlKzCVOiRO', 'function': {'arguments': '{"input":"3"}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 55, 'total_tokens': 69, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-24c94cbd-2962-48cf-a447-af888eb6ef86-0', tool_calls=[{'name': 'magic_function', 'args': {'input': '3'}, 'id': 'call_UuxSgpGaqzX84sNlKzCVOiRO', 'type': 'tool_call'}], usage_metadata={'input_tokens': 55, 'output_tokens': 14, 'total_tokens': 69, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}})]}}
------
{'input': 'what is the value of magic_function(3)?', 'output': 'Agent stopped due to a step timeout.'}

为整个运行设置单个最大超时的另一种方法是直接使用 python stdlib asyncio 库。

import asyncio

from langgraph.prebuilt import create_react_agent

langgraph_agent_executor = create_react_agent(model, tools=tools)


async def stream(langgraph_agent_executor, inputs):
    async for chunk in langgraph_agent_executor.astream(
        {"messages": [("human", query)]}
    ):
        print(chunk)
        print("------")


try:
    task = asyncio.create_task(
        stream(langgraph_agent_executor, {"messages": [("human", query)]})
    )
    await asyncio.wait_for(task, timeout=3)
except asyncio.TimeoutError:
    print("Task Cancelled.")

API 参考：create_react_agent

{'agent': {'messages': [AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_km17xvoY7wJ5yNnXhb5V9D3I', 'function': {'arguments': '{"input":"3"}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 55, 'total_tokens': 69, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_45c6de4934', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-b44a04e5-9b68-4020-be36-98de1593eefc-0', tool_calls=[{'name': 'magic_function', 'args': {'input': '3'}, 'id': 'call_km17xvoY7wJ5yNnXhb5V9D3I', 'type': 'tool_call'}], usage_metadata={'input_tokens': 55, 'output_tokens': 14, 'total_tokens': 69, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}})]}}
------
Task Cancelled.

early_stopping_method

在 LangChain 中

使用 LangChain 的 AgentExecutor，您可以配置 early_stopping_method，以返回字符串“Agent 因迭代限制或时间限制而停止。” ("force") 或最后一次提示 LLM 做出响应 ("generate")。

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")


prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)


@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    return "Sorry there was an error, please try again."


tools = [magic_function]

agent = create_tool_calling_agent(model, tools, prompt=prompt)
agent_executor = AgentExecutor(
    agent=agent, tools=tools, early_stopping_method="force", max_iterations=1
)

result = agent_executor.invoke({"input": query})
print("Output with early_stopping_method='force':")
print(result["output"])

API 参考：AgentExecutor | create_tool_calling_agent | ChatPromptTemplate | tool | ChatOpenAI

Output with early_stopping_method='force':
Agent stopped due to max iterations.

在 LangGraph 中

在 LangGraph 中，您可以显式处理 agent 外部的响应行为，因为可以访问完整状态。

from langgraph.errors import GraphRecursionError
from langgraph.prebuilt import create_react_agent

RECURSION_LIMIT = 2 * 1 + 1

langgraph_agent_executor = create_react_agent(model, tools=tools)

try:
    for chunk in langgraph_agent_executor.stream(
        {"messages": [("human", query)]},
        {"recursion_limit": RECURSION_LIMIT},
        stream_mode="values",
    ):
        print(chunk["messages"][-1])
except GraphRecursionError:
    print({"input": query, "output": "Agent stopped due to max iterations."})

API 参考：create_react_agent

content='what is the value of magic_function(3)?' additional_kwargs={} response_metadata={} id='81fd2e50-1e6a-4871-87aa-b7c1225913a4'
content='' additional_kwargs={'tool_calls': [{'id': 'call_aaEzj3aO1RTnB0uoc9rYUIhi', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None} response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 55, 'total_tokens': 69, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'tool_calls', 'logprobs': None} id='run-476bc4b1-b7bf-4607-a31c-ddf09dc814c5-0' tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_aaEzj3aO1RTnB0uoc9rYUIhi', 'type': 'tool_call'}] usage_metadata={'input_tokens': 55, 'output_tokens': 14, 'total_tokens': 69, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}}
content='Sorry there was an error, please try again.' name='magic_function' id='dcbe7e3e-0ed4-467d-a729-2f45916ff44f' tool_call_id='call_aaEzj3aO1RTnB0uoc9rYUIhi'
content="It seems there was an error when trying to compute the value of `magic_function(3)`. Let's try that again." additional_kwargs={'tool_calls': [{'id': 'call_jr4R8uJn2pdXF5GZC2Dg3YWS', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None} response_metadata={'token_usage': {'completion_tokens': 40, 'prompt_tokens': 87, 'total_tokens': 127, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'tool_calls', 'logprobs': None} id='run-d94b8932-6e9e-4ab1-99f7-7dca89887ffe-0' tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_jr4R8uJn2pdXF5GZC2Dg3YWS', 'type': 'tool_call'}] usage_metadata={'input_tokens': 87, 'output_tokens': 40, 'total_tokens': 127, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}}
{'input': 'what is the value of magic_function(3)?', 'output': 'Agent stopped due to max iterations.'}

trim_intermediate_steps

在 LangChain 中

使用 LangChain 的 AgentExecutor，您可以使用 trim_intermediate_steps 修剪长时间运行的 agent 的中间步骤，它可以是一个整数（表示 agent 应保留最后 N 个步骤）或自定义函数。

例如，我们可以修剪值，以便 agent 仅看到最近的中间步骤。

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")


prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)


magic_step_num = 1


@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    global magic_step_num
    print(f"Call number: {magic_step_num}")
    magic_step_num += 1
    return input + magic_step_num


tools = [magic_function]

agent = create_tool_calling_agent(model, tools, prompt=prompt)


def trim_steps(steps: list):
    # Let's give the agent amnesia
    return []


agent_executor = AgentExecutor(
    agent=agent, tools=tools, trim_intermediate_steps=trim_steps
)


query = "Call the magic function 4 times in sequence with the value 3. You cannot call it multiple times at once."

for step in agent_executor.stream({"input": query}):
    pass

API 参考：AgentExecutor | create_tool_calling_agent | ChatPromptTemplate | tool | ChatOpenAI

Call number: 1
Call number: 2
Call number: 3
Call number: 4
Call number: 5
Call number: 6
Call number: 7
Call number: 8
Call number: 9
Call number: 10
Call number: 11
Call number: 12
Call number: 13
Call number: 14
``````output
Stopping agent prematurely due to triggering stop condition
``````output
Call number: 15

在 LangGraph 中

当传入 提示模板 时，我们可以像以前一样使用 prompt。

from langgraph.errors import GraphRecursionError
from langgraph.prebuilt import create_react_agent
from langgraph.prebuilt.chat_agent_executor import AgentState

magic_step_num = 1


@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    global magic_step_num
    print(f"Call number: {magic_step_num}")
    magic_step_num += 1
    return input + magic_step_num


tools = [magic_function]


def _modify_state_messages(state: AgentState):
    # Give the agent amnesia, only keeping the original user query
    return [("system", "You are a helpful assistant"), state["messages"][0]]


langgraph_agent_executor = create_react_agent(
    model, tools, prompt=_modify_state_messages
)

try:
    for step in langgraph_agent_executor.stream(
        {"messages": [("human", query)]}, stream_mode="updates"
    ):
        pass
except GraphRecursionError as e:
    print("Stopping agent prematurely due to triggering stop condition")

API 参考：create_react_agent

Call number: 1
Call number: 2
Call number: 3
Call number: 4
Call number: 5
Call number: 6
Call number: 7
Call number: 8
Call number: 9
Call number: 10
Call number: 11
Call number: 12
Stopping agent prematurely due to triggering stop condition

后续步骤

您现在已经学习了如何将 LangChain agent executor 迁移到 LangGraph。

接下来，查看其他 LangGraph 操作指南。