先补齐必要概念
适合:会写 HTTP 接口,刚开始调用生成模型。
- 模型请求
- 包含模型 ID、任务指令、用户输入及可用工具;凭据与真实用户身份由调用端管理。
- 候选动作
- 模型给出的工具名与参数。它表达建议,执行端还要核对格式、权限和业务条件。
- 调用关联
- call_id 把某次函数请求与它的返回结果关联;业务幂等键需由业务另外定义。
- 完成状态
- HTTP、生成过程和业务任务各有状态;只有业务验收能说明任务完成。
- 响应样本
- 本实验由作者编写的 Responses 形状样本,用于检查程序分支;真实 API 以当次返回和官方契约为准。
原理怎样一步步成立?
- 组装完整请求
任务指令说明允许行为,用户输入描述目标,工具定义提供局部参数契约。
- 分类接口结果
先检查 HTTP 结果,再检查生成状态和 output 的条目类型,保留拒绝与不完整原因。
- 执行受限动作
从可信会话绑定身份,独立校验参数、访问权限和调用预算;本实验只允许读取订单。
- 回传与核对
保留原始输出条目,把函数结果以对应 call_id 回传,最后对照工具回执核对候选答案。
入门 · 理解原理
接口返回了 200,应用下一步该做什么?
本层目标:分清传输、生成和业务三种状态,说明模型输出如何成为应用输入。
从订单查询开始
假设用户要查询 o-1 的物流状态。普通后端接口查询数据库后就能返回状态;生成模型还可能先要求调用 read_order。程序需要把这个工具请求变成一次经过校验的查询,再将查询结果返回模型。整条链中,真实订单状态来自业务系统。
先画三个框:请求组装、模型响应、业务执行。请求中放任务、已获准证据和可用工具。当前租户、登录用户、密钥、预算放在服务端可信状态中;参数里出现一个 tenant,并不会自动改变登录身份。工具名称也只在配置好的列表内分发。
三种成功分别回答不同问题
| 观察 | 能支持的判断 | 下一步 |
|---|---|---|
| HTTP 200 | 服务端交付了一个响应 | 检查状态与条目 |
| status=completed | 这一轮生成已结束 | 分辨文本、工具候选和拒绝 |
| 获得工具回执 | 该次查询有可核对事实 | 检查最终答案是否使用同一事实 |
| 最终文本说已退款 | 模型提出了一句描述 | 核对实际业务账本与批准对象 |
本项目只读取订单和政策,最终交付一份待核对的解释,不生成真实退款或运输动作。页面显示文字与外部业务效果之间,必须保留明确边界。
先认识四种响应
output_text 提供候选文本;function_call 给出工具名、参数和 call_id;refusal 表达拒绝;incomplete 表示生成未完整结束。它们决定不同分支。不能把所有条目拼成字符串后靠关键词猜状态,也不能把半段工具参数补齐后执行。
实验采用非流式 Responses 形状样本,整体完成后才处理本轮工具候选。真实流式接口需按对应调用的完整状态解析;这要与网络是否已关闭分开。先完成本层分类,再进入实现层和输出生命周期的扩展阅读。
一个可验证的反例
o-2 属于另一个租户。模型给出完全合法的 {order_id: "o-2"},结构校验能够通过,但 execute_read 会按可信身份拒绝,结果中不会出现订单字段。这个现象说明参数形状与资源权限分别需要检查。
运行实验,观察反例
使用作者编写的非流式 Responses 形状样本、教学订单与可信身份模拟,验证请求组装、只读校验、call_id 回传、预算和答案核对。程序不调用在线模型或真实订单服务。
Python 3.10+ · 默认运行只使用标准库 · 在你的电脑运行
- 下载本页的 Agent 应用入门实验包,解压后进入 agent-application-lab-v1 目录。
- 使用 Python 3.10+ 执行上方命令;默认回放只需标准库与包内数据。
- 对照输出与检查点,再运行 python3 -m unittest test_application -v,并完成当前层任务。
python3 model_response_contract.py查看本入口脚本
"""A bounded read-only tool loop using authored Responses-shaped fixtures.
This file makes no API requests. The course separately documents the real API request.
The fixture response shape is deliberately limited to the non-streaming cases below.
"""
import json
ORDERS = {
"o-1": {"tenant": "shop-a", "user": "u-1", "status": "delivered"},
"o-2": {"tenant": "shop-b", "user": "u-2", "status": "processing"},
}
PRINCIPAL = {"tenant": "shop-a", "user": "u-1"}
TOOL = {
"type": "function", "name": "read_order", "description": "Read an authorized order status.",
"strict": True, "parameters": {"type": "object", "properties": {"order_id": {"type": "string"}},
"required": ["order_id"], "additionalProperties": False},
}
def build_request(order_id, model="reader-selected-model"):
return {"model": model, "input": [
{"role": "developer", "content": "Use read_order for order facts. Tool access is checked by the application. Return an object with order_id, status and source. Do not perform writes."},
{"role": "user", "content": "What is the status of order " + order_id + "?"},
], "tools": [TOOL], "max_output_tokens": 500}
def classify_response(response, http_status=200):
if http_status != 200:
return {"kind": "http_error", "retryable": http_status == 429 or 500 <= http_status < 600}
if not isinstance(response, dict):
return {"kind": "unsupported"}
status = response.get("status")
if status == "incomplete":
return {"kind": "incomplete", "reason": response.get("incomplete_details")}
if status in ("failed", "cancelled"):
return {"kind": status}
if status != "completed" or not isinstance(response.get("output"), list):
return {"kind": "unsupported"}
calls, texts = [], []
for item in response["output"]:
if not isinstance(item, dict):
return {"kind": "unsupported"}
if item.get("type") == "function_call":
if not all(isinstance(item.get(key), str) and item[key] for key in ("call_id", "name", "arguments")):
return {"kind": "invalid_tool_call"}
calls.append(item)
elif item.get("type") == "message":
if item.get("status") not in (None, "completed") or not isinstance(item.get("content"), list):
return {"kind": "unsupported"}
for content in item["content"]:
if not isinstance(content, dict):
return {"kind": "unsupported"}
if content.get("type") == "refusal":
return {"kind": "refused"}
if content.get("type") != "output_text" or not isinstance(content.get("text"), str):
return {"kind": "unsupported"}
texts.append(content["text"])
elif item.get("type") != "reasoning":
return {"kind": "unsupported"}
if calls:
return {"kind": "tool_requests", "calls": calls}
return {"kind": "text", "text": "".join(texts)} if texts else {"kind": "empty"}
def execute_read(call, principal):
if call["name"] != "read_order":
raise ValueError("unknown_tool")
try:
args = json.loads(call["arguments"])
except (ValueError, TypeError):
raise ValueError("invalid_arguments") from None
if not isinstance(args, dict) or set(args) != {"order_id"} or not isinstance(args["order_id"], str):
raise ValueError("invalid_arguments")
order = ORDERS.get(args["order_id"])
if not order or order["tenant"] != principal["tenant"] or order["user"] != principal["user"]:
# No protected resource fields are returned when access is denied.
raise ValueError("permission_denied")
return {"order_id": args["order_id"], "status": order["status"], "source": "read_order"}
def run_loop(responses, principal=None, max_turns=3, max_calls=2):
principal = principal or PRINCIPAL
trace, results, seen_ids, calls_used = [], [], set(), 0
transcript = build_request("o-1")["input"]
for turn, response in enumerate(responses, 1):
if turn > max_turns:
break
event = classify_response(response)
trace.append({"turn": turn, "kind": event["kind"]})
if event["kind"] == "tool_requests":
# Preserve output items, including reasoning items when present, before
# appending call-correlated tool results to the next request input.
transcript.extend(response["output"])
if calls_used + len(event["calls"]) > max_calls:
return {"status": "budget_exhausted", "trace": trace, "results": results, "calls": calls_used}
for call in event["calls"]:
if call["call_id"] in seen_ids:
return {"status": "duplicate_call_id", "trace": trace, "results": results, "calls": calls_used}
seen_ids.add(call["call_id"])
calls_used += 1
try:
result = execute_read(call, principal)
except ValueError as error:
return {"status": str(error), "trace": trace, "results": results, "calls": calls_used}
result_item = {"type": "function_call_output", "call_id": call["call_id"], "output": json.dumps(result, sort_keys=True)}
results.append(result_item)
transcript.append(result_item)
trace[-1]["nextInputTypes"] = [item.get("type", "message") for item in transcript]
elif event["kind"] == "text":
try:
candidate = json.loads(event["text"])
except ValueError:
candidate = None
confirmed = isinstance(candidate, dict) and set(candidate) == {"order_id", "status", "source"} and any(candidate == json.loads(result["output"]) for result in results)
return {"status": "answer_ready" if confirmed else "unverified_answer", "trace": trace, "results": results, "calls": calls_used}
else:
return {"status": event["kind"], "trace": trace, "results": results, "calls": calls_used}
return {"status": "budget_exhausted", "trace": trace, "results": results, "calls": calls_used}
def tool_response(order_id="o-1", call_id="call-1", name="read_order"):
return {"status": "completed", "output": [{"type": "function_call", "call_id": call_id, "name": name, "arguments": json.dumps({"order_id": order_id})}]}
def text_response(candidate):
return {"status": "completed", "output": [{"type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": json.dumps(candidate)}]}]}
def demo():
normal = run_loop([tool_response(), text_response({"order_id": "o-1", "status": "delivered", "source": "read_order"})])
denied = run_loop([tool_response("o-2")])
loop = run_loop([tool_response(call_id="c-" + str(i)) for i in range(5)], max_turns=2)
refused = {"status": "completed", "output": [{"type": "message", "content": [{"type": "refusal", "refusal": "fixture refusal"}]}]}
return {"normal": normal["status"], "toolResultCallId": normal["results"][0]["call_id"],
"crossTenant": denied["status"], "deniedResults": len(denied["results"]),
"boundedLoop": loop["status"], "boundedCalls": loop["calls"],
"refusal": classify_response(refused)["kind"],
"incomplete": classify_response({"status": "incomplete", "incomplete_details": {"reason": "max_output_tokens"}})["kind"],
"scope": "authored_responses_fixtures_no_api"}
if __name__ == "__main__":
print(json.dumps(demo(), ensure_ascii=False, sort_keys=True))
本地运行的预期输出
{"boundedCalls": 2, "boundedLoop": "budget_exhausted", "crossTenant": "permission_denied", "deniedResults": 0, "incomplete": "incomplete", "normal": "answer_ready", "refusal": "refused", "scope": "authored_responses_fixtures_no_api", "toolResultCallId": "call-1"}- normal=answer_ready,最终对象与该次 read_order 回执一致。
- crossTenant=permission_denied 且 deniedResults=0。
- boundedLoop=budget_exhausted 且 boundedCalls=2。
- refusal 与 incomplete 使用独立处理路径。
本层验收任务
把正常文本、合法工具候选、跨租户工具候选、拒绝、输出截断、HTTP 429 六种输入分到处理路径,并写明每种情况下程序已经知道的事实。
完成后逐条核对
- HTTP 状态与生成状态分别列出。
- 工具候选在授权前没有业务回执。
- 拒绝与不完整输入不会进入普通答案成功路径。
- 能解释合法 JSON 为何仍会被资源权限拒绝。
保存自己的过程、代码与结果。这里提供验收要求,暂不自动评分或保存课程掌握状态。
收起答案,检查理解
HTTP 200、status=completed,但 output 中是 function_call,订单查询已经完成了吗?
展开参考推导
目前只有完整候选动作。程序还需在可信身份下校验并执行 read_order,拿到回执后回传对应 call_id,再核对答案与订单事实。completed 只描述这轮生成。
延伸原理与知识练习
遇到不熟悉的原理,先阅读实现、连续追问和迁移案例,再独立说明前提与边界。作答与笔记保存到原有账号记录。
本专题的全部关联解析与练习(2 道)
依据与验证范围
原理依据来自公开资料;数字、案例和任务是本站教学设计。离线实验验证本页注明的范围,学习效果仍需通过独立任务与反馈判断。