import argparse import asyncio import importlib.metadata import json import os from pathlib import Path import time def save(path, data): temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n") temporary.replace(path) async def execute(model_id, offline): from agents import Agent, ModelRetrySettings, ModelSettings, OpenAIChatCompletionsModel, Runner, function_tool, set_tracing_disabled from openai import AsyncOpenAI import httpx2 set_tracing_disabled(True) observations = {"requests": [], "tools": []} @function_tool def lookup_stock(sku: str) -> str: """Read stock from a fixed demo catalog. Accept only QR-SKU-001.""" if sku != "QR-SKU-001": raise ValueError("Unknown demo SKU") result = {"sku": sku, "stock": 7, "source": "DEMO-CATALOG-20261008"} observations["tools"].append({"name": "lookup_stock", "arguments": {"sku": sku}, "output": result}) return json.dumps(result) def capture(request): body = json.loads(request.content) observations["requests"].append({"path": request.url.path, "body": body}) if request.url.path != "/v1/chat/completions": raise AssertionError("Unexpected protocol") index = len(observations["requests"]) if index == 1: if body["tools"][0]["function"]["name"] != "lookup_stock": raise AssertionError("Missing tool schema") message = {"role": "assistant", "content": None, "tool_calls": [{"id": "call_demo_stock", "type": "function", "function": {"name": "lookup_stock", "arguments": '{"sku":"QR-SKU-001"}'}}]} finish = "tool_calls" elif index == 2: assistant = next(item for item in body["messages"] if item.get("tool_calls")) tool = next(item for item in body["messages"] if item["role"] == "tool") if tool["tool_call_id"] != assistant["tool_calls"][0]["id"] or json.loads(tool["content"])["stock"] != 7: raise AssertionError("Tool history mismatch") message = {"role": "assistant", "content": "QR-SKU-001 库存 7 件，来源 DEMO-CATALOG-20261008。"} finish = "stop" else: raise AssertionError("Unexpected third model request") return httpx2.Response(200, json={"id": f"chatcmpl-fixture-{index}", "object": "chat.completion", "created": 1, "model": model_id, "choices": [{"index": 0, "message": message, "finish_reason": finish}], "usage": {"prompt_tokens": 100, "completion_tokens": 20, "total_tokens": 120}}) client_options = {"api_key": "offline-fixture" if offline else os.environ["QUICKROUTER_API_KEY"], "base_url": "https://api.quickrouter.ai/v1", "max_retries": 0, "timeout": 45.0} if offline: client_options["http_client"] = httpx2.AsyncClient(transport=httpx2.MockTransport(capture)) client = AsyncOpenAI(**client_options) agent = Agent(name="DemoStockReader", instructions="先调用 lookup_stock 查询 QR-SKU-001，再用工具实际结果回答库存和来源。不要猜测数据，不要调用第二次工具。", model=OpenAIChatCompletionsModel(model=model_id, openai_client=client), tools=[lookup_stock], model_settings=ModelSettings(max_tokens=160, parallel_tool_calls=False, retry=ModelRetrySettings(max_retries=0))) started = time.monotonic() try: result = await Runner.run(agent, "QR-SKU-001 还有多少库存？请给来源。", max_turns=3) history = result.to_input_list() usage = result.context_wrapper.usage final = str(result.final_output) return {"status": "passed" if len(observations["tools"]) == 1 and "7" in final and "DEMO-CATALOG-20261008" in final else "needs-review", "mode": "offline" if offline else "live", "sdk": importlib.metadata.version("openai-agents"), "openai": importlib.metadata.version("openai"), "model": model_id, "protocol": "/v1/chat/completions", "elapsedSeconds": round(time.monotonic() - started, 3), "toolExecutions": observations["tools"], "history": history, "finalOutput": final, "usage": {"requests": usage.requests, "inputTokens": usage.input_tokens, "outputTokens": usage.output_tokens, "totalTokens": usage.total_tokens}, "billedUsd": None, "offlineRequests": observations["requests"] if offline else [], "boundary": "Fixed read-only catalog; no external write. Usage is not actual billing. Other models and hosted OpenAI tools unverified."} finally: await client.close() def main(): parser = argparse.ArgumentParser() parser.add_argument("--model", default=os.environ.get("QUICKROUTER_MODEL", "glm-5.3")) parser.add_argument("--run", action="store_true") parser.add_argument("--offline", action="store_true") parser.add_argument("--output", default="agents-tool-loop-record.json") args = parser.parse_args() if args.run and args.offline: parser.error("Choose --run or --offline") if not args.run and not args.offline: print(json.dumps({"mode": "preview", "model": args.model, "protocol": "/v1/chat/completions", "maxTurns": 3, "maxRetries": 0, "tracing": False, "task": "Read QR-SKU-001 stock via one local tool; normally two model requests", "run": "--run --output UNIQUE_RECORD.json", "offline": "--offline --output UNIQUE_FIXTURE.json"}, ensure_ascii=False, indent=2)) return if args.run and not os.environ.get("QUICKROUTER_API_KEY"): parser.error("Set QUICKROUTER_API_KEY privately") if not args.model or len(args.model) > 160 or any(ord(char) < 32 for char in args.model): parser.error("Invalid model ID") path = Path(args.output) path.parent.mkdir(parents=True, exist_ok=True) try: with path.open("x") as handle: json.dump({"status": "submitting", "model": args.model, "mode": "offline" if args.offline else "live", "billedUsd": None}, handle) except FileExistsError: parser.error("Record exists; inspect it rather than resubmit") try: record = asyncio.run(execute(args.model, args.offline)) except Exception as error: record = {"status": "unknown", "errorType": type(error).__name__, "model": args.model, "billedUsd": None, "boundary": "No retry; inspect account logs before another live run. Error text and headers omitted."} save(path, record) print(json.dumps({"status": "unknown", "errorType": type(error).__name__, "record": str(path)})) raise SystemExit(1) save(path, record) print(json.dumps({"status": record["status"], "record": str(path), "modelRequests": record["usage"]["requests"], "toolExecutions": len(record["toolExecutions"])}, ensure_ascii=False)) if record["status"] != "passed": raise SystemExit(2) if __name__ == "__main__": main()