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}'"
Type hints 是必需的,因为它们会定义 tool 的输入 schema。Docstring 应信息充分且简洁,以帮助模型理解 tool 的用途。
@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}!"
当 tool 在 LangGraph Server 上运行时,通过 runtime.server_info 访问 assistant ID、graph ID 和 authenticated user:
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"
当 tool 未在 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."
from langchain.agents import create_agentfrom langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponsefrom typing import Callable@wrap_model_calldef state_based_tools( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]) -> ModelResponse: """Filter tools based on conversation State.""" # Read from State: check if user has authenticated state = request.state is_authenticated = state.get("authenticated", False) message_count = len(state["messages"]) # Only enable sensitive tools after authentication if not is_authenticated: tools = [t for t in request.tools if t.name.startswith("public_")] request = request.override(tools=tools) elif message_count < 5: # Limit tools early in conversation tools = [t for t in request.tools if t.name != "advanced_search"] request = request.override(tools=tools) return handler(request)agent = create_agent( model="gpt-5.4", tools=[public_search, private_search, advanced_search], middleware=[state_based_tools])
根据 Store 中的用户偏好或 feature flags 过滤 tools:
from dataclasses import dataclassfrom langchain.agents import create_agentfrom langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponsefrom typing import Callablefrom langgraph.store.memory import InMemoryStore@dataclassclass Context: user_id: str@wrap_model_calldef store_based_tools( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]) -> ModelResponse: """Filter tools based on Store preferences.""" user_id = request.runtime.context.user_id # Read from Store: get user's enabled features store = request.runtime.store feature_flags = store.get(("features",), user_id) if feature_flags: enabled_features = feature_flags.value.get("enabled_tools", []) # Only include tools that are enabled for this user tools = [t for t in request.tools if t.name in enabled_features] request = request.override(tools=tools) return handler(request)agent = create_agent( model="gpt-5.4", tools=[search_tool, analysis_tool, export_tool], middleware=[store_based_tools], context_schema=Context, store=InMemoryStore())
根据 Runtime Context 中的用户权限过滤 tools:
from dataclasses import dataclassfrom langchain.agents import create_agentfrom langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponsefrom typing import Callable@dataclassclass Context: user_role: str@wrap_model_calldef context_based_tools( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]) -> ModelResponse: """Filter tools based on Runtime Context permissions.""" # Read from Runtime Context: get user role if request.runtime is None or request.runtime.context is None: # If no context provided, default to viewer (most restrictive) user_role = "viewer" else: user_role = request.runtime.context.user_role if user_role == "admin": # Admins get all tools pass elif user_role == "editor": # Editors can't delete tools = [t for t in request.tools if t.name != "delete_data"] request = request.override(tools=tools) else: # Viewers get read-only tools tools = [t for t in request.tools if t.name.startswith("read_")] request = request.override(tools=tools) return handler(request)agent = create_agent( model="gpt-5.4", tools=[read_data, write_data, delete_data], middleware=[context_based_tools], context_schema=Context)
from langchain.tools import toolfrom langchain.agents import create_agentfrom langchain.agents.middleware import AgentMiddleware, ModelRequest, ToolCallRequest# A tool that will be added dynamically at runtime@tooldef calculate_tip(bill_amount: float, tip_percentage: float = 20.0) -> str: """Calculate the tip amount for a bill.""" tip = bill_amount * (tip_percentage / 100) return f"Tip: ${tip:.2f}, Total: ${bill_amount + tip:.2f}"class DynamicToolMiddleware(AgentMiddleware): """Middleware that registers and handles dynamic tools.""" def wrap_model_call(self, request: ModelRequest, handler): # Add dynamic tool to the request # This could be loaded from an MCP server, database, etc. updated = request.override(tools=[*request.tools, calculate_tip]) return handler(updated) def wrap_tool_call(self, request: ToolCallRequest, handler): # Handle execution of the dynamic tool if request.tool_call["name"] == "calculate_tip": return handler(request.override(tool=calculate_tip)) return handler(request)agent = create_agent( model="gpt-4o", tools=[get_weather], # Only static tools registered here middleware=[DynamicToolMiddleware()],)# The agent can now use both get_weather AND calculate_tipresult = agent.invoke({ "messages": [{"role": "user", "content": "Calculate a 20% tip on $85"}]})