1、判断用户是否已登录，并根据登录状态决定用户可以调用哪些工具

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable

@wrap_model_call
def 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]
)

@wrap_model_call 装饰器是一个中间件装饰器，用于拦截模型调用请求（ModelRequest）在真正调用模型前对请求内容进行修改，它接收一个函数，并返回一个包装后的中间件，该中间件会在每次模型推理前执行

1
2
3
4
def state_based_tools(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
  • request：包含当前模型调用所需的所有信息，包括提示、工具列表、对话状态（state）等。
  • handler：原始的模型调用函数，中间件处理完后需将修改后的 request 传给它。

2、根据用户配置的工具，重写模型可调用工具列表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable
from langgraph.store.memory import InMemoryStore

@dataclass
class Context:
    user_id: str

@wrap_model_call
def 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()
)

@dataclass 装饰器装饰了Context类，定义类运行时上下文的结构，要求每次调用Agent时必须提供user_id，这个user_id将用于从Store（用的内存缓存）中查询该用户的个性化配置。

3、根据权限调整用户可用工具

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable

@dataclass
class Context:
    user_role: str

@wrap_model_call
def 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
)

4、自定义错误处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage


@wrap_tool_call
def handle_tool_errors(request, handler):
    """Handle tool execution errors with custom messages."""
    try:
        return handler(request)
    except Exception as e:
        # Return a custom error message to the model
        return ToolMessage(
            content=f"Tool error: Please check your input and try again. ({str(e)})",
            tool_call_id=request.tool_call["id"]
        )

agent = create_agent(
    model="gpt-4.1",
    tools=[search, get_weather],
    middleware=[handle_tool_errors]
)

5、系统提示

可以使用system_prompt引导代理如何处理任务

1
2
3
4
5
agent = create_agent(
    model,
    tools,
    system_prompt="你是一个乐于助人的助手。回答要简洁准确。"
)

可以使用SystemMessage对象，可以更灵活的构建提示结构，尤其适用于需要模型提供特定功能的场景，如Anthropic（一个模型提供商，提供了专属的功能如缓存）提示缓存功能

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain.agents import create_agent
from langchain.messages import SystemMessage, HumanMessage

literary_agent = create_agent(
    model="anthropic:claude-sonnet-4-5",
    system_prompt=SystemMessage(
        content=[
            {
                "type": "text",
                "text": "你是一个负责分析文学作品的 AI 助手。"
            },
            {
                "type": "text",
                "text": "<《傲慢与偏见》全文内容>",
                "cache_control": {"type": "ephemeral"}
            }
        ]
    )
)

result = literary_agent.invoke(
    {"messages": [HumanMessage("请分析《傲慢与偏见》中的主要主题。")]}
)
  • cache_control: {"type": "ephemeral"} 是 Anthropic 特有的字段，用于指示模型临时缓存该文本块。
  • 缓存后，在后续使用相同系统提示的请求中，该大段文本（如整本小说）无需重复传输，从而降低延迟和 API 调用成本。
  • 这种模式特别适合需要在系统提示中嵌入大量上下文（如文档、代码库、知识库）的场景。

6、动态提示词，根据用户角色选择合适的提示词

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from typing import TypedDict

from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest


class Context(TypedDict):
    user_role: str

@dynamic_prompt
def user_role_prompt(request: ModelRequest) -> str:
    """Generate system prompt based on user role."""
    user_role = request.runtime.context.get("user_role", "user")
    base_prompt = "You are a helpful assistant."

    if user_role == "expert":
        return f"{base_prompt} Provide detailed technical responses."
    elif user_role == "beginner":
        return f"{base_prompt} Explain concepts simply and avoid jargon."

    return base_prompt

agent = create_agent(
    model="gpt-4.1",
    tools=[web_search],
    middleware=[user_role_prompt],
    context_schema=Context
)

# The system prompt will be set dynamically based on context
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Explain machine learning"}]},
    context={"user_role": "expert"}
)

7、名字

你可以给Agent设置一个可选名称，名称主要用于多智能体系统，当把agent嵌入到更大的工作流或写作架构时，会用其作为节点标识符（如研究助手->写作助手->审核助手）

1
2
3
4
5
agent = create_agent(
    model,
    tools,
    name="research_assistant"
)

8、结构化输出

当希望agent返回严格符合特定格式的数据而不是自由文本时，可以通过response_format 参数提供格式化输出支持

ToolStrategy 是一种通用的结构化输出实现方式。它通过模拟一次“虚拟工具调用”，让模型以工具参数的形式返回结构化数据。
该策略适用于所有支持工具调用（tool calling）的模型，尤其在以下情况推荐使用：

  • 模型不原生支持结构化输出（如 GPT-4o 的 response_format={"type": "json_schema"}）
  • 原生结构化输出不稳定或不可靠
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from pydantic import BaseModel
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy

# 1. 定义期望的输出结构（使用 Pydantic）
class ContactInfo(BaseModel):
    name: str
    email: str
    phone: str

# 2. 创建代理，并指定结构化输出格式
agent = create_agent(
    model="gpt-4.1-mini",
    tools=[search_tool],
    response_format=ToolStrategy(ContactInfo)  # ← 关键：启用结构化输出
)

# 3. 调用代理
result = agent.invoke({
    "messages": [{"role": "user", "content": "从以下内容中提取联系信息：John Doe, john@example.com, (555) 123-4567"}]
})

# 4. 获取结构化结果
contact = result["structured_response"]
print(contact)
# 输出：ContactInfo(name='John Doe', email='john@example.com', phone='(555) 123-4567')

ToolStrategy 是模型输出内容后模拟了虚拟工具调用拼成我们想要的结构化数据，并不是在模型输出时直接输出的结构化数据，如果想让模型直接按我们的想法输出就需要看看模型提供商支不支持了，如果想让模型直接输出应使用ProviderStrategy

1
2
3
4
5
6
from langchain.agents.structured_output import ProviderStrategy

agent = create_agent(
    model="gpt-4.1",
    response_format=ProviderStrategy(ContactInfo)
)

好消息，从langchain1.0后，可以直接写成response_format=ContactInfo 会自动选择使用ToolStrategy还是ProviderStrategy

9、记忆

agent通过消息状态维护历史对话，除此之外还可以配置agent使用自定义状态结构，在对话过程中记住额外的信息

  • 自定义状态必须继承 AgentState （使用 TypedDict 形式）。
  • 有两种方式定义自定义状态：
    1. 通过中间件（middleware）——推荐方式
    2. 通过 create_agent 的 state_schema 参数

当你希望特定的中间件或工具能够访问自定义状态字段时，应使用中间件方式定义状态。这种方式将状态、工具和中间件逻辑紧密关联，便于模块化管理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from langchain.agents import AgentState
from langchain.agents.middleware import AgentMiddleware
from typing import Any

# 1. 定义自定义状态：在 AgentState 基础上添加 user_preferences 字段
class CustomState(AgentState):
    user_preferences: dict  # 用于存储用户偏好，如解释风格、详细程度等

# 2. 创建中间件，并绑定自定义状态和专属工具
class CustomMiddleware(AgentMiddleware):
    state_schema = CustomState          # ← 指定该中间件使用的状态结构
    tools = [tool1, tool2]              # ← 可选：为该中间件注册专用工具

    def before_model(self, state: CustomState, runtime) -> dict[str, Any] | None:
        # 在模型调用前，可读取或修改状态
        if state["user_preferences"].get("style") == "technical":
            # 可动态调整系统提示或工具行为
            return {"system_prompt": "Use technical language."}
        return None

# 3. 创建代理，挂载中间件
agent = create_agent(
    model,
    tools=tools,
    middleware=[CustomMiddleware()]
)

调用时传入自定义状态：

1
2
3
4
result = agent.invoke({
    "messages": [{"role": "user", "content": "我更喜欢技术性解释"}],
    "user_preferences": {"style": "technical", "verbosity": "detailed"}
})

10、流式输出

前面我们看到，通过 agent.invoke() 可以获取代理的最终完整响应。但如果代理需要执行多个步骤（例如多次工具调用），整个过程可能耗时较长。为了提升用户体验，我们可以使用 stream() 方法，在代理执行过程中实时返回中间进展。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from langchain.messages import AIMessage, HumanMessage

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "搜索 AI 新闻并总结发现"}]},
    stream_mode="values"
):
    # 每个 chunk 包含当前完整的状态（state）
    latest_message = chunk["messages"][-1]  # 获取最新一条消息

    if latest_message.content:
        if isinstance(latest_message, HumanMessage):
            print(f"用户: {latest_message.content}")
        elif isinstance(latest_message, AIMessage):
            print(f"代理: {latest_message.content}")
    elif latest_message.tool_calls:
        # 如果是工具调用，打印工具名称
        tool_names = [tc["name"] for tc in latest_message.tool_calls]
        print(f"正在调用工具: {tool_names}")