代理是一个模型在循环中调用工具,直到给定任务完成。 核心代理循环图
Agent = Model + HarnessHarness 的职责:在给定任务中,在正确的时间向模型提供正确上下文。
Harness 是该循环周围的一切:模型、prompt、工具,以及任何塑造其行为的 middleware。 create_agent 是高度可配置的 harness。最简单的形式如下:
from langchain.agents import create_agent

agent = create_agent("openai:gpt-5.4", tools=tools)
直接配置基础项:model=tools=system_prompt=。如需更高级能力,请使用 middleware 扩展 harness。

核心组件

Model

传入模型标识符字符串("provider:model")或已初始化的模型实例。参数、provider 设置和动态模型选择请参阅 Models
from langchain.agents import create_agent

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

Tools

传入任意 Python callable、LangChain tool 或 tool dict。工具定义、context 访问和动态工具选择请参阅 Tools
from langchain.tools import tool

@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Results for: {query}"

agent = create_agent("openai:gpt-5.4", tools=[search])

System prompt

塑造代理处理任务的方式。接受字符串或 SystemMessage。如需运行时动态 prompts,请使用 middleware
agent = create_agent(
    "openai:gpt-5.4",
    tools=tools,
    system_prompt="You are a helpful assistant. Be concise and accurate.",
)

结构化输出

使用 response_format= 从代理返回经过验证的 schema。策略和示例请参阅 Structured output
from pydantic import BaseModel
from langchain.agents import create_agent

class Answer(BaseModel):
    summary: str
    confidence: float

agent = create_agent("openai:gpt-5.4", tools=tools, response_format=Answer)
result = agent.invoke({"messages": [{"role": "user", "content": "Summarize AI trends"}]})
result["structured_response"]  # Answer(summary=..., confidence=...)

Name

可选标识符。在 multi-agent 系统中将此代理作为 subgraph 嵌入时,它会用作节点名称。
agent = create_agent("openai:gpt-5.4", tools=tools, name="research_assistant")
要用自定义字段扩展代理的状态 schema,请在 create_agent 上使用 state_schema,或通过 middleware 定义它。详情请参阅 Memory

调用

使用 LangSmith 跟踪该循环的每一步、调试工具调用并评估代理输出。按照 tracing quickstart 完成设置。也建议设置 LangSmith Engine,它会监控 traces、检测问题并提出修复建议。
可以通过向代理的 State 传入更新来调用代理。所有代理的状态中都包含一个消息序列;要调用代理,请传入一条新消息以及 thread_id,让代理可以持久化并恢复对话历史:
from langchain.agents import create_agent
from langchain_core.utils.uuid import uuid7
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model="google_genai:gemini-3.5-flash",
    tools=[],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": str(uuid7())}}

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

# A follow-up turn on the same conversation: reuse the same thread_id to keep history
result = agent.invoke(
    {"messages": [{"role": "user", "content": "What about tomorrow?"}]},
    config=config,
)
使用 thread_id 持久化对话历史要求代理配置 checkpointer。部署到 LangSmith 时,会自动配置 checkpointer。在本地,请显式传入一个,例如 create_agent(..., checkpointer=InMemorySaver())
如果还需要向工具和 middleware 传入每次运行的配置(例如用户 ID、API keys 或 feature flags),请将其作为 contextconfig 一起传入。使用 context_schema 定义这些数据的形状,并通过 runtime.context 访问:
from dataclasses import dataclass

from langchain.agents import create_agent
from langchain_core.utils.uuid import uuid7
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver


@dataclass
class Context:
    user_id: str


agent = create_agent(
    model="google_genai:gemini-3.5-flash",
    tools=[],
    context_schema=Context,
    checkpointer=InMemorySaver(),
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's the weather in San Francisco?"}]},
    config={"configurable": {"thread_id": str(uuid7())}},
    context=Context(user_id="user-123"),
)
thread_id 限定对话范围(消息历史、checkpoints),而 context 携带工具和 middleware 在调用时读取的每次运行数据。二者通常一起传入。更多信息请参阅 tool contextRuntime

流式传输

前面展示了如何使用 invoke 调用代理以获取最终响应。如果代理执行多个步骤,可能需要一段时间。要显示中间进度,可以在消息产生时将其流式传回。
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"):
    # 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]}")
更多流式传输详情请参阅 Streaming

配置 harness

create_agent 具有很强的扩展性。Middleware 是自定义的原语:每个部分处理一个关注点,在正确时刻接入代理循环,并可与其他部分自由组合。只选择用例需要的能力,跳过其余部分。 常见模式已作为一等 middleware 预构建。任何自定义能力都可以通过编写一个 middleware 实现。 Middleware 生命周期图 当代理承担复杂工作时,需要在几个关键领域获得支持。Middleware 生态系统覆盖了这些领域:

执行环境

工具、文件系统、sandboxes 和代码执行

上下文管理

Summarization、memory、skills 和 prompt caching

规划与委派

用于并行、隔离工作的 todo lists 和 subagents

容错

重试、fallbacks 和调用限制

Guardrails

PII detection 和内容控制

引导

高影响动作前的 human-in-the-loop 审批

执行环境

当代理可以采取动作,而不仅仅是生成文本时,它们才真正有用。执行环境为代理提供工作区:可调用的工具、可跨轮次读写文件的文件系统,以及用于运行脚本或 shell 命令的代码执行能力。
from langchain.agents import create_agent
from deepagents.backends import StateBackend
from deepagents.middleware import FilesystemMiddleware

agent = create_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[search],
    middleware=[FilesystemMiddleware(backend=StateBackend())],
)
请参阅 FilesystemMiddlewareSandboxesInterpreters

上下文管理

每次模型调用都有固定上下文窗口。随着代理运行并累积历史、工具结果和中间步骤,该窗口会逐渐填满。Summarization 会在溢出前压缩历史;memory 会在启动时加载持久指令,让知识跨会话保留;skills 会按需浮现领域知识,而不是预先加载所有内容。
from deepagents.backends import StateBackend
from deepagents.middleware import FilesystemMiddleware, MemoryMiddleware, SkillsMiddleware, SummarizationMiddleware

backend = StateBackend()
model = "anthropic:claude-sonnet-4-6"

agent = create_agent(
    model=model,
    tools=[search],
    middleware=[
        FilesystemMiddleware(backend=backend),
        SummarizationMiddleware(model=model, backend=backend),
        MemoryMiddleware(backend=backend, sources=["./AGENTS.md"]),
        SkillsMiddleware(backend=backend, sources=["./skills/"]),
    ],
)
请参阅 SummarizationMiddlewareMemoryMiddlewareSkillsMiddlewareContext engineering

规划与委派

复杂任务通常会超出单个上下文窗口可处理的范围。委派让主代理可以将工作拆分成多个部分,交给各自在隔离上下文中运行的 subagents,并专注于协调而不是执行。工作可以并行运行,主代理的上下文保持干净。
from deepagents.backends import StateBackend
from deepagents.middleware import FilesystemMiddleware
from deepagents.middleware.subagents import SubAgentMiddleware
from langchain.agents import create_agent
from langchain.agents.middleware import TodoListMiddleware
from langchain.tools import tool


@tool
def search(query: str) -> str:
    """Search for a query and return a short summary."""
    return f"Search results for: {query}"


backend = StateBackend()

agent = create_agent(
    model="google_genai:gemini-3.5-flash",
    tools=[search],
    middleware=[
        FilesystemMiddleware(backend=backend),
        TodoListMiddleware(),
        SubAgentMiddleware(
            backend=backend,
            subagents=[
                {
                    "name": "researcher",
                    "description": "Searches and returns a structured summary.",
                    "system_prompt": "Use the search tool to research the question and summarize key points.",
                    "tools": [search],
                    "model": "anthropic:claude-sonnet-4-6",
                    "middleware": [],
                }
            ],
        ),
    ],
)
请参阅 SubAgentMiddlewareSubagents

容错

生产环境中的代理会遇到开发中很少出现的故障:rate limits、模型超时、瞬时 API 错误。容错 middleware 会在基础设施层处理这些问题,因此你的工具和业务逻辑不需要为每次调用包裹 try/catch。
from langchain.agents import create_agent
from langchain.agents.middleware import ModelRetryMiddleware, ToolRetryMiddleware
from langchain.tools import tool


@tool
def search(query: str) -> str:
    """Search for a query and return a short summary."""
    return f"Search results for: {query}"


agent = create_agent(
    model="google_genai:gemini-3.5-flash",
    tools=[search],
    middleware=[
        ModelRetryMiddleware(max_retries=3),
        ToolRetryMiddleware(max_retries=2),
    ],
)
请参阅 ModelRetryMiddlewareToolRetryMiddlewarePrebuilt middleware

Guardrails

有些策略不能只写在 prompt 中,它们需要无论模型怎么做都能确定性执行。Guardrails 会在数据流经代理循环时拦截它,在工具结果进入模型上下文之前应用合规规则或内容策略。
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
from langchain.tools import tool


@tool
def search(query: str) -> str:
    """Search for a query and return a short summary."""
    return f"Search results for: {query}"


agent = create_agent(
    model="google_genai:gemini-3.5-flash",
    tools=[search],
    middleware=[PIIMiddleware("email")],
)
请参阅 PIIMiddlewarePrebuilt middleware

引导

完全自主并不总是合适。Steering 让你无需重构代理,就能将人工放在特定决策点上,例如破坏性写入、昂贵 API 调用或任何需要判断的操作之前。代理会暂停并等待;人工批准、编辑或拒绝;执行继续。
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool


@tool
def search(query: str) -> str:
    """Search for a query and return a short summary."""
    return f"Search results for: {query}"


agent = create_agent(
    model="google_genai:gemini-3.5-flash",
    tools=[search],
    middleware=[HumanInTheLoopMiddleware(interrupt_on={"write_file": True})],
)
请参阅 HumanInTheLoopMiddlewareHuman-in-the-loop
create_deep_agent 会为长时间运行的编码和研究任务预组装这套栈(默认包含 filesystem、summarization、subagents 和 prompt caching)。完整预构建 harness 请参阅 Deep Agents
Middleware 资源: