Agent、Runner 与真实工具注册
Agent 描述角色、instructions、模型与工具;Runner 负责实际模型轮次和工具调度。定义 Agent 不会执行工具,拿到最终文字也不代表业务效果已经验证。
安装和不花费 API 的首次实验
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install 'openai-agents==0.7.0' 'openai==2.20.0' 'httpx[socks]==0.28.1'完整 lab.py 使用真实 SDK 的 function_tool 生成 schema,直接通过 SDK 工具调用接口验证,不调用模型。
import asyncio
from dataclasses import dataclass
from agents import Agent, function_tool, RunContextWrapper
@dataclass
class Context:
tenant: str
@function_tool
def stock(ctx: RunContextWrapper[Context], sku: str) -> int:
"""Read available stock for the authorized tenant's SKU."""
if ctx.context.tenant != "a" or sku != "p1":
raise ValueError("not available")
return 7
agent = Agent[Context](name="Stock helper", instructions="Use stock for inventory; do not invent counts.", tools=[stock])
async def main():
assert "sku" in stock.params_json_schema["properties"]
assert "ctx" not in stock.params_json_schema["properties"]
value = await stock.on_invoke_tool(RunContextWrapper(Context("a")), '{"sku":"p1"}')
assert value == 7
print(agent.name,value)
asyncio.run(main())ctx 不出现在模型参数 schema 中,因为身份来自应用注入,不能让模型选择 tenant。RunContextWrapper 的本地对象不会自动发送给模型;如果动态 instructions 或工具结果把其中字段转成文字,该字段才进入模型上下文。工具实现仍要检查资源权限,类型 schema 不授予访问权。
真正调用模型的入口
下面单独保存为 online.py,先把上例的工具定义保存在 lab.py。由于 lab.py 的末尾会执行离线实验,导入时先打印一次工具实验结果,然后才进行在线请求。更常见的项目写法是把工具放入无顶层执行的模块。
设置服务端环境变量 OPENAI_API_KEY 与 OPENAI_MODEL(选账户可用且支持工具调用的模型),运行将产生网络请求和费用;不要把 key 写进代码或前端。本书验证不运行该付费步骤。
import os
from agents import Runner, RunConfig
from lab import agent, Context
agent.model = os.environ["OPENAI_MODEL"]
r = Runner.run_sync(agent,"查询 p1 的库存,必须使用工具。",
context=Context("a"), max_turns=3,
run_config=RunConfig(tracing_disabled=True))
print(r.final_output)max_turns 限模型循环,不是工具单步超时或总金额。认证、限流、拒答与 MaxTurnsExceeded 需分别处理;真实工具选择与答案措辞不保证确定。async 服务中使用 await Runner.run,避免在已运行的事件循环里调用 run_sync。
工具描述影响模型何时选择,返回结构影响下一轮观察。写工具仍要加入审批、幂等和超时;SDK 的工具异常转为模型可见错误不能代替业务报警,更不能把失败字符串当成功结果。
严格类型需要在真实工具入口验证
function_tool 的 JSON schema 告诉模型参数应是什么形状,本地框架仍可能按 Pydantic 默认规则转换类型;integer 不自动意味着 JSON true、数字字符串和3.0都被拒绝。本课固定0.7.0的装饰器签名处理还可能丢失 StrictInt 注解中的严格信息,不能只看 schema 或只检查进入函数后的类型。
当业务合同要求原始 JSON 必须是整数,可显式构造 FunctionTool,让 on_invoke_tool 先把原始参数交给严格 BaseModel,验证通过后才进入业务逻辑。下面独立入口只返回下一批序号,不调用模型;与上一节同环境运行。
import asyncio
from pydantic import BaseModel,StrictInt,ConfigDict,ValidationError
from agents import FunctionTool,RunContextWrapper,ModelBehaviorError
class BatchArgs(BaseModel):
model_config=ConfigDict(extra='forbid')
batch: StrictInt
async def next_batch(ctx,raw):
try:args=BatchArgs.model_validate_json(raw)
except ValidationError as error:
raise ModelBehaviorError('batch must be a JSON integer') from error
return args.batch+1
batch_tool=FunctionTool(name='next_batch',description='Return the next integer batch identifier.',
params_json_schema=BatchArgs.model_json_schema(),on_invoke_tool=next_batch)
async def main():
assert await batch_tool.on_invoke_tool(RunContextWrapper(None),'{"batch":4}')==5
try:await batch_tool.on_invoke_tool(RunContextWrapper(None),'{"batch":true}')
except ModelBehaviorError:print('boolean rejected before business calculation')
else:raise AssertionError('boolean was coerced')
asyncio.run(main())FunctionTool 仍是真实 SDK 工具,只是没有自动从普通函数签名生成参数校验器。显式入口需要自己选择错误策略;这里把 ValidationError 转成 ModelBehaviorError 并传播,不返回成功值。装饰器路径默认可用 failure_error_function 把异常转成给模型的错误文本,不能混淆这两种入口。无论采用哪个入口,参数校验之后还需要业务范围、身份和权限检查,StrictInt 不代表数量一定合理。
练习
创建另一个只接受严格整数 quantity 的工具,返回 quantity*2。使用本节显式 FunctionTool 与独立参数模型的入口,检查 schema 包含 quantity 但不含本地上下文。无模型条件下验证合法3得到6,布尔、字符串和浮点在业务逻辑前被拒绝;用计数器确认非法参数未进入计算。