from langchain_openai import ChatOpenAIfrom langchain.agents import create_agentfrom langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponsebasic_model = ChatOpenAI(model="gpt-4.1-mini")advanced_model = ChatOpenAI(model="gpt-4.1")@wrap_model_calldef dynamic_model_selection(request: ModelRequest, handler) -> ModelResponse: """Choose model based on conversation complexity.""" message_count = len(request.state["messages"]) if message_count > 10: # Use an advanced model for longer conversations model = advanced_model else: model = basic_model return handler(request.override(model=model))agent = create_agent( model=basic_model, # Default model tools=tools, middleware=[dynamic_model_selection])
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-4.1", tools=[public_search, private_search, advanced_search], middleware=[state_based_tools])
根据 Store 中的用户偏好或功能标志过滤工具：
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-4.1", tools=[search_tool, analysis_tool, export_tool], middleware=[store_based_tools], context_schema=Context, store=InMemoryStore())
根据运行时上下文中的用户权限过滤工具：
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-4.1", 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"}]})
================================ Human Message =================================Find the most popular wireless headphones right now and check if they're in stock
================================= Tool Message =================================Product WH-1000XM5: 10 units in stock
推理：“我已经有了最流行的型号及其库存状态。我现在可以回答用户的问题了。”
行动：生成最终答案
================================== Ai Message ==================================I found wireless headphones (model WH-1000XM5) with 10 units in stock...
from langchain.messages import AIMessage, HumanMessagefor chunk in agent.stream({ "messages": [{"role": "user", "content": "Search for AI news and summarize the findings"}]}, stream_mode="values"): # Each chunk contains the full state at that point latest_message = chunk["messages"][-1] if latest_message.content: if isinstance(latest_message, HumanMessage): print(f"User: {latest_message.content}") elif isinstance(latest_message, AIMessage): print(f"Agent: {latest_message.content}") elif latest_message.tool_calls: print(f"Calling tools: {[tc['name'] for tc in latest_message.tool_calls]}")
