打造 AI 代理迴圈：工具、記憶與安全迭代

# 帶有硬性迭代上限的代理迴圈。模型 -> tool_calls -> 執行 ->
# 把結果回饋進去 -> 重複，直到模型以文字作答，或我們撞到
# 上限。這個上限，就是「代理」與「失控帳單」之間的差別。
from openai import OpenAI
import json

client = OpenAI(
    api_key="sk-brievio-...",
    base_url="https://api.brievio.com/v1",
)

MAX_ITERS = 8   # 多數任務 2～4 輪就能完成；8 是相當寬裕的緩衝。

def run_agent(question: str, model: str = "claude-sonnet-4-6") -> str:
    messages = [
        {"role": "system", "content": "You are a helpful research agent. "
         "Use the tools when you need live data. Answer directly when you "
         "already know enough."},
        {"role": "user", "content": question},
    ]

    for step in range(MAX_ITERS):
        resp = client.chat.completions.create(
            model=model,
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
        )
        msg = resp.choices[0].message

        # 沒有請求任何工具 -> 這就是最終答案。完成。
        if not msg.tool_calls:
            return msg.content

        # 把 assistant 這一輪「原封不動」地附加進去 — 它帶著
        # 後續訊息必須引用的 tool_call id。
        messages.append(msg)

        # 執行每一個被請求的呼叫，並為每個 id 附加一則 tool 訊息。
        for call in msg.tool_calls:
            result = dispatch(call)              # 見下一段程式碼
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,          # 必須對應該呼叫的 id
                "content": json.dumps(result),
            })
        # 迴圈:模型現在看到工具輸出，並繼續往下走。

    # 撞到上限卻仍未解決。大聲失敗 — 別默默地永遠迴圈下去。
    raise RuntimeError(f"agent did not finish within {MAX_ITERS} iterations")

# 帶驗證的工具派發。模型負責「提議」一個呼叫；你的程式碼
# 負責「裁決」它。每一個引數都是不可信的模型輸出 — 先解析它、
# 確認名稱是你註冊過的，並在執行前驗證型別。
def get_weather(city: str, unit: str = "celsius") -> dict:
    if not isinstance(city, str) or not city.strip():
        raise ValueError("city must be a non-empty string")
    if unit not in ("celsius", "fahrenheit"):
        raise ValueError(f"unsupported unit: {unit!r}")
    return {"city": city, "temp": 18, "unit": unit, "sky": "clear"}

# 白名單:模型只能呼叫你明確註冊過的東西。
TOOL_IMPLS = {"get_weather": get_weather}

TOOLS = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get current weather for a city.",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string"},
                "unit": {"type": "string",
                         "enum": ["celsius", "fahrenheit"]},
            },
            "required": ["city"],
        },
    },
}]

def dispatch(call) -> dict:
    name = call.function.name
    fn = TOOL_IMPLS.get(name)
    if fn is None:
        # 模型幻覺出一個工具。別讓迴圈崩潰 — 把這個錯誤
        # 當成 tool 結果回傳，讓模型可以自我修正。
        return {"error": f"unknown tool: {name}"}

    try:
        args = json.loads(call.function.arguments)   # 永遠是一個 JSON 字串
    except json.JSONDecodeError:
        return {"error": "arguments were not valid JSON"}

    try:
        return fn(**args)
    except (TypeError, ValueError) as e:
        # 引數有問題(型別錯、缺欄位、超出範圍)。把訊息
        # 回饋回去;模型通常會用修正過的呼叫重試。
        return {"error": str(e)}

# 一個「每次執行」的成本／權杖預算護欄。每一輪迴圈都會重送一份
# 「不斷成長」的脈絡(歷史 + 工具輸出),所以成本每輪攀升。每次
# 呼叫後讀取那個誠實的 usage 物件,替它定價,並在本次執行
# 超過預算時停下 — 與迭代上限各自獨立。
from decimal import Decimal

# Brievio 公布的費率，每 100 萬權杖的美元數(約比官方參考價低 15%)。
RATES = {
    "claude-sonnet-4-6": {"in": Decimal("2.55"), "out": Decimal("12.75")},
    "claude-haiku-4-5":  {"in": Decimal("0.85"), "out": Decimal("4.25")},
}

def call_cost(model: str, usage) -> Decimal:
    r = RATES[model]
    m = Decimal("1000000")
    return usage.prompt_tokens * r["in"] / m + usage.completion_tokens * r["out"] / m

RUN_BUDGET = Decimal("0.10")   # 每次代理執行 10 美分，硬性上限。

def run_agent_budgeted(question: str, model: str = "claude-sonnet-4-6") -> str:
    messages = [{"role": "user", "content": question}]
    spent = Decimal("0")

    for step in range(MAX_ITERS):
        resp = client.chat.completions.create(
            model=model, messages=messages, tools=TOOLS, tool_choice="auto",
        )
        spent += call_cost(model, resp.usage)   # 每一輪都累計「真實」權杖
        if spent > RUN_BUDGET:
            raise RuntimeError(f"run exceeded ${RUN_BUDGET} (spent ${spent:.4f})")

        msg = resp.choices[0].message
        if not msg.tool_calls:
            return msg.content

        messages.append(msg)
        for call in msg.tool_calls:
            messages.append({"role": "tool", "tool_call_id": call.id,
                             "content": json.dumps(dispatch(call))})

    raise RuntimeError(f"agent did not finish within {MAX_ITERS} iterations")