LangChain 核心组件之Agents

📌 摘要

LangChain Agents 是将语言模型与工具结合的系统，能够推理任务、选择工具并迭代解决问题。核心组件包括：

  1. 模型：作为代理的推理引擎，支持静态和动态配置。静态模型初始化后保持不变，动态模型可根据运行时状态选择不同模型。
  2. 工具：赋予代理执行动作的能力，支持顺序/并行调用、动态选择和错误处理。工具可以是函数或协程，通过装饰器自定义属性。
  3. 代理工作流：基于图的运行时环境，包含模型节点、工具节点等，通过LangGraph实现。代理会循环调用工具直到达成目标或达到迭代限制。

Agents（代理）

Agents 将语言模型与 tools 结合，用于创建能够推理任务、决定使用哪些工具，并迭代推进解决方案的系统。

create_agent 提供了一个可投产的 Agent 实现。

一个 LLM Agent 通过循环调用工具来达成目标。Agent 会一直运行，直到满足停止条件 —— 即模型输出最终结果或达到迭代上限。

action

observation

finish

input

model

tools

output

create_agent 构建了一个基于图的 Agent 运行时，使用 LangGraph。图由节点（步骤）和边（连接）组成，定义代理如何处理信息。Agent 会沿着图执行节点，比如调用模型的模型节点、执行工具的工具节点，或者执行中间件节点。

了解更多请参见 Graph API。

核心组件

模型

model 是 Agent 的推理引擎。它可以通过多种方式指定，支持静态模型和动态模型选择。

静态模型

静态模型在创建 Agent 时配置一次，并在执行过程中保持不变。这是最常见且最直接的方式。

要从 model identifier string 初始化静态模型：

from langchain.agents import create_agent

agent = create_agent("openai:gpt-5", tools=tools)

提示：模型标识字符串支持自动推断（例如，"gpt-5" 会被推断为 "openai:gpt-5"）。请参见 init_chat_model 参考文档，查看完整的模型标识字符串映射列表。

如果需要更细粒度的模型配置，可以直接使用提供商包初始化模型实例。在这个示例中，我们使用 ChatOpenAI。更多可用聊天模型类请参见 Chat models。

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="gpt-5",
    temperature=0.1,
    max_tokens=1000,
    timeout=30
    # ... (other params)
)
agent = create_agent(model, tools=tools)

模型实例让你可以完全控制配置。需要设置 temperature、max_tokens、timeouts、base_url 等提供商特定参数时，建议使用模型实例。请参见 Models 和 Providers。

import { createAgent } from "langchain";
import { ChatOpenAI } from "@langchain/openai";

const model = new ChatOpenAI({
  model: "gpt-4.1",
  temperature: 0.1,
  maxTokens: 1000,
  timeout: 30
});

const agent = createAgent({
  model,
  tools: []
});
动态模型

动态模型会在运行时根据当前状态和上下文进行选择。这使得可以实现复杂的路由逻辑和成本优化。

要使用动态模型，请创建一个中间件，使用 @wrap_model_call 装饰器修改请求中的模型：

from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse

basic_model = ChatOpenAI(model="gpt-4.1-mini")
advanced_model = ChatOpenAI(model="gpt-4.1")

@wrap_model_call
def dynamic_model_selection(request: ModelRequest, handler) -> ModelResponse:
    """根据对话复杂度选择模型。"""
    message_count = len(request.state["messages"])

    if message_count > 10:
        # 对于较长对话使用更强的模型
        model = advanced_model
    else:
        model = basic_model

    return handler(request.override(model=model))

agent = create_agent(
    model=basic_model,  # 默认模型
    tools=tools,
    middleware=[dynamic_model_selection]
)

警告：已预绑定工具的模型（即已调用 bind_tools）在使用结构化输出时不受支持。如果需要在结构化输出中进行动态模型选择，请确保传递给中间件的模型未预绑定。

import { ChatOpenAI } from "@langchain/openai";
import { createAgent, createMiddleware } from "langchain";

const basicModel = new ChatOpenAI({ model: "gpt-4.1-mini" });
const advancedModel = new ChatOpenAI({ model: "gpt-4.1" });

const dynamicModelSelection = createMiddleware({
  name: "DynamicModelSelection",
  wrapModelCall: (request, handler) => {
    const messageCount = request.messages.length;
    return handler({
        ...request,
        model: messageCount > 10 ? advancedModel : basicModel,
    });
  },
});

const agent = createAgent({
  model: "gpt-4.1-mini",
  tools,
  middleware: [dynamicModelSelection],
});

提示：有关模型配置详情，请参见 Models。有关动态模型选择模式，请参见 Dynamic model in middleware。

工具

工具为 Agent 提供执行动作的能力。Agent 不仅仅是简单将模型与工具绑定，它还支持：

  • 在单次提示下顺序调用多个工具
  • 在适当时并行调用工具
  • 基于先前结果进行动态工具选择
  • 工具重试逻辑与错误处理
  • 跨工具调用维持状态持久性

更多信息请参见 Tools。

静态工具

静态工具在创建 Agent 时定义，并在执行过程中保持不变。这是最常见且最直接的方式。

要定义静态工具，请将工具列表传递给 Agent：

提示：工具可以是普通 Python 函数或协程。使用 tool decorator 可以自定义工具名称、描述、参数模式等属性。

from langchain.tools import tool
from langchain.agents import create_agent


@tool
def search(query: str) -> str:
    """搜索信息。"""
    return f"Results for: {query}"

@tool
def get_weather(location: str) -> str:
    """获取某个地点的天气信息。"""
    return f"Weather in {location}: Sunny, 72°F"

agent = create_agent(model, tools=[search, get_weather])

import * as z from "zod";
import { createAgent, tool } from "langchain";

const search = tool(
  ({ query }) => `Results for: ${query}`,
  {
    name: "search",
    description: "Search for information",
    schema: z.object({
      query: z.string().describe("The query to search for"),
    }),
  }
);

const getWeather = tool(
  ({ location }) => `Weather in ${location}: Sunny, 72°F`,
  {
    name: "get_weather",
    description: "Get weather information for a location",
    schema: z.object({
      location: z.string().describe("The location to get weather for"),
    }),
  }
);

const agent = createAgent({
  model: "gpt-4.1",
  tools: [search, getWeather],
});

如果提供空的工具列表，则 Agent 将仅由单一 LLM 节点组成，不具备工具调用能力。

动态工具

动态工具是在运行时修改 Agent 可用工具集，而不是提前定义全部工具。并非所有工具都适用于每种情况。工具太多会让模型过载（增加上下文），工具太少则会限制能力。动态工具选择允许根据认证状态、用户权限、功能开关或会话阶段调整可用工具。

根据是否在创建时已知所有工具，动态工具有两种策略：

  • 过滤预注册工具
  • 运行时工具注册
过滤预注册工具

当所有可能工具在 Agent 创建时已知时，可以预先注册它们，然后根据状态、权限或上下文动态筛选哪些工具向模型暴露。

State

仅在对话达到特定里程碑后启用高级工具：

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:
    """根据会话状态筛选工具。"""
    state = request.state
    is_authenticated = state.get("authenticated", False)
    message_count = len(state["messages"])

    # 仅在认证后启用敏感工具
    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:
        # 在对话初期限制工具
        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]
)

import { createMiddleware, tool } from "langchain";
import { createDeepAgent } from "deepagents";

const stateBasedTools = createMiddleware({
    name: "StateBasedTools",
    wrapModelCall: (request, handler) => {
        const state = request.state as typeof request.state & {
            authenticated?: boolean;
        };
        const isAuthenticated = state.authenticated ?? false;
        const messageCount = state.messages.length;

        let filteredTools = request.tools;

        if (!isAuthenticated) {
            filteredTools = request.tools.filter(
                (t: any) => typeof t.name === "string" && t.name.startsWith("public_"),
            );
        } else if (messageCount < 5) {
            filteredTools = request.tools.filter(
                (t: any) => typeof t.name === "string" && t.name !== "advanced_search",
            );
        }

        return handler({ ...request, tools: filteredTools });
    },
});

const agent = await createDeepAgent({
    model: "claude-sonnet-4-20250514",
    tools: tools,
    middleware: [stateBasedTools] as any,
});
Store

根据存储中的用户偏好或功能开关筛选工具：

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:
    """根据 Store 偏好筛选工具。"""
    user_id = request.runtime.context.user_id

    # 从 Store 读取用户启用的功能
    store = request.runtime.store
    feature_flags = store.get(("features",), user_id)

    if feature_flags:
        enabled_features = feature_flags.value.get("enabled_tools", [])
        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()
)

import { createMiddleware } from "langchain";
import { createDeepAgent } from "deepagents";
import * as z from "zod";
import { InMemoryStore } from "@langchain/langgraph";

const contextSchema = z.object({
  userId: z.string(),
});

const storeBasedTools = createMiddleware({
  name: "StoreBasedTools",
  contextSchema,
  wrapModelCall: async (request, handler) => {
    const userId =
      (request.runtime?.context as { userId?: string } | undefined)?.userId ??
        "user-123";

    // 从 Store 读取用户启用的功能
    const runtimeStore = request.runtime?.store as InMemoryStore | undefined;
    const rawFlags = (await runtimeStore?.get(
      ["features"],
      userId as string,
    )) as unknown;
    const featureFlags = rawFlags as FeatureFlags | undefined;

    let filteredTools = request.tools;

    if (featureFlags) {
      const enabledFeatures = featureFlags.enabledTools || [];
      filteredTools = request.tools.filter((t) =>
        enabledFeatures.includes(t.name as string)
      );
    }

    return handler({ ...request, tools: filteredTools });
  },
});

const agent = await createDeepAgent({
  model: "claude-sonnet-4-20250514",
  backend: new StoreBackend(),
  store,
  checkpointer,
  tools,
  middleware: [storeBasedTools] as any,
});
运行时上下文

根据运行时上下文中的用户权限筛选工具：

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:
    """根据运行时上下文权限筛选工具。"""
    if request.runtime is None or request.runtime.context is None:
        user_role = "viewer"
    else:
        user_role = request.runtime.context.user_role

    if user_role == "admin":
        pass
    elif user_role == "editor":
        tools = [t for t in request.tools if t.name != "delete_data"]
        request = request.override(tools=tools)
    else:
        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
)

import * as z from "zod";
import { createMiddleware } from "langchain";
import { createDeepAgent } from "deepagents";

const contextSchema = z.object({
  userRole: z.string(),
});

const contextBasedTools = createMiddleware({
  name: "ContextBasedTools",
  contextSchema,
  wrapModelCall: (request, handler) => {
    const userRole = request.runtime.context.userRole;

    let filteredTools = request.tools;

    if (userRole === "admin") {
      // 管理员查看所有工具
    } else if (userRole === "editor") {
      filteredTools = request.tools.filter((t) => t.name !== "delete_data");
    } else {
      filteredTools = request.tools.filter(
        (t) => (t.name as string).startsWith("read_"),
      );
    }

    return handler({ ...request, tools: filteredTools });
  },
});

const agent = await createDeepAgent({
  model: "claude-sonnet-4-20250514",
  store,
  checkpointer,
  tools,
  middleware: [contextBasedTools] as any,
});

这种方法适用于：

  • 所有可能的工具在编译/启动时已知
  • 你希望基于权限、功能开关或会话状态进行过滤
  • 工具是静态的，但其可用性是动态的

请参见 Dynamically selecting tools 了解更多示例。

运行时工具注册

当工具在运行时动态发现或创建（例如从 MCP 服务器加载、基于用户数据生成或者从远程注册表获取）时，需要同时注册工具并动态处理它们的执行。

这需要两个中间件钩子：

  1. wrap_model_call - 将动态工具添加到请求中
  2. wrap_tool_call - 处理动态添加工具的执行
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, ModelRequest, ToolCallRequest

# 运行时将动态添加的工具
@tool
def calculate_tip(bill_amount: float, tip_percentage: float = 20.0) -> str:
    """计算账单小费。"""
    tip = bill_amount * (tip_percentage / 100)
    return f"Tip: ${tip:.2f}, Total: ${bill_amount + tip:.2f}"

class DynamicToolMiddleware(AgentMiddleware):
    """注册并处理动态工具的中间件。"""

    def wrap_model_call(self, request: ModelRequest, handler):
        updated = request.override(tools=[*request.tools, calculate_tip])
        return handler(updated)

    def wrap_tool_call(self, request: ToolCallRequest, handler):
        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],
    middleware=[DynamicToolMiddleware()],
)

# Agent 现在可以使用 get_weather 和 calculate_tip
result = agent.invoke({
    "messages": [{"role": "user", "content": "Calculate a 20% tip on $85"}]
})

import { createAgent, createMiddleware, tool } from "langchain";
import * as z from "zod";

// 运行时将动态添加的工具
const calculateTip = tool(
  ({ billAmount, tipPercentage = 20.0 }) => {
    const tip = billAmount * (tipPercentage / 100);
    return `Tip: $${tip.toFixed(2)}, Total: $${(billAmount + tip).toFixed(2)}`;
  },
  {
    name: "calculate_tip",
    description: "Calculate the tip amount for a bill.",
    schema: z.object({
      billAmount: z.number(),
      tipPercentage: z.number().optional(),
    }),
  }
);

const dynamicToolMiddleware = createMiddleware({
  name: "DynamicToolMiddleware",
  wrapModelCall: (request, handler) => {
    return handler({
      ...request,
      tools: [...request.tools, calculateTip],
    });
  },
  wrapToolCall: async (request, handler) => {
    if (request.tool.name === "calculate_tip") {
      return handler({
        ...request,
        tool: calculateTip,
      });
    }
    return handler(request);
  },
});

const agent = await createAgent({
  model: "gpt-4o",
  tools: [getWeather],
  middleware: [dynamicToolMiddleware],
});
工具错误处理

要自定义工具错误处理方式，请使用 wrap_tool_call 装饰器创建中间件：

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):
    """使用自定义消息处理工具执行错误。"""
    try:
        return handler(request)
    except Exception as e:
        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]
)

import { createAgent, createMiddleware, ToolMessage } from "langchain";

const handleToolErrors = createMiddleware({
  name: "HandleToolErrors",
  wrapToolCall: async (request, handler) => {
    try {
      return await handler(request);
    } catch (error) {
      return new ToolMessage({
        content: `Tool error: Please check your input and try again. (${error})`,
        tool_call_id: request.toolCall.id!,
      });
    }
  },
});

const agent = createAgent({
  model: "gpt-4.1",
  tools: [
    /* ... */
  ],
  middleware: [handleToolErrors],
});

当工具失败时，Agent 将返回一个带有自定义错误消息的 ToolMessage。

ReAct 循环中的工具使用

Agents 遵循 ReAct（“Reasoning + Acting”）模式，在简短的推理步骤和有针对性的工具调用之间交替，并将结果观察反馈到后续决策中，直到给出最终答案。

示例：ReAct 循环

提示：识别当前最流行的无线耳机，并验证其是否有库存。

  1. 推理：“热门性是时效性的，我需要使用搜索工具。”
  2. 行动：调用 search_products("wireless headphones")
  3. 工具返回库存信息
  4. 推理：“我需要确认排名第一的型号是否有货。”
  5. 行动：调用 check_inventory("WH-1000XM5")
  6. 输出最终答案

系统提示

你可以通过提供 prompt 来塑造 Agent 的任务方式。system_prompt 参数可以是字符串：

agent = create_agent(
    model,
    tools,
    system_prompt="You are a helpful assistant. Be concise and accurate."
)

const agent = createAgent({
  model,
  tools,
  systemPrompt: "You are a helpful assistant. Be concise and accurate.",
});

当未提供 system_prompt 时，Agent 会直接从消息中推断任务。system_prompt 参数可以接受字符串或 SystemMessage。使用 SystemMessage 可以更好地控制提示结构，这对于某些提供商特性（例如 Anthropic 的提示缓存）非常有用。

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": "You are an AI assistant tasked with analyzing literary works.",
            },
            {
                "type": "text",
                "text": "<the entire contents of 'Pride and Prejudice'>",
                "cache_control": {"type": "ephemeral"}
            }
        ]
    )
)

result = literary_agent.invoke(
    {"messages": [HumanMessage("Analyze the major themes in 'Pride and Prejudice'.")]}
)

在这个示例中，cache_control 字段为 {"type": "ephemeral"}，告诉 Anthropic 将该内容块缓存，从而减少重复请求的延迟和成本。

import { createAgent } from "langchain";
import { SystemMessage, HumanMessage } from "@langchain/core/messages";

const literaryAgent = createAgent({
  model: "anthropic:claude-sonnet-4-5",
  systemPrompt: new SystemMessage({
    content: [
      {
        type: "text",
        text: "You are an AI assistant tasked with analyzing literary works.",
      },
      {
        type: "text",
        text: "<the entire contents of 'Pride and Prejudice'>",
        cache_control: { type: "ephemeral" }
      }
    ]
  })
});

const result = await literaryAgent.invoke({
  messages: [new HumanMessage("Analyze the major themes in 'Pride and Prejudice'.")]}
);
动态系统提示

对于需要基于运行时上下文或 Agent 状态修改系统提示的高级用例，可以使用 middleware。

@dynamic_prompt 装饰器可创建根据模型请求生成系统提示的中间件：

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:
    """根据用户角色生成系统提示。"""
    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
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Explain machine learning"}]},
    context={"user_role": "expert"}
)

提示：有关消息类型和格式的更多信息，请参见 Messages。有关完整的中间件文档，请参见 Middleware。

名称

可以为 Agent 设置可选 name 参数。该名称在将 Agent 作为子图添加到 多 Agent 系统 时用作节点标识符：

agent = create_agent(
    model,
    tools,
    name="research_assistant"
)

警告：建议使用 snake_case（例如 research_assistant），而不是包含空格或特殊字符的名称。一些模型提供商会拒绝包含空格或特殊字符的名称。对工具名称也适用相同规则。

Invocation（调用）

你可以通过向 Agent 传递对其 State 更新来调用 Agent。所有 Agent 的状态中都包含一系列消息；要调用 Agent，请传入新的用户消息：

result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's the weather in San Francisco?"}]}
)

await agent.invoke({
  messages: [{ role: "user", content: "What's the weather in San Francisco?" }],
})

有关从 Agent 流式传输步骤或令牌的用法，请参见 Streaming 指南。

否则，Agent 遵循 LangGraph Graph API，支持所有关联方法，例如 stream 和 invoke。

提示：使用 LangSmith 跟踪、调试和评估你的 Agent。

Advanced concepts（高级概念）

Structured output（结构化输出）

在某些情况下，你可能希望 Agent 返回特定格式的输出。LangChain 通过 response_format 参数提供结构化输出策略。

ToolStrategy

ToolStrategy 使用人工工具调用来生成结构化输出。它适用于任何支持工具调用的模型。当提供商原生结构化输出不可用或不可靠时，应使用 ToolStrategy。

from pydantic import BaseModel
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy

class ContactInfo(BaseModel):
    name: str
    email: str
    phone: str

agent = create_agent(
    model="gpt-4.1-mini",
    tools=[search_tool],
    response_format=ToolStrategy(ContactInfo)
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "Extract contact info from: John Doe, john@example.com, (555) 123-4567"}]
})

result["structured_response"]
# ContactInfo(name='John Doe', email='john@example.com', phone='(555) 123-4567')
ProviderStrategy

ProviderStrategy 使用模型提供商的原生结构化输出生成。这更加可靠，但仅在提供商支持原生结构化输出时可用：

from langchain.agents.structured_output import ProviderStrategy

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

注意：从 langchain 1.0 起，仅传递一个 schema（例如 response_format=ContactInfo）会在模型支持原生结构化输出时默认使用 ProviderStrategy；否则会回退为 ToolStrategy。

import * as z from "zod";
import { createAgent } from "langchain";

const ContactInfo = z.object({
  name: z.string(),
  email: z.string(),
  phone: z.string(),
});

const agent = createAgent({
  model: "gpt-4.1",
  responseFormat: ContactInfo,
});

const result = await agent.invoke({
  messages: [
    {
      role: "user",
      content: "Extract contact info from: John Doe, john@example.com, (555) 123-4567",
    },
  ],
});

console.log(result.structuredResponse);
// {
//   name: 'John Doe',
//   email: 'john@example.com',
//   phone: '(555) 123-4567'
// }

提示：有关结构化输出的更多信息，请参见 Structured output。

Memory（记忆）

Agent 会自动通过消息状态维护会话历史。你也可以配置 Agent 使用自定义状态 schema，在对话期间记住额外信息。

存储在状态中的信息可以视为 Agent 的 短期记忆：

自定义状态 schema 必须扩展 AgentState 并作为 TypedDict。

定义自定义状态有两种方式：

  1. 通过 middleware（推荐）
  2. 通过 create_agent 上的 state_schema
通过 middleware 定义状态

当你的自定义状态需要被特定中间件钩子和附加到该中间件的工具访问时，请使用 middleware 定义自定义状态。

from langchain.agents import AgentState
from langchain.agents.middleware import AgentMiddleware
from typing import Any


class CustomState(AgentState):
    user_preferences: dict


class CustomMiddleware(AgentMiddleware):
    state_schema = CustomState
    tools = [tool1, tool2]

    def before_model(self, state: CustomState, runtime) -> dict[str, Any] | None:
        ...

agent = create_agent(
    model,
    tools=tools,
    middleware=[CustomMiddleware()]
)

# Agent 现在可以跟踪除消息之外的额外状态
result = agent.invoke({
    "messages": [{"role": "user", "content": "I prefer technical explanations"}],
    "user_preferences": {"style": "technical", "verbosity": "detailed"},
})
通过 state_schema 定义状态

使用 state_schema 参数作为快捷方式，定义仅在工具中使用的自定义状态。

from langchain.agents import AgentState


class CustomState(AgentState):
    user_preferences: dict

agent = create_agent(
    model,
    tools=[tool1, tool2],
    state_schema=CustomState
)

Streaming（流式传输）

如前所述，Agent 的 invoke 方法会返回最终响应。如果 Agent 执行多个步骤，这可能需要一段时间。要展示中间进度，可按出现顺序流式返回消息。

from langchain.messages import AIMessage, HumanMessage

for chunk in agent.stream({
    "messages": [{"role": "user", "content": "Search for AI news and summarize the findings"}]
}, stream_mode="values"):
    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]}")

const stream = await agent.stream(
  {
    messages: [{
      role: "user",
      content: "Search for AI news and summarize the findings"
    }],
  },
  { streamMode: "values" }
);

for await (const chunk of stream) {
  const latestMessage = chunk.messages.at(-1);
  if (latestMessage?.content) {
    console.log(`Agent: ${latestMessage.content}`);
  } else if (latestMessage?.tool_calls) {
    const toolCallNames = latestMessage.tool_calls.map((tc) => tc.name);
    console.log(`Calling tools: ${toolCallNames.join(", ")}`);
  }
}

提示：有关流式传输的更多细节，请参见 Streaming。

Middleware（中间件）

Middleware 为自定义 Agent 行为提供了强大的扩展能力，可以在执行的不同阶段进行配置。

中间件可以用于：

  • 在模型调用之前处理状态（例如消息修剪、上下文注入）
  • 修改或验证模型响应（例如 guardrails、内容过滤）
  • 使用自定义逻辑处理工具执行错误
  • 根据状态或上下文实现动态模型选择
  • 添加自定义日志、监控或分析

中间件与 Agent 执行无缝集成，允许你在关键点拦截并修改数据流，而无需更改核心 Agent 逻辑。

提示：有关完整的中间件文档，包括 @before_model、@after_model、@wrap_tool_call 等装饰器，请参见 Middleware。

Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

图像领域的预训练

图像预训练经历了“监督学习（ImageNet 时代）→ 自监督学习（对比学习与掩码建模时代）→ 多模态统一（CLIP 与大模型时代）”用更少标注、更低成本，获得更强的视觉理解、泛化与生成能力。未来，视觉预训练将与语言、音频、动作等多模态深度融合，迈向真正的通用人工智能。

avatar 2048 AI社区
cover

数据安全与研发效率，如何兼得？将“研发大脑”私有化，我们提升了30%的交付可控性

avatar 2048 AI社区
cover

Obsidian-Graphify-让你的笔记库自己长出知识图谱

avatar 2048 AI社区