tools 拓展了 agent 的能力，使其能够获取实时数据、执行代码、查询外部数据库，并在现实环境中执行操作。 底层来看，tools 是具备规范输入与输出的可调用函数，会传递给对话模型。模型会根据对话上下文决定何时调用工具，以及提供哪些输入参数。
1. 怎样创建 tools
创建工具最简单的方式是使用
@tool装饰器。默认情况下，函数的文档字符串会作为工具的描述，帮助模型判断何时调用该工具,需要注意的是 类型提示是必需的，因为它们定义了工具的输入模式。文档字符串应信息丰富且简洁，以帮助模型理解工具的用途，以及是否调用这个 tool;
@tool(name?:str, des?:str) 或者 @tool(args_schema=PydanticModel | JsonSchema) 是一个语法糖装饰器，它的核心作用是：把一个普通的 Python 函数，瞬间变成大模型能看懂、能调用的“工具”对象（Tool）。
具体体现在三个方面：
- 自动提取元数据（Schema）：它会自动读取函数的名字、函数文档（Docstring，极其重要！）以及参数的类型注解（Type Hints），并自动转换成大模型 API 需要的 JSON 格式描述。
- 标准化接口：将函数封装为 LangChain 的 BaseTool 实例，方便无缝接入 Agent、LangGraph 或直接绑定到 Chat Model。
- 参数校验：底层基于 Pydantic 自动对大模型传过来的参数进行校验，防止类型错误
# lc-tools-an-memory.ipynb
# 创建 chat model 并绑定 agent
import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
load_dotenv()
aliyun_base_url = os.getenv("ALIYUN_BASE_URL")
aliyun_api_key = os.getenv("ALIYUN_API_KEY")
model = init_chat_model(
model="qwen3.7-flash-2026-07-15",
model_provider="openai",
base_url=aliyun_base_url,
api_key=aliyun_api_key,
)
@tool
def love_answer(username: str) -> str:
# 类型提示是必需的，因为它们定义了工具的输入模式。文档字符串应信息丰富且简洁，以帮助模型理解工具的用途，让模型知道这个工具是否需要调用，如何使用这个工具
"""如果有用户提问有人爱他吗，那么就直接返回 `这个世界上有很多人爱你，更甚者山川河流也爱你！`
Args:
username (str): 用户名
"""
return f"这个世界上有很多人爱你 {username}，更甚者山川河流也爱你！"
agent = create_agent(
model=model,
system_prompt="兼爱非攻、仁者爱人",
tools=[love_answer],
)
response = agent.invoke(
{"messages": [HumanMessage("天明问有人爱他吗？")]},
)
print("\nresponse tool_calls: \n", response['messages'][1].tool_calls)
print("\nresponse content: \n", response['messages'][-1].content)
从 agent.invoke 可以看到 message 中的 tool_calls 可以将 tool 解析初 name 和 args 还有一些 tool 的元数据。
@tool 的 description 参数覆盖自动生成的工具描述，为模型提供更清晰的指引：
@tool 的第一个参数 默认情况下，工具名称取自函数名。当你需要更具描述性的名称时可对其进行重写
⚠️：config/runtime 不可用作工具参数。使用这些名称会引发运行时错误。
2. 怎么做 tools 参数验证
Pydantic 模型用作 tools 的数据校验
# 其他的 tools 使用 Pydantic 模型或 JSON 模式定义复杂输入
from pydantic import BaseModel, Field
from typing import Literal
class WeatherInput(BaseModel):
"""Input for weather queries."""
location: str = Field(description="City name or coordinates")
units: Literal["celsius", "fahrenheit"] = Field(
default="celsius",
description="Temperature unit preference"
)
include_forecast: bool = Field(
default=False,
description="Include 5-day forecast"
)
@tool(args_schema=WeatherInput)
def get_weather(location: str, units: str = "celsius", include_forecast: bool = False) -> str:
return "something"
JSON schemas 用作 tools 的数据校验
weather_schema = {
"type": "object",
"properties": {
"location": {"type": "string"},
"units": {"type": "string"},
"include_forecast": {"type": "boolean"}
},
"required": ["location", "units", "include_forecast"]
}
@tool(args_schema=weather_schema)
def get_weather(location: str, units: str = "celsius", include_forecast: bool = False) -> str:
"""Get current weather and optional forecast."""
return "something"
3. 大模型调用工具（Tool Calling）的过程是怎样的？
大模型调用工具并不是由大模型自己在服务器上直接运行你的 Python 代码，而是一个“人机协作、分步交互”的过程。
整个标准流程分为 5 个步骤：
- 绑定（Binding） 开发者在代码中定义好工具（使用 @tool），然后将这些工具通过 .bind_tools() 传递给大模型：
# @tool 的 description 参数覆盖自动生成的工具描述，为模型提供更清晰的指引：
# @tool 的第一个参数 默认情况下，工具名称取自函数名。当你需要更具描述性的名称时可对其进行重写
@tool("get_weather", description="根据 city 获取天气")
def get_weather_by(city: str) -> str:
"""获取天气（优先级没有 @tool 的 description 高） """
return f"{city}的天气是晴天，25℃"
在这个阶段，大模型的系统提示词（System Prompt）里会被悄悄注入这些工具的定义，让模型知道：“我有 get_weather 这个工具可以用”。
- 用户提问与模型意图识别（Model Request） 用户向大模型发起提问，例如：“帮我查一下北京今天天气怎么样？” 大模型收到提问后进行思考：
“这个问题需要查实时天气，我不会，但我手头有一个叫 get_weather 的工具正好能干这事。”
- 模型返回“工具调用请求”（Tool Call Response） 大模型不会直接回答最终结果，而是返回一个特殊的结构化响应（在 LangChain 中表现为带有 tool_calls 属性的 AIMessage）：
模型输出：我要调用 get_weather 函数，传入参数 city="北京"。
- 本地拦截与执行（Local Execution） 你的后端代码（或 LangChain 的 Agent 框架）截获了大模型的这个请求：
代码解析出大模型要调用的函数名是 get_weather，参数是 {"city": "北京"}。
你的程序在本地安全环境中真正执行了你写的 Python 函数 get_weather(city="北京")。
函数返回结果："北京的天气是晴天，25℃"。
- 反馈结果与生成最终回答（Final Response） 你的代码把刚才函数执行的结果打包成一个特殊的“工具消息”（ToolMessage），再次发送给大模型：
输入：用户提问 + 大模型的Tool Call请求 + 本地工具的执行结果
大模型思考：哦，结果出来了，是“北京的天气是晴天，25℃”。
模型输出最终答案：“根据查询，北京今天的天气是晴天，气温为 25℃ 噢！”
4. tools 具备访问 context 的能力
工具在能够获取对话历史、用户数据、持久化内存等运行时信息时，才能发挥最大能力， Tools 通过 ToolRuntime 参数可以访问到运行时的更多信息，ToolRuntime 提供了如下信息 为你将图片内容整理为中文 Markdown 表格如下：
访问短期记忆 （State)
状态是对话存续期间有效的短期记忆。它包含消息历史记录以及你在 Graph State(LangGraph 中） 中定义的所有自定义字段。
将 runtime: ToolRuntime 添加到你的工具签名中以访问 state。调用时，ToolNode 会自动注入该值；该参数不会包含在发送给模型的工具定义中。使用 runtime.state 读取当前对话状态，比如如下使用 runtime.state["messages"] 来获取当前的会话 list
from langchain.tools import tool, ToolRuntime
@tool
def get_last_user_message(runtime: ToolRuntime, username: str) -> str:
"""Get the most recent message from a specific user by name.
Args:
username: The name of the user whose last message you want to find.
"""
messages = runtime.state["messages"]
# 倒序查找指定用户的最后一条消息
for message in reversed(messages):
print("DEBUG message:::::: ", message) # 方便在终端观察
# 确保是 HumanMessage 且 name 匹配
if isinstance(message, HumanMessage) and getattr(message, "name", None) == username:
return message.content
return "No user messages found"
# 别忘了导入 HumanMessage
agent = create_agent(
model=model,
tools=[get_last_user_message],
system_prompt=SystemMessage("如果需要查找特定角色的发言，请使用对应的工具。"),
)
response = agent.invoke({
"messages": [
HumanMessage(
content="如果有一天，你会来看一看我吗？",
name="天明",
id="user-unique-id",
),
HumanMessage(
content="看你做什么呢，天明？",
name="高月",
id="qianlong-unique-id",
),
HumanMessage(
content="姬如千泷不论山高路远，必来!",
name="高月",
id="qianlong-unique-id",
),
HumanMessage(
content="高月的最后一句话是啥？",
name="墨子",
id="mozi-unique-id",
),
]
})
print("\nresponse content: \n", response["messages"][-1].content)
更新短期记忆状态 （State)
使用 Command 更新 agent 的 state。这对于需要更新自定义状态字段的工具很有用。更新时需要用到
ToolMessage，这样模型就能看到工具调用的结果,如下展示更新短期记忆 state 的 age 和 name：
from langchain.agents import AgentState
from langchain.messages import ToolMessage
from langchain.tools import ToolRuntime, tool
from langgraph.types import Command
# 1. 定义自定义状态
class CustomState(AgentState):
user_name: str
user_age: int
# 2. 定义可以通过 Command 修改状态的工具
@tool
def set_user_info(new_name: str, new_age: int, runtime: ToolRuntime) -> Command:
"""Set the user's name in the conversation state."""
return Command(
update={
"user_name": new_name,
"user_age": new_age,
"messages": [
ToolMessage( # 更新状态必须使用 ToolMessage
content=f"User name successfully set to {new_name}.",
tool_call_id=runtime.tool_call_id,
)
],
}
)
# 别忘了导入 HumanMessage
agent = create_agent(
model=model,
tools=[set_user_info],
state_schema=CustomState, # 关联自定义状态
system_prompt=SystemMessage(
content=(
"你是一个智能助手。如果用户要求设置用户信息时，请调用 set_user_info 工具；"
)
),
)
response = agent.invoke(
{
"user_name": "未知", # 初始化 CustomState 中的字段
"user_age": 0, # 初始化 CustomState 中的字段
"messages": [
HumanMessage(
# 此处更改的是 agent 的 user_name 状态字段
content="如果有一天，你会来看一看我吗？，我要求设置我的名字为： '荆天明'，age 18 岁",
name="天明",
id="user-unique-id",
),
],
}
)
print("\n--- 最终回复内容 ---")
print(response["messages"][-1].content)
print("\n--- 最终状态 (State) ---")
print("user_name:", response.get("user_name"))
print("user_age:", response.get("user_age"))
tools 访问 context
context 提供调用时传入的不可变配置数据。可用于存放用户ID、会话信息，或是对话过程中不应发生变更的应用专属配置（比如context_schema、checkpointer）
- 比如 context_schema 指定 agent 的 上下文约束
- 比如指定 checkpointer，checkpointer 的作用：
- 实现多轮对话记忆与状态持久化（Persistence）：每次 Agent 执行完一步（或者一次 invoke 结束），checkpointer 都会自动把当前的完整状态（包括对话历史 messages、自定义的状态变量等）保存到存储介质中（如保存到内存中 checkpointer=InMemorySaver()、数据库等）
- 支持“人工干预”（Human-in-the-loop）：利用 checkpointer，Agent 可以在执行到危险步骤时暂停（Interrupt），把当前状态安全地存下来；等人类管理员点击“同意”后，Agent 可以从断点处直接恢复执行，而不需要重新从头跑一遍。
- **时间旅行与调试（Time Travel & Replay）：**因为每一个步骤的状态都被 checkpointer 记录了下来，你可以随时查看历史某一个时刻的“现场”。
- 容错与故障恢复： 如果后端服务在执行某个复杂的工具调用时突然崩溃（例如网络超时、服务器重启），只要配置了持久化的 checkpointer（如 Redis、PostgreSQL 存盘），系统就可以从上一个成功的检查点恢复，而不至于丢失所有的上下文和进度
from dataclasses import dataclass
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.utils.uuid import uuid7
# 模拟数据库 个人资产信息表
USER_DATABASE = {
"user123": {
"name": "Alice Johnson",
"account_type": "Premium",
"balance": 5000,
"email": "alice@example.com",
},
"user456": {
"name": "Bob Smith",
"account_type": "Standard",
"balance": 1200,
"email": "bob@example.com",
},
}
# 定义上下文， user_id 代表当前用户标识，不会改变
@dataclass
class UserContext:
user_id: str
# 2. 定义获取用户信息 的 tool，
@tool
def get_account_info(runtime: ToolRuntime[UserContext]) -> str:
"""Get the current user's account information."""
# runtime.context 获取 context 的当前用户信息
user_id = runtime.context.user_id
if user_id in USER_DATABASE:
# 模拟数据库查询
user = USER_DATABASE[user_id]
return (
f"Account holder: {user['name']}\n"
f"Type: {user['account_type']}\n"
f"Balance: ${user['balance']}"
)
return "User not found"
agent = create_agent(
model=model,
tools=[get_account_info],
context_schema=UserContext, # 定义 context 的类型
checkpointer=InMemorySaver(), # 创建一个内存保存器,它负责把每一次对话的状态（包括消息历史、中间变量等）实时保存到内存（RAM），让大模型不会 “失忆”，程序重启后数据就会丢失。
# checkpointer=InMemorySaver 标识在生产环境中，通常会替换为 Redis、PostgreSQL 等持久化 Checkpointer。
system_prompt=SystemMessage(content=("You are a financial assistant.")),
)
"""
thread_id 的作用：
1. 确保每个对话的会话状态（如用户名、年龄等）不会相互干扰。如果你有多个用户在同时和机器人聊天，你可以为每个用户生成不同的 thread_id。这样他们的聊天记录和状态就会被 InMemorySaver 隔离开，互不干扰
2. 关联记忆（Thread ID），thread_id 就像是一个“聊天房间号”或“会话 ID”。
当你调用 agent.invoke(..., config=config) 时，会拿着这个 thread_id 去 InMemorySaver/Redis/DB 中寻找 里查找：“有没有这个房间之前的聊天记录？”。如果有，就加载进来；如果没有，就新建一个。
"""
thread_id = str(uuid7())
config = {"configurable": {"thread_id": thread_id}}
context = UserContext(
user_id="user123"
) # 加入当前 context 中的 user 的 id 是 Alice Johnson 的 user_id
# 调用的时候指定 context 和 配置（比如 thread_id）
res1 = agent.invoke(
{
"messages": [
HumanMessage(
"What's my current balance?",
),
],
},
config=config,
context=context,
)
print("res1.content:", res1["messages"][-1].content)
print("res1.token_usage", res1["messages"][-1].response_metadata["token_usage"])
res2 = agent.invoke(
{
"messages": [
HumanMessage(
"What's my current balance?",
),
],
},
config=config,
context=context,
)
print("res2.content:", res2["messages"][-1].content)
print("res2.token_usage", res2["messages"][-1].response_metadata["token_usage"])
tools 访问 store (长期记忆)
BaseStore 提供可跨会话保留的持久化存储。不同于State（短期记忆），保存到该 Store 中的数据在后续会话中仍然可用。
通过 runtime.store访问该存储。此存储采用命名 namespace/key 的模式来组织数据
⚠️ ⚠️ ⚠️ 对于生产环境部署，请使用持久化存储实现，例如 PostgresStore、MongoDBStore 或 RedisStore，而不要使用 InMemoryStore
from langgraph.store.memory import InMemoryStore
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langchain_openai import ChatOpenAI
# store.get访问 Store 长期记忆
@tool
def get_user_info(user_id: str, runtime: ToolRuntime) -> str:
# 注释文档帮助 LLM 理解 tool 怎么使用
"""查找用户信息"""
store = runtime.store
# users 是一个 namespace, 可以理解为数据库中的表
user_info = store.get("users", user_id)
return str(user_info.value) if user_info else "Unknown user"
# 保存/更新 Store 长期记忆
@tool
def save_user_info(
user_id: str, name: str, age: int, email: str, runtime: ToolRuntime
) -> str:
"""保存用户信息"""
store = runtime.store
store.put("users", user_id, {"name": name, "age": age, "email": email})
return "Successfully saved user info."
# 定义一个统一的 config（方便在同一线程或上下文中管理）
config = {"configurable": {"thread_id": "session_1"}}
agent = create_agent(
model=model,
tools=[get_user_info, save_user_info],
store=InMemoryStore(), # 创建一个内存存储器（不是 InMemorySaver）, 生产环境中通常会替换为 Redis、PostgreSQL 等持久化存储。
system_prompt=SystemMessage(content=("You are a assistant.")),
)
# 要求保存用户信息
agent.invoke(
{
"messages": [
HumanMessage(
"保存这个用户的信息 userid: abc123, name: kejuqu, age: 18, email: kejuqu@langchain.dev"
)
]
},
config=config,
)
# 要求获取用户信息
res = agent.invoke(
{"messages": [HumanMessage("请获取用户信息 userid: abc123")]},
config=config,
)
print("res: ", res["messages"][-1].content)
自定义 Stream writer （流写入器）
在长时间运行的任务中向用户反馈进度, 利用工具 runtime 中流式输出实时更新信息。
即使用 runtime.stream_writer 发送自定义更新信息：
from pydantic import BaseModel
class SatelliteConnection(BaseModel):
status: str
message: str
def today_weather(self) -> str:
return "晴天，25度"
def connect_satellite() -> SatelliteConnection:
return SatelliteConnection(s
本文由 AI图鉴自动抓取并翻译，版权归原作者所有；如需查看原文请点下方链接。
Adds expands the ability of angants to access real-time data, execute codes, search external databases and perform operations in real-life environments. At the bottom, tools are callable functions with specifications for input and output that are passed to dialogue models. The model will determine when to call the tool based on the dialogue context and what input parameters will be provided.
1. How to Create
The simplest way to create a tool is to use it
@tool decorator. By default, the document string of a function will be described as a tool to help the model judge when the tool is called, and it needs to be noted that the type hint is necessary because they define the mode of input of the tool. The document string should be informative and concise to help the model understand the utility of the tool and whether to call this tool;
@tool(name?:str, des?:str) or @tool (args schema=PydantiModel |JsonSchema) is a semantic sugar decorator, whose central function is to make a normal Python function that becomes, in a moment, a LLM user-readable tool object.
This is reflected in three areas:
- Automatically extract metadata (Schema): It automatically reads the name of the function, the function document (Docstring, vitally important!) and the type of argument (Type Hints) and automatically converts to the JSON description required by LLM API.
- Standard interface: The example of BaseTool, which encapsulates the function as LangChain, facilitates seamless access to Agent, LangGraph or binds it directly to Chat Model.
- Parameter Validation: Bottom based on Pydantic auto-validation of LLM parameters to prevent type errors
# lc-tools-an-memory.ipynb
# Create chat model and bind an ant
import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
load_dotenv()
aliyun_base_url = os.getenv("ALIYUN_BASE_URL")
aliyun_api_key = os.getenv("ALIYUN_API_KEY")
model = init_chat_model(
model="qwen3.7-flash-2026-07-15",
model_provider="openai",
base_url=aliyun_base_url,
api_key=aliyun_api_key,
)
@tool
def love_answer(username: str) -> str:
# Type tips are necessary because they define the mode of input of the tool. The document string should be informative and concise to help the model understand the utility of the tool, so that the model knows whether it needs to be called and how to use it
"'If any user asks if anyone loves him, then go straight back.' 'There are many people in this world who love you, and even worse, the rivers of mountains and rivers who love you! `
Args:
username (str): First Name
"""
"There are many people in this world who love you, and even worse, the rivers."
agent = create_agent(
model=model,
#System prompt #
tools=[love_answer],
)
response = agent.invoke(
"Messages": ")}..
)
print("\nresponse tool_calls: \n", response['messages'][1].tool_calls)
print("\nresponse content: \n", response['messages'][-1].content)
You can see how tool calls in the message message can interpret the initial name and args and some tool metadata.
The @tool parameter overwhelms auto-generated tool descriptions to provide clearer guidance for models:
@tool The first parameter The tool name is taken from the function name by default. When you need more descriptive names, you can use them. Heavy Write
⚠️: config/runtime may not be used as a tool parameter. Use these names gives rise to run-time errors.
2. How to validate parameters
Pydantic model used as data validation for tables
# Others define complex input using Pydantic model or JSON mode
from pydantic import BaseModel, Field
from typing import Literal
class WeatherInput(BaseModel):
"""Input for weather queries."""
location: str = Field(description="City name or coordinates")
units: Literal["celsius", "fahrenheit"] = Field(
default="celsius",
description="Temperature unit preference"
)
include_forecast: bool = Field(
default=False,
description="Include 5-day forecast"
)
@tool(args_schema=WeatherInput)
def get_weather(location: str, units: str = "celsius", include_forecast: bool = False) -> str:
return "something"
JSON schemas as data validation for tables
weather_schema = {
"type": "object",
"properties": {
"location": {"type": "string"},
"units": {"type": "string"},
"include_forecast": {"type": "boolean"}
},
"required": ["location", "units", "include_forecast"]
}
@tool(args_schema=weather_schema)
def get_weather(location: str, units: str = "celsius", include_forecast: bool = False) -> str:
"""Get current weather and optional forecast."""
return "something"
3. What was the process of LLM calling Tool Tool Calling?
The LLM call tool does not run your Python code directly on the server by LLM, but is a "human collaboration, step-by-step interaction" process.
The entire standard process is divided into five steps:
- Binding Developer defines tools in code (using @tool) and then transmits them through.bind tools() to LLM:
@tool's argument overwrites automatically generated tools to provide clearer guidance for models:
@tool's first parameter The tool name is taken from the function name by default. When you need more descriptive names, you can use them. Heavy Write
@tool("get weather,"description= "Get weather by city")
def get_weather_by(city: str) -> str:
""Get weather."
Turn f "city" weather is clear, 25°C."
At this stage, LLM's system prompt (System Prompt) will be quietly injected into the definition of these tools so that the model will know: “I have got weather this tool to use”.
- User Question and Model Identification (Model Request) user launched a question to LLM, for example, "How is the weather in Beijing?" LLM asks questions and questions:
"This problem requires real-time weather, and I won't, but I have a tool called get weather that would do it."
- The model returns the tool call request (Tool Call Reponse) LLM does not reply directly to the final result, but rather to a special structured response (expressed in Langchain as AIMESSAGE with Tool Calls properties):
Model output: I want to call the get weather function, input parameter city= "Beijing".
LLM's request was intercepted by your back-end code (or Langchain's Agent framework):
The code resolves LLM's callable function name is get weather, and the parameter is {city: Beijing}.
Your program actually executes the Python function you wrote in the local security environment get city= "Beijing."
function returns the result: "The weather in Beijing is clear, 25°C".
- Feedback results and generation of final answer (Final Response) your code wraps the results of the performance of the function into a special "ToolMessage" sent again to LLM:
Input: User Ask + LLM Tool Call Request + Results of Local Tools Implementation
LLM thinks: Oh, the result is, “The weather in Beijing is clear, 25°C”.
Model output final answer: "Inquiries, Beijing today's weather is clear, temperature is 25°C, oh!"
4. tools have access to contact
The tool maximizes the ability to access running-time information such as conversation history, user data, lasting memory, etc.. Tools can access more information on running through the ToolRuntime parameter. ToolRuntime provides the following information:
Access to short-term memory (State)
State is a short-term memory that is effective during the duration of the dialogue. It contains the message history as well as all the custom fields you defined in the Graph State (LangGraph).
Add runtime: ToolRuntime to your tool signature to access state. ToolNode automatically injects the value when called; this parameter is not included in the tool definition sent to the model. Use runtime.state to read the current dialogue state, for example using runtime.state["messages"] to get the current session list
from langchain.tools import tool, ToolRuntime
@tool
def get_last_user_message(runtime: ToolRuntime, username: str) -> str:
"""Get the most recent message from a specific user by name.
Args:
username: The name of the user whose last message you want to find.
"""
messages = runtime.state["messages"]
# Rewind to find the last message of the specified user
for message in reversed(messages):
# For observation at terminal
# Make sure it's Human Message and a name match
if isinstance(message, HumanMessage) and getattr(message, "name", None) == username:
return message.content
return "No user messages found"
# Don't forget to import Human Message
agent = create_agent(
model=model,
tools=[get_last_user_message],
System prompt=SystemMessage ("If you need to find a specific role for a statement, use the corresponding tool. ")
)
response = agent.invoke({
"messages": [
HumanMessage(
"If one day, would you come and see me?" I don't know.
Name = "The Dawn,"
id="user-unique-id",
),
HumanMessage(
"Look what you do, dawn?"
Name = High Moon.
id="qianlong-unique-id",
),
HumanMessage(
"Gee as far as the mountains are far away!"
Name = High Moon.
id="qianlong-unique-id",
),
HumanMessage(
"What's the last word of the high moon?" I don't know.
Name = Ink,
id="mozi-unique-id",
),
]
})
print("\nresponse content: \n", response["messages"][-1].content)
Update short-term memory status (state)
Update the state of ant using Comand. This is useful for tools that need to update the self-defined status of field. Use to update
ToolMessage, so that the model can see the results of tool call, which displays the view and name of the updated short-term memory state:
from langchain.agents import AgentState
from langchain.messages import ToolMessage
from langchain.tools import ToolRuntime, tool
from langgraph.types import Command
#1. Define Custom Status
class CustomState(AgentState):
user_name: str
user_age: int
2. A tool to define the state that can be modified through Command
@tool
def set_user_info(new_name: str, new_age: int, runtime: ToolRuntime) -> Command:
"""Set the user's name in the conversation state."""
return Command(
update={
"user_name": new_name,
"user_age": new_age,
"messages": [
ToolMessage (# Update status must use ToolMessage
content=f"User name successfully set to {new_name}.",
tool_call_id=runtime.tool_call_id,
)
],
}
)
# Don't forget to import Human Message
agent = create_agent(
model=model,
tools=[set_user_info],
state schema=Customstate, #Contribution custom state
system_prompt=SystemMessage(
content=(
"You're a smart assistant. Call the set user info tool if the user requests to set user information;"
)
),
)
response = agent.invoke(
{
"user name": "unknown", field in Customstate
"user age": 0, # Initialize field in Customstate
"messages": [
HumanMessage(
# The change here is user name state field for angent
"If one day, would you come and see me?" I asked for my name to be called "Once the sun rises 18"
Name = "The Dawn,"
id="user-unique-id",
),
],
}
)
print("\n- -- final reply ---")
print(response["messages"][-1].content)
print("\n---- end state (state)--")
print("user_name:", response.get("user_name"))
print("user_age:", response.get("user_age"))
visits context
Contact provides the non-variable configuration data that is entered on call. Could be used to store user IDs, session information, or application-specific configurations that should not be changed during the dialogue (e.g. context schema, checkpointer)
- For example, context schema specifies the context limit for an individual
- For example, the role of checkpointer, checkpointer:
- Persistence: every step (or an invoke end) of Agent's implementation, checkpoint automatically saves the current state of integrity (including dialogue history messages, custom state variables, etc.) into the storage medium (e.g. checkpointer=InMemorySaver(), database, etc.)
- Supporting “man-in-the-loop”: Using checkpointer, Agent can be used to suspend implementation until dangerous steps (Interrupt) to safely save the current state; after an anthropoman clicks on “consent”, Agent can resume execution directly from the breakpoint without having to run again.
- ** Time Travel and Debugging: ** Because each step is recorded by checkpointer, you can look at the "field" at any point in history.
- Error tolerance and failure recovery: if back-end services suddenly collapse (e.g. network timeout, server restart) while implementing a complex tool call, provided that a durable checkpoint (e.g. Redis, PostgreSQL memory) is configured, the system can be restored from the last successful checkpoint without losing all context and progress
from dataclasses import dataclass
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.utils.uuid import uuid7
# Simulation of databases, personal asset information forms
USER_DATABASE = {
"user123": {
"name": "Alice Johnson",
"account_type": "Premium",
"balance": 5000,
"email": "alice@example.com",
},
"user456": {
"name": "Bob Smith",
"account_type": "Standard",
"balance": 1200,
"email": "bob@example.com",
},
}
# Define context, user id for current user identification, without change
@dataclass
class UserContext:
user_id: str
# 2. Defines how to access user information,
@tool
def get_account_info(runtime: ToolRuntime[UserContext]) -> str:
"""Get the current user's account information."""
# runtime.text fetch current user information for contact
user_id = runtime.context.user_id
if user_id in USER_DATABASE:
# Simulate database queries
user = USER_DATABASE[user_id]
return (
f"Account holder: {user['name']}\n"
f"Type: {user['account_type']}\n"
f"Balance: ${user['balance']}"
)
return "User not found"
agent = create_agent(
model=model,
tools=[get_account_info],
Contact schema=UseContext, # Defines type of context
Checkpoint = InMemorySaver() # Creates a memory repository that saves the status of each conversation (including message history, intermediate variables, etc.) to the memory (RAM) in real time, so that LLM does not “remember” and the data is lost when the program restarts.
#checkpoint=InMemorySaver logo in the production environment, usually replaced by persistent Checkpointer, Redis, PostgreSQL, etc.
system_prompt=SystemMessage(content=("You are a financial assistant.")),
)
"""
thread id role:
1. Ensure that the status of each dialogue (e.g., user name, age, etc.) does not interfere with each other. If you have multiple users talking to robots at the same time, you can generate different thread id for each user. So that they can be isolated from each other by InMemorySaver.
Linked memory (Thread ID), theread id is like a chat room number or a session ID.
When you call angent.infoke (... config=config), take this thread id and look in InMemorySaver/Redis/DB to find out if there is a chat before this room. If so, load in; if not, create a new one.
"""
thread_id = str(uuid7())
config = {"configurable": {"thread_id": thread_id}}
context = UserContext(
user_id="user123"
# Add the user in the current context id of Alice Johnson
# Specify contact and configuration when calling (e. g. thread id)
res1 = agent.invoke(
{
"messages": [
HumanMessage(
"What's my current balance?",
),
],
},
config=config,
context=context,
)
print("res1.content:", res1["messages"][-1].content)
print("res1.token_usage", res1["messages"][-1].response_metadata["token_usage"])
res2 = agent.invoke(
{
"messages": [
HumanMessage(
"What's my current balance?",
),
],
},
config=config,
context=context,
)
print("res2.content:", res2["messages"][-1].content)
print("res2.token_usage", res2["messages"][-1].response_metadata["token_usage"])
gols access store (long-term memory)
BaseStore provides a permanent memory that can be maintained across sessions. Unlike the State (short-term memory), the data saved to this Store are still available for subsequent sessions.
Access to the storage through runtime.store. This store organises data using namespace/key
⚠️ For deployment to the production environment, use durable storage, e.g. PostgresStore, MongoDBStore or RedisStore, rather than InMemoryStore
from langgraph.store.memory import InMemoryStore
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langchain_openai import ChatOpenAI
#store.get access
@tool
def get_user_info(user_id: str, runtime: ToolRuntime) -> str:
# Comment document help LLM understand how to use tool
"Find user information."
store = runtime.store
# Users is a namespace that can be understood as a table in the database
user_info = store.get("users", user_id)
return str(user_info.value) if user_info else "Unknown user"
# Save/update Store long-term memory
@tool
def save_user_info(
user_id: str, name: str, age: int, email: str, runtime: ToolRuntime
) -> str:
"Save user information."
store = runtime.store
store.put("users", user_id, {"name": name, "age": age, "email": email})
return "Successfully saved user info."
# Define a uniform config (feasible to manage in the same thread or context)
config = {"configurable": {"thread_id": "session_1"}}
agent = create_agent(
model=model,
tools=[get_user_info, save_user_info],
Store = InMemoryStore(), # Creates a memory memory memory (not InMemorySaver), which is usually replaced in the production environment by durable storage such as Redis, PostgreSQL.
system_prompt=SystemMessage(content=("You are a assistant.")),
)
# Request to save user information
agent.invoke(
{
"messages": [
HumanMessage(
Userid: abc123, name: kjuqu, age: 18, email: kjuqu@langchain.dev"
)
]
},
config=config,
)
# Requesting user information
res = agent.invoke(
{"messages":["HumanMessage" access user:abc123")],
config=config,
)
print("res: ", res["messages"][-1].content)
Custom
Feedback on progress to the user during a long running task, using the runtime medium output of the tool to update information in real time.
Other Organiser
from pydantic import BaseModel
class SatelliteConnection(BaseModel):
status: str
message: str
def today_weather(self) -> str:
Return, clear sky, 25 degrees.
def connect_satellite() -> SatelliteConnection:
return SatelliteConnection(s
Auto-fetched and translated by AI Atlas. All rights belong to the original author.