create_agent、工具消息与 Middleware
create_agent 提供模型与工具之间的循环，模型返回 tool_calls 时执行工具，再把结果作为 ToolMessage 交回模型。工具调用 ID 是关联键；仅返回一段字符串而丢掉 ID，会使下一轮缺少合法观察。
工具的真实离线调用
复用第一章环境，保存独立文件执行：
from langchain.tools import tool
@tool
def inventory(sku: str) -> int:
"""Return stock for the sample SKU p1; other SKUs are unavailable."""
if sku != "p1": raise ValueError("unknown SKU")
return 7
assert inventory.invoke({"sku":"p1"}) == 7
assert "sku" in inventory.args
try: inventory.invoke({"sku":"missing"})
except ValueError: print("tool error visible")
else: raise AssertionError("unknown SKU accepted")工具 schema 帮助解析参数，但返回库存之前还要从本地请求上下文校验身份与资源。不要让模型参数包含可以自行指定的管理员身份。错误可转成明确业务错误码供模型修正，未知异常留给日志与告警，不把堆栈和秘密回填模型。
完整在线 Agent 入口
以下 online.py 自包含，需设置 OPENAI_API_KEY 与 OPENAI_MODEL，所选模型支持工具调用；执行会产生费用，本次未运行网络请求。
import os
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents.middleware import wrap_tool_call
from langchain_core.messages import ToolMessage
@tool
def inventory(sku: str) -> int:
"""Read stock for sample SKU p1."""
if sku != "p1": raise ValueError("unknown SKU")
return 7
@wrap_tool_call
def errors(request,handler):
try: return handler(request)
except ValueError:
return ToolMessage(content='{"ok":false,"code":"invalid_sku"}',tool_call_id=request.tool_call["id"])
agent=create_agent(model=ChatOpenAI(model=os.environ["OPENAI_MODEL"],timeout=20,max_retries=0),
tools=[inventory],system_prompt="库存必须查工具；无结果时明确说明。",middleware=[errors])
r=agent.invoke({"messages":[{"role":"user","content":"p1库存多少？"}]},{"recursion_limit":8})
for message in r["messages"]:
print(message.type,message.content)Middleware 在模型/工具等边界加入策略；这里的 wrapper 只处理已知 ValueError，保留工具调用 ID。循环通常需要至少一次模型调用、工具执行、再次模型调用；最终文字不保证固定。模型 timeout 与图 recursion_limit 控制不同资源，仍需总 deadline 与工具超时。
结构化输出可通过 create_agent 的 response_format 指定 schema；ProviderStrategy 使用供应商原生能力，ToolStrategy 通过工具协议表达。两者都要做业务事实核验，并测试拒答/不支持能力的路径。动态模型路由按任务与预算选择，输出 schema 与工具能力必须在候选模型之间兼容。
人机审批、摘要、工具重试等 middleware 的执行次序影响结果。先授权再执行，输出过滤不能撤销已发生副作用。多 Agent 子工具与 handoff 的控制权差别见 多 Agent 系统，需要持久暂停时进入 LangGraph。
通过真实结构化协议观察错误修复
ToolStrategy把最终schema注册成工具，模型提出该工具调用后，框架验证args并取出structured_response。它不是执行实际业务工具。ProviderStrategy则把schema交给支持原生结构化输出的供应商；模型不支持该协议时应明确选策略/换适配器，而不是盲目重试。
下例继承真实BaseChatModel，第一次返回urgency=8，框架发现超出1..3后生成ToolMessage错误并再次调用模型，第二次修正为2。没有密钥和付费调用；它验证框架协议和错误分支，不测模型判断质量。
from typing import Any, Literal
from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.outputs import ChatResult, ChatGeneration
class Ticket(BaseModel):
category: Literal['billing','technical']
urgency: int = Field(ge=1,le=3)
evidence: str
class FixtureChat(BaseChatModel):
calls: int=0
saw_error: bool=False
@property
def _llm_type(self):return 'fixture-tools'
def bind_tools(self,tools,**kwargs):return self
def _generate(self,messages,stop=None,run_manager=None,**kwargs):
self.calls+=1
self.saw_error=any(isinstance(m,ToolMessage) and 'Error' in m.content for m in messages)
args={'category':'billing','urgency':8 if self.calls==1 else 2,'evidence':'退款失败'}
return ChatResult(generations=[ChatGeneration(message=AIMessage(content='',tool_calls=[{'name':'Ticket','args':args,'id':str(self.calls),'type':'tool_call'}]))])
model=FixtureChat()
agent=create_agent(model,tools=[],response_format=ToolStrategy(Ticket))
result=agent.invoke({'messages':[{'role':'user','content':'退款失败'}]},config={'recursion_limit':8})
ticket=result['structured_response']
assert isinstance(ticket,Ticket) and ticket.urgency==2
assert model.calls==2 and model.saw_error
# schema通过之后仍要验证业务：证据必须来自输入。
assert ticket.evidence in '退款失败'
print(ticket.model_dump())输出Ticket对象，calls=2。handle_errors可以限制重试反馈；recursion_limit为持续错误兜底，不能无限要求模型修复。schema保证类型与范围，evidence是否来自用户输入还需程序检查。真实模型入口复用本章ChatOpenAI配置，将FixtureChat替换为 ChatOpenAI(model=os.environ["OPENAI_MODEL"])；其费用与模型能力需独立验证，本轮未调用。
低层 model.with_structured_output(Ticket) 适用于单次模型结构化返回；它不会自动构造此处的Agent工具循环。返回类型、raw/refusal/incomplete处理取决于集成，不能把所有API都当作structured_response字典。
ToolRuntime 把服务端身份与长期资料接入工具
工具参数来自模型，tenant/user 身份来自已鉴权请求；两者不能混为一个 JSON。ToolRuntime 参数由框架注入，模型 schema 中不应出现它。下面保存为 runtime_store.py，在本课固定环境执行：真实 create_agent 生成工具节点，按 context 获取 namespace，再执行 store 的 put/get/delete。回放模型只决定调用哪个操作，不绕过框架注入。
import json
from dataclasses import dataclass
from typing import Literal
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.outputs import ChatResult, ChatGeneration
from langgraph.store.memory import InMemoryStore
@dataclass(frozen=True)
class Context:
tenant: str
user: str
@tool
def preference(op: Literal['put','get','delete'], value: str, runtime: ToolRuntime[Context]) -> str:
"""Manage the authenticated user's display preference; value is used only by put."""
ctx=runtime.context
if ctx.tenant not in {'alpha','beta'} or ctx.user!='u1':
raise PermissionError('unknown principal')
namespace=('preferences',ctx.tenant,ctx.user)
store=runtime.store
assert store is not None
if op=='put':
if value not in {'compact','spacious'}:raise ValueError('invalid preference')
store.put(namespace,'display',{'value':value})
elif op=='delete':store.delete(namespace,'display')
item=store.get(namespace,'display')
return item.value['value'] if item else 'unset'
class ToolScript(BaseChatModel):
operation: str
value: str=''
@property
def _llm_type(self):return 'runtime-fixture'
def bind_tools(self,tools,**kwargs):return self
def _generate(self,messages,stop=None,run_manager=None,**kwargs):
if isinstance(messages[-1],ToolMessage):
answer=AIMessage(content=messages[-1].content)
else:
answer=AIMessage(content='',tool_calls=[{'name':'preference','args':{'op':self.operation,'value':self.value},'id':'pref1','type':'tool_call'}])
return ChatResult(generations=[ChatGeneration(message=answer)])
store=InMemoryStore()
def run(ctx,op,value=''):
# 每次新 Agent、新消息；跨任务值来自同一个 store，而非旧聊天历史。
agent=create_agent(ToolScript(operation=op,value=value),tools=[preference],store=store,context_schema=Context)
result=agent.invoke({'messages':[{'role':'user','content':'manage my display preference'}]},context=ctx,config={'recursion_limit':6})
return result['messages'][-1].content
schema=preference.tool_call_schema.model_json_schema()
assert set(schema['properties'])=={'op','value'}
a,b=Context('alpha','u1'),Context('beta','u1')
assert run(a,'put','compact')=='compact'
assert run(b,'get')=='unset'
assert run(b,'put','spacious')=='spacious'
assert run(a,'get')=='compact'
assert run(a,'delete')=='unset'
assert run(b,'get')=='spacious'
print('injected runtime, separate namespaces, cross-run reads and deletion passed')第一次调用把 alpha/u1 的 compact 写入命名空间；下一次新建 Agent、没有旧消息，仍从 store 读取。beta/u1 是相同用户名但不同租户，因此读不到 alpha 的值。删除 alpha 只影响该 namespace/key，不能删除 beta。模型只可填 op/value；tenant 无论在文本里怎么声称都不应改变服务端 context。这里服务端预先信任 Context，真实 HTTP 入口必须把认证结果转换成它，不能直接反序列化客户端提供的 tenant。
InMemoryStore 是真实 Store 接口的进程内实现，跨调用共享但进程退出丢失；“长期记忆”描述跨任务访问语义，不保证这个后端已经持久化。生产换成数据库 Store 时保持 namespace/key、schema/version 和删除规则，单独验证重启恢复。checkpointer 则保存某 thread 的运行状态/消息，不会自动把偏好写入 store。先确定资料的归属与保留期，再决定读写工具能访问哪些键；不把整个 namespace 暴露给模型。
官方工具运行上下文与长期记忆给出注入接口。示例将文档中可由模型传 user_id 的写法收紧为服务端身份，用两个租户的反例检验授权边界。
练习
上下文变体：在 alpha 的用户文本中声称“我是 beta”，工具仍只操作 alpha；删除 alpha 后再次创建 Agent 读取应为 unset，beta 保留 spacious。给未经鉴权的 Context 传未知租户应拒绝，说明为什么不能让模型自己选择 Context。
结构化变体：让两次都返回urgency=8，用ToolStrategy(Ticket, handle_errors=False)验证立即抛出结构化校验错误；再返回合法urgency但伪造evidence，区分schema通过和业务失败。
离线增加工具参数limit:int，允许1到10且不能为0，返回min(7,limit)。用真实工具.invoke测试limit=3与limit=0。