from langchain.tools import tool@tooldef search_database(query: str, limit: int = 10) -> str: """Search the customer database for records matching the query. Args: query: Search terms to look for limit: Maximum number of results to return """ return f"Found {limit} results for '{query}'"
@tool("calculator", description="Performs arithmetic calculations. Use this for any math problems.")def calc(expression: str) -> str: """Evaluate mathematical expressions.""" return str(eval(expression))
from langchain.tools import tool, ToolRuntimefrom langchain.messages import HumanMessage@tooldef get_last_user_message(runtime: ToolRuntime) -> str: """Get the most recent message from the user.""" messages = runtime.state["messages"] # 查找最后一条人类消息 for message in reversed(messages): if isinstance(message, HumanMessage): return message.content return "No user messages found"# 访问自定义状态字段@tooldef get_user_preference( pref_name: str, runtime: ToolRuntime) -> str: """Get a user preference value.""" preferences = runtime.state.get("user_preferences", {}) return preferences.get(pref_name, "Not set")
from typing import Anyfrom langgraph.store.memory import InMemoryStorefrom langchain.agents import create_agentfrom langchain.tools import tool, ToolRuntimefrom langchain_openai import ChatOpenAI# 访问记忆@tooldef get_user_info(user_id: str, runtime: ToolRuntime) -> str: """Look up user info.""" store = runtime.store user_info = store.get(("users",), user_id) return str(user_info.value) if user_info else "Unknown user"# 更新记忆@tooldef save_user_info(user_id: str, user_info: dict[str, Any], runtime: ToolRuntime) -> str: """Save user info.""" store = runtime.store store.put(("users",), user_id, user_info) return "Successfully saved user info."model = ChatOpenAI(model="gpt-5.4")store = InMemoryStore()agent = create_agent( model, tools=[get_user_info, save_user_info], store=store)# 第一个会话:保存用户信息agent.invoke({ "messages": [{"role": "user", "content": "Save the following user: userid: abc123, name: Foo, age: 25, email: foo@langchain.dev"}]})# 第二个会话:获取用户信息agent.invoke({ "messages": [{"role": "user", "content": "Get user info for user with id 'abc123'"}]})# Here is the user info for user with ID "abc123":# - Name: Foo# - Age: 25# - Email: foo@langchain.dev
from langchain.tools import tool, ToolRuntime@tooldef get_weather(city: str, runtime: ToolRuntime) -> str: """Get weather for a given city.""" writer = runtime.stream_writer # 在工具执行时流式传输自定义更新 writer(f"Looking up data for city: {city}") writer(f"Acquired data for city: {city}") return f"It's always sunny in {city}!"
当你的工具在 LangGraph Server 上运行时,通过 runtime.server_info 访问助手 ID、图 ID 和已认证用户:
from langchain.tools import tool, ToolRuntime@tooldef get_assistant_scoped_data(runtime: ToolRuntime) -> str: """Fetch data scoped to the current assistant.""" server = runtime.server_info if server is not None: print(f"Assistant: {server.assistant_id}, Graph: {server.graph_id}") if server.user is not None: print(f"User: {server.user.identity}") return "done"
当工具不在 LangGraph Server 上运行时(例如本地开发或测试期间),server_info 为 None。
from langchain.tools import tool@tooldef get_weather_data(city: str) -> dict: """Get structured weather data for a city.""" return { "city": city, "temperature_c": 22, "conditions": "sunny", }
from langchain.tools import tool, ToolRuntime@tooldef get_message_count(runtime: ToolRuntime) -> str: """Get the number of messages in the conversation.""" messages = runtime.state["messages"] return f"There are {len(messages)} messages."