《AI Agent 开发入门》18.2 真实 API 冒烟（可选）

import os
from minagent import run
from minagent.anthropic_adapter import AnthropicModel

MODEL_ID = "claude-opus-5-5"            # 占位：实际 id 以 models.list() 查询为准
model = AnthropicModel(model=MODEL_ID)  # 密钥从 os.environ["ANTHROPIC_API_KEY"] 读

result = run(model, build_registry(), QUESTION, system=SYSTEM)
print(result.final)

import json
from minagent.types import Message, ToolCall
from minagent.anthropic_adapter import to_anthropic

msgs = [
    Message(role="system", content="你是严谨的助手。"),
    Message(role="user", content="asyncio 适合什么场景？"),
    Message(role="assistant", content="我先检索一下资料。",
            tool_calls=[ToolCall("toolu_1", "search", {"query": "asyncio I/O 密集", "k": 2})]),
    Message(role="tool", content='[{"doc_id": "python-async-advanced"}]', tool_call_id="toolu_1"),
]
system, out = to_anthropic(msgs)
print("system =", repr(system))
print("messages =", json.dumps(out, ensure_ascii=False))

system = '你是严谨的助手。'
messages = [{"role": "user", "content": "asyncio 适合什么场景？"}, {"role": "assistant", "content": [{"type": "text", "text": "我先检索一下资料。"}, {"type": "tool_use", "id": "toolu_1", "name": "search", "input": {"query": "asyncio I/O 密集", "k": 2}}]}, {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "toolu_1", "content": "[{\"doc_id\": \"python-async-advanced\"}]"}]}]

# record_smoke.py（一次性，需网络与密钥；本节不运行它）
import json
from pathlib import Path
from minagent.anthropic_adapter import AnthropicModel


class RecordingClient:
    """包一层 Anthropic client，把每次响应原样存成夹具帧。"""

    def __init__(self, client, path: str) -> None:
        self._client = client
        self._path = Path(path)
        self._frames: list[dict] = []

    @property
    def messages(self):
        return self

    def create(self, **kwargs):
        resp = self._client.messages.create(**kwargs)
        self._frames.append(resp.model_dump(mode="json"))
        self._path.write_text(json.dumps(
            {"model": kwargs["model"], "recorded_at": "2026-10-10T10:00:00+08:00",
             "frames": self._frames}, ensure_ascii=False, indent=2), encoding="utf-8")
        return resp


# 只在明确同意、密钥已设时执行：
# from anthropic import Anthropic
# model = AnthropicModel(model=MODEL_ID, client=RecordingClient(Anthropic(), "fixtures/smoke.json"))
# run(model, build_registry(), QUESTION, system=SYSTEM)

{
  "model": "claude-opus-5-5",
  "recorded_at": "2026-10-10T10:00:00+08:00",
  "frames": [
    {
      "id": "msg_01AAA", "type": "message", "role": "assistant",
      "model": "claude-opus-5-5",
      "content": [
        {"type": "text", "text": "我先检索一下资料。"},
        {"type": "tool_use", "id": "toolu_1", "name": "search",
         "input": {"query": "asyncio I/O 密集", "k": 2}}
      ],
      "stop_reason": "tool_use", "stop_sequence": null,
      "usage": {"input_tokens": 88, "output_tokens": 20}
    },
    {
      "id": "msg_01BBB", "type": "message", "role": "assistant",
      "model": "claude-opus-5-5",
      "content": [
        {"type": "text",
         "text": "asyncio 适合 I/O 密集型任务，事件循环在单线程内调度协程。[python-async-advanced#0]"}
      ],
      "stop_reason": "end_turn", "stop_sequence": null,
      "usage": {"input_tokens": 160, "output_tokens": 24}
    }
  ]
}

# smoke_offline.py
from minagent import Tool, ToolRegistry, read, run, search_tool
from minagent.recorded import RecordedModel

SYSTEM = "你是严谨的助手。回答每个论断时在句末标注来源 [doc_id#chunk_id]。"
QUESTION = "asyncio 适合什么场景？"

model = RecordedModel("fixtures/smoke.json")
print(f"模型: {model.model}  录制于 {model.recorded_at}")
result = run(model, build_registry(), QUESTION, system=SYSTEM)
for s in result.steps:
    print(f"[{s.index}] {s.kind:<9} {s.detail}")
print("最终回答:", result.final)

模型: claude-opus-5-5  录制于 2026-10-10T10:00:00+08:00
[0] tool_call search({'query': 'asyncio I/O 密集', 'k': 2}) -> [{"doc_id": "python-async-advanced", "chunk_id": 0, "score": 3.508, "snippet": "asyncio 适合 I/O 密集型任务，事件循环在单线程内调度大量协程。"}, {"doc_id": "python-concurrency-performance", "chunk_id": 2, "score": 1.693, "snippet": "I/O 密集型任务在等待期间会释放 GIL，因此多线程仍有收益。"}]
[1] final     asyncio 适合 I/O 密集型任务，事件循环在单线程内调度协程。[python-async-advanced#0]
最终回答: asyncio 适合 I/O 密集型任务，事件循环在单线程内调度协程。[python-async-advanced#0]