《AI Agent 工程实战》18.3 上线、观察与迭代 + 真实 API 端到端冒烟（可选）

"""18.3 真实 API 冒烟（可选；需自备密钥、会消耗额度，本机未执行）。"""
import os
from anthropic import Anthropic
from smoke3 import (  # 复用离线脚本的语料、工具与校验
    CORPUS, QUESTION, SYSTEM, call_tool, run_smoke, schemas, validate_citations,
)


def to_anthropic(messages):
    out = []
    for m in messages:
        if m.role == "system":
            continue
        if m.role == "tool":
            out.append({"role": "user", "content": [
                {"type": "tool_result", "tool_use_id": m.tool_call_id, "content": m.content}]})
        elif m.role == "assistant" and m.tool_calls:
            blocks = ([{"type": "text", "text": m.content}] if m.content else []) + [
                {"type": "tool_use", "id": tc.id, "name": tc.name, "input": tc.arguments}
                for tc in m.tool_calls]
            out.append({"role": "assistant", "content": blocks})
        else:
            out.append({"role": m.role, "content": m.content})
    return out


class AnthropicModel:
    """真实模型适配器：把 ModelResponse 契约翻译成 Anthropic messages API。"""

    def __init__(self, model: str) -> None:
        self.model = model
        self._client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])  # 只从环境读

    def complete(self, messages, tools) -> "ModelResponse":
        system = next((m.content for m in messages if m.role == "system"), "")
        resp = self._client.messages.create(
            model=self.model, max_tokens=1024, system=system,
            messages=to_anthropic(messages),
            tools=[{"name": t["name"], "description": t["description"],
                    "input_schema": t["parameters"]} for t in tools])
        calls = [ToolCall(b.id, b.name, b.input) for b in resp.content if b.type == "tool_use"]
        text = "".join(b.text for b in resp.content if b.type == "text")
        return ModelResponse(
            message=Message(role="assistant", content=text, tool_calls=calls),
            stop_reason="tool_use" if calls else (resp.stop_reason or "end_turn"),
            usage={"input_tokens": resp.usage.input_tokens,
                   "output_tokens": resp.usage.output_tokens})


def main() -> None:
    assert os.environ.get("ANTHROPIC_API_KEY"), "请先 export ANTHROPIC_API_KEY"
    model = AnthropicModel(model=os.environ.get("RC_MODEL", "claude-sonnet-4-5"))
    final, steps, uin, uout = run_smoke(model)
    print(final)
    for cite, ok in validate_citations(final):
        print(f"  {cite:<20} {'OK' if ok else 'MISSING'}")
    print(f"input={uin} output={uout} cost=${uin/1e6*3 + uout/1e6*15:.6f}")


if __name__ == "__main__":
    main()

"""18.3 端到端冒烟：真/假模型共用同一条三跳问答通路（离线默认）。
   类型定义（Message / ToolCall / ModelResponse）沿用 18.2.1，此处省略。"""
import json
import re
import time

CORPUS: dict[str, list[str]] = {
    "gil-and-cpu": ["CPU 密集型任务受 GIL 限制，多线程无法并行，应改用多进程。",
                    "I/O 密集型任务在等待期间会释放 GIL，因此多线程仍有收益。"],
    "asyncio-basics": ["asyncio 适合 I/O 密集型任务，事件循环在单线程内调度协程。"],
    "memory-model": ["Python 的引用计数在对象引用归零时立即回收。"],
}


def search(query: str, k: int = 2) -> str:
    terms = [t.lower() for t in query.split() if t]
    scored = sorted(
        ((sum(txt.lower().count(t) for t in terms), d, c, txt)
         for d, chunks in CORPUS.items() for c, txt in enumerate(chunks)),
        key=lambda x: -x[0])
    hits = [{"doc_id": d, "chunk_id": c, "score": s, "snippet": txt}
            for s, d, c, txt in scored[:k] if s]
    return json.dumps(hits, ensure_ascii=False)


def read(doc_id: str, chunk_id: int) -> str:
    return CORPUS[doc_id][chunk_id]


TOOLS: dict[str, tuple[dict, object]] = {
    "search": ({"type": "object", "properties": {"query": {"type": "string"},
                "k": {"type": "integer"}}, "required": ["query"]}, search),
    "read": ({"type": "object", "properties": {"doc_id": {"type": "string"},
              "chunk_id": {"type": "integer"}}, "required": ["doc_id", "chunk_id"]}, read),
}


def schemas() -> list[dict]:
    return [{"name": n, "parameters": p} for n, (p, _) in TOOLS.items()]


def call_tool(tc) -> str:
    return TOOLS[tc.name][1](**tc.arguments)


class FakeModel:
    """脚本驱动：每步要么给出一串 ToolCall，要么给出终结文本。"""

    def __init__(self, script: list, name: str = "fake") -> None:
        self._script, self._i, self.name = list(script), 0, name

    def complete(self, messages, tools) -> "ModelResponse":
        step = self._script[self._i]
        self._i += 1
        if isinstance(step, str):
            return ModelResponse(message=Message(role="assistant", content=step))
        if isinstance(step, ModelResponse):
            return step
        return ModelResponse(message=Message(role="assistant", tool_calls=list(step)),
                             stop_reason="tool_use")


PRICES = {"input": 3.0, "output": 15.0}
SYSTEM = ("你是严谨的研究助理。回答每个论断时在句末标注来源 [doc_id#chunk_id]，"
          "只能引用 search/read 返回过的块。")
QUESTION = "Python 里 CPU 密集和 I/O 密集的并发方案该怎么选？"
FINAL = ("CPU 密集型受 GIL 限制，应改用多进程。[gil-and-cpu#0]\n"
         "I/O 密集型适合 asyncio，事件循环单线程调度协程。[asyncio-basics#0]")


def turn(calls, text="", inp=0, out=0) -> "ModelResponse":
    return ModelResponse(
        message=Message(role="assistant", content=text, tool_calls=list(calls)),
        stop_reason="tool_use" if calls else "end_turn",
        usage={"input_tokens": inp, "output_tokens": out})


def fake_script() -> list:
    return [
        turn([ToolCall("t1", "search", {"query": "GIL CPU 密集 多线程", "k": 2})], inp=120, out=18),
        turn([ToolCall("t2", "read", {"doc_id": "gil-and-cpu", "chunk_id": 0})], inp=210, out=12),
        turn([ToolCall("t3", "search", {"query": "asyncio 事件循环", "k": 2})], inp=320, out=16),
        turn([ToolCall("t4", "read", {"doc_id": "asyncio-basics", "chunk_id": 0})], inp=420, out=10),
        turn([], text=FINAL, inp=560, out=64),
    ]


def run_smoke(model, max_steps: int = 8):
    messages = [Message(role="system", content=SYSTEM),
                Message(role="user", content=QUESTION)]
    steps, usage_in, usage_out = [], 0, 0
    for i in range(max_steps):
        resp = model.complete(messages, schemas())
        usage_in += resp.usage.get("input_tokens", 0)
        usage_out += resp.usage.get("output_tokens", 0)
        messages.append(resp.message)
        if not resp.message.tool_calls:
            steps.append(f"[{i}] final  {resp.message.content.splitlines()[0]} ...")
            return resp.message.content, steps, usage_in, usage_out
        for tc in resp.message.tool_calls:
            messages.append(Message(role="tool", content=call_tool(tc), tool_call_id=tc.id))
            steps.append(f"[{i}] tool   {tc.name}({tc.arguments})")
    return "", steps, usage_in, usage_out


def validate_citations(final: str) -> list[tuple[str, bool]]:
    return [(f"{d}#{c}", d in CORPUS and int(c) < len(CORPUS[d]))
            for d, c in re.findall(r"\[([\w-]+)#(\d+)\]", final)]


def main() -> None:
    model = FakeModel(fake_script(), name="fake-offline")
    t0 = time.perf_counter()
    final, steps, uin, uout = run_smoke(model)
    cost = uin / 1e6 * PRICES["input"] + uout / 1e6 * PRICES["output"]
    print(f"模型: {model.name}（离线，未触网）")
    print("== 步骤轨迹 ==")
    for s in steps:
        print(" ", s)
    print("== 最终回答 ==")
    print(final)
    print("== 引用可回溯校验 ==")
    rows = validate_citations(final)
    for cite, ok in rows:
        print(f"  {cite:<20} {'OK' if ok else 'MISSING'}")
    print(f"  引用 {len(rows)} 条，全部可回溯 = {all(ok for _, ok in rows)}")
    print("== 用量与成本 ==")
    print(f"  input={uin}  output={uout}  cost=${cost:.6f}  耗时={(time.perf_counter()-t0)*1000:.1f}ms")


if __name__ == "__main__":
    main()

模型: fake-offline（离线，未触网）
== 步骤轨迹 ==
  [0] tool   search({'query': 'GIL CPU 密集 多线程', 'k': 2})
  [1] tool   read({'doc_id': 'gil-and-cpu', 'chunk_id': 0})
  [2] tool   search({'query': 'asyncio 事件循环', 'k': 2})
  [3] tool   read({'doc_id': 'asyncio-basics', 'chunk_id': 0})
  [4] final  CPU 密集型受 GIL 限制，应改用多进程。[gil-and-cpu#0] ...
== 最终回答 ==
CPU 密集型受 GIL 限制，应改用多进程。[gil-and-cpu#0]
I/O 密集型适合 asyncio，事件循环单线程调度协程。[asyncio-basics#0]
== 引用可回溯校验 ==
  gil-and-cpu#0        OK
  asyncio-basics#0     OK
  引用 2 条，全部可回溯 = True
== 用量与成本 ==
  input=1630  output=120  cost=$0.006690  耗时=0.2ms