如何从旧版 LangChain 代理迁移到 LangGraph
这里我们重点介绍如何从旧版 LangChain 代理迁移到更灵活的 LangGraph 代理。LangChain 代理(特别是 AgentExecutor)有多个配置参数。在本笔记本中,我们将展示这些参数如何使用 create_react_agent 预构建辅助方法映射到 LangGraph react 代理执行器。
在 LangGraph 中,图取代了 LangChain 的代理执行器。它管理代理的周期,并在其状态中将 scratchpad 作为消息进行跟踪。LangChain 的“代理”对应于您提供的 state_modifier 和 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 风格代理的基本创建和使用,其功能是相同的。首先,让我们定义一个模型和工具,然后我们将使用它们来创建一个代理。
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)?"
对于 LangChain AgentExecutor,我们定义一个提示,其中包含代理草稿的占位符。 可以按如下方式调用代理
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})
{'input': 'what is the value of magic_function(3)?',
'output': 'The value of `magic_function(3)` is 5.'}
LangGraph 的 react 代理执行器 管理一个由消息列表定义的状态。它将继续处理列表,直到代理的输出中没有工具调用。 要启动它,我们输入一个消息列表。输出将包含图的整个状态——在本例中是对话历史。
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,
}
{'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 代理,你必须传入一个提示模板。 你可以使用它来控制代理。
使用 LangGraph react 代理执行器,默认情况下没有提示。 你可以通过几种方式实现对代理的类似控制
- 以系统消息作为输入传入
- 使用系统消息初始化代理
- 使用一个在传递给模型之前转换消息的函数来初始化代理。
让我们看看下面的所有这些。我们将传入自定义指令,以使代理以西班牙语回复。
首先,使用 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 代理执行器 传递一个自定义系统消息。
LangGraph 的预构建 create_react_agent
不直接将提示模板作为参数,而是采用一个 state_modifier
参数。 这会在调用 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, state_modifier=system_message
)
messages = langgraph_agent_executor.invoke({"messages": [("user", query)]})
我们还可以传入一个任意函数。 此函数应接收一个消息列表并输出一个消息列表。我们可以在这里对消息进行各种类型的任意格式化。 在这种情况下,我们只是在消息列表的开头添加一个 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}"),
]
)
def _modify_state_messages(state: AgentState):
return prompt.invoke({"messages": state["messages"]}).to_messages() + [
("user", "Also say 'Pandamonium!' after the answer.")
]
langgraph_agent_executor = create_react_agent(
model, tools, state_modifier=_modify_state_messages
)
messages = langgraph_agent_executor.invoke({"messages": [("human", query)]})
print(
{
"input": query,
"output": messages["messages"][-1].content,
}
)
{'input': 'what is the value of magic_function(3)?', 'output': 'El valor de magic_function(3) es 5. ¡Pandamonium!'}
内存
在 LangChain 中
使用 LangChain 的 AgentExecutor,你可以添加聊天 Memory,以便它可以进行多轮对话。
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"
]
)
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 中
向代理添加一个 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, state_modifier=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
)
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)
{'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}"),
]
)
def _modify_state_messages(state: AgentState):
return prompt.invoke({"messages": state["messages"]}).to_messages()
langgraph_agent_executor = create_react_agent(
model, tools, state_modifier=_modify_state_messages
)
for step in langgraph_agent_executor.stream(
{"messages": [("human", query)]}, stream_mode="updates"
):
print(step)
{'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_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 代理执行器 将所有消息附加到中心状态。 因此,只需查看完整状态即可轻松查看任何中间步骤。
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
{'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})
[1m> Entering new AgentExecutor chain...[0m
[32;1m[1;3mLo 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?[0m
[1m> Finished chain.[0m
{'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 中,每个步骤都会影响递归限制,因此我们需要乘以二(并加一)才能获得等效的结果。
如果达到递归限制,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."})
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})
[1m> Entering new AgentExecutor chain...[0m
[32;1m[1;3mLo 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.[0m
[1m> Finished chain.[0m
{'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 代理,你可以控制两个级别的超时。
你可以设置一个 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."})
{'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.")
{'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,以返回一个字符串,内容为“由于迭代限制或时间限制,代理已停止。”("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"])
Output with early_stopping_method='force':
Agent stopped due to max iterations.
在 LangGraph 中
在 LangGraph 中,你可以显式地在代理外部处理响应行为,因为可以访问完整状态。
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."})
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 来修剪长时间运行的代理的中间步骤,它可以是一个整数(表示代理应保留最后 N 个步骤)或一个自定义函数。
例如,我们可以修剪该值,使代理仅看到最近的中间步骤。
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
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 中
当传入提示模板时,我们可以像之前一样使用 state_modifier
。
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, state_modifier=_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")
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 代理执行器迁移到 LangGraph。
接下来,查看其他的 LangGraph 使用指南。