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"] # Find the last human message for message in reversed(messages): if isinstance(message, HumanMessage): return message.content return "No user messages found"# Access custom state fields@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# Access memory@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"# Update memory@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)# First session: save user infoagent.invoke({ "messages": [{"role": "user", "content": "Save the following user: userid: abc123, name: Foo, age: 25, email: foo@langchain.dev"}]})# Second session: get user infoagent.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 # Stream custom updates as the tool executes 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 toolfrom langgraph.prebuilt import ToolNodefrom langgraph.graph import StateGraph, MessagesState, START, END@tooldef search(query: str) -> str: """Search for information.""" return f"Results for: {query}"@tooldef calculator(expression: str) -> str: """Evaluate a math expression.""" return str(eval(expression))# Create the ToolNode with your toolstool_node = ToolNode([search, calculator])# Use in a graphbuilder = StateGraph(MessagesState)builder.add_node("tools", tool_node)# ... add other nodes and edges
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, ToolRuntimefrom langgraph.prebuilt import ToolNode@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."tool_node = ToolNode([get_message_count])
