代理是一个模型在循环中调用工具,直到给定任务完成。
Agent = Model + Harness Harness 的职责:在给定任务中,在正确的时间向模型提供正确上下文。
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 )
传入任意 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" )
可以通过向代理的 State 传入更新来调用代理。所有代理的状态中都包含一个消息序列 ;要调用代理,请传入一条新消息以及 thread_id,让代理可以持久化并恢复对话历史:
Google
OpenAI
Anthropic
OpenRouter
Fireworks
Baseten
Ollama
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),请将其作为 context 与 config 一起传入。使用 context_schema 定义这些数据的形状,并通过 runtime.context 访问:
Google
OpenAI
Anthropic
OpenRouter
Fireworks
Baseten
Ollama
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 context 和 Runtime 。
流式传输
前面展示了如何使用 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 ] } " )
配置 harness
create_agent 具有很强的扩展性。Middleware 是自定义的原语:每个部分处理一个关注点,在正确时刻接入代理循环,并可与其他部分自由组合。只选择用例需要的能力,跳过其余部分。
常见模式已作为一等 middleware 预构建。任何自定义能力都可以通过编写一个 middleware 实现。
当代理承担复杂工作时,需要在几个关键领域获得支持。Middleware 生态系统覆盖了这些领域:
执行环境 工具、文件系统、sandboxes 和代码执行
上下文管理 Summarization、memory、skills 和 prompt caching
规划与委派 用于并行、隔离工作的 todo lists 和 subagents
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 ())],
)
请参阅 FilesystemMiddleware 、Sandboxes 、Interpreters 。
上下文管理
每次模型调用都有固定上下文窗口。随着代理运行并累积历史、工具结果和中间步骤,该窗口会逐渐填满。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/" ]),
],
)
请参阅 SummarizationMiddleware 、MemoryMiddleware 、SkillsMiddleware 、Context engineering 。
规划与委派
复杂任务通常会超出单个上下文窗口可处理的范围。委派让主代理可以将工作拆分成多个部分,交给各自在隔离上下文中运行的 subagents,并专注于协调而不是执行。工作可以并行运行,主代理的上下文保持干净。
Google
OpenAI
Anthropic
OpenRouter
Fireworks
Baseten
Ollama
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" : [],
}
],
),
],
)
请参阅 SubAgentMiddleware 、Subagents 。
生产环境中的代理会遇到开发中很少出现的故障:rate limits、模型超时、瞬时 API 错误。容错 middleware 会在基础设施层处理这些问题,因此你的工具和业务逻辑不需要为每次调用包裹 try/catch。
Google
OpenAI
Anthropic
OpenRouter
Fireworks
Baseten
Ollama
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 ),
],
)
请参阅 ModelRetryMiddleware 、ToolRetryMiddleware 、Prebuilt middleware 。
Guardrails
有些策略不能只写在 prompt 中,它们需要无论模型怎么做都能确定性执行。Guardrails 会在数据流经代理循环时拦截它,在工具结果进入模型上下文之前应用合规规则或内容策略。
Google
OpenAI
Anthropic
OpenRouter
Fireworks
Baseten
Ollama
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" )],
)
请参阅 PIIMiddleware 、Prebuilt middleware 。
完全自主并不总是合适。Steering 让你无需重构代理,就能将人工放在特定决策点上,例如破坏性写入、昂贵 API 调用或任何需要判断的操作之前。代理会暂停并等待;人工批准、编辑或拒绝;执行继续。
Google
OpenAI
Anthropic
OpenRouter
Fireworks
Baseten
Ollama
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 })],
)
请参阅 HumanInTheLoopMiddleware 、Human-in-the-loop 。
create_deep_agent 会为长时间运行的编码和研究任务预组装这套栈(默认包含 filesystem、summarization、subagents 和 prompt caching)。完整预构建 harness 请参阅 Deep Agents 。
Middleware 资源: