第 11 章：生产可观测

出处

来源：usage_cost_api.ipynb · 状态：已读

一句话

可观测是控制面的一部分：看见 token 和美元、按工作空间分账、在 session 上封顶、把推理钉在合规地域。 仪表盘如果不能触发 archive、降级或拒绝下一轮，就只是装饰。

核心概念

Admin Usage / Cost：历史分析，不是实时监控

observability/usage_cost_api.ipynb 两套端点：

	Usage	Cost
内容	时间桶里的 token：uncached input、output、cache 写/读、server tool	美元，按服务类型拆
粒度	1m（最多 1440 桶）/ 1h（168）/ 1d（31）	只有 1d（31）
钥匙	Admin key，sk-ant-admin...，环境变量，定期轮换	同左

常见用途：用量监控、按 workspace 做 chargeback、看 cache 命中、给财务出摘要。分组、过滤、分页、CSV 导出是基本功。

限制要当制度记：

• Priority Tier 费用不出现在 Cost 端点（计费模型不同）。Usage 里能看见用量，钱对不上时先查这个。
• API 为历史分析设计，不是实时监控。实时仍靠你自己的日志 / OTel（第 8 章 SDK 那条）。
• 字段名会变；生产代码对未知值要宽容。
• 小日期范围先试，再谈保留策略。

Session 预算：实验室 Pareto 的运行时闸门

CMA_cap_session_spend.ipynb（和第 9 章交叉）：第 10 章在离线 eval 上找 cost/task；生产里一次 runaway 咨询或 PTC 循环会在评测集看不见的地方烧完预算。session spend cap 是硬停，不是报表。顾问咨询没有单次上限，cookbook 明确说靠这个 cap 兜底。

和 webhook 一起：idle / budget 事件把「该停了」推给你的服务，而不是你轮询到账单惊吓。

地理钉扎和资源生命周期

CMA_pin_inference_geo.ipynb：合规要求推理发生在指定区域时，这是产品配置，不是运维口头保证。CMA_operate_in_production.ipynb 的 archive / delete 是可观测的另一半——workspace 里堆积的 agent、env、session、vault 若不纳入生命周期，用量 API 会把尸体也算进趋势。

对照 recipe（默认折叠；机制片段摘自官方 notebook，全文在 GitHub）：

observability/usage_cost_api.ipynb — 组织级用量和成本；cache 效率；分账

GitHub 全文

def get_daily_costs(client, days_back=7):
    """Get cost data for the last N days."""
    end_time = datetime.combine(datetime.utcnow(), time.min)
    start_time = end_time - timedelta(days=days_back)

    params = {
        "starting_at": start_time.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "ending_at": end_time.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "bucket_width": "1d",  # Only 1d supported for cost API
        "limit": min(days_back, 31),  # Max 31 days per request
    }

    return client._make_request("cost_report", params)


def analyze_cost_data(response):
    """Process and display cost data."""
    if not response or not response.get("data"):
        print("No cost data found.")
        return

    total_cost_minor_units = 0
    daily_costs = []

    for bucket in response["data"]:
        date = bucket["starting_at"][:10]

        # Sum all costs in this bucket
        bucket_cost = 0
        for result in bucket["results"]:
            # Convert string amounts to float if needed
            amount = result.get("amount", 0)
            if isinstance(amount, str):
                try:
                    amount = float(amount)
                except (ValueError, TypeError):
                    amount = 0
            bucket_cost += amount

        daily_costs.append(
            {
                "date": date,
                "cost_minor_units": bucket_cost,
                "cost_usd": bucket_cost / 100,  # Convert to dollars
            }
        )

        total_cost_minor_units += bucket_cost

    total_cost_usd = total_cost_minor_units / 100

    print("💰 Cost Summary:")
    print(f"Total cost: ${total_cost_usd:.4f}")
    print(f"Average daily cost: ${total_cost_usd / len(daily_costs):.4f}")

    return daily_costs


# Example usage
if client:
    cost_response = get_daily_costs(client, days_back=7)
    daily_costs = analyze_cost_data(cost_response)

managed_agents/CMA_cap_session_spend.ipynb — 单 session 花费封顶

GitHub 全文

env = client.beta.environments.create(
    name="budget-demo",
    config={"type": "anthropic_cloud", "networking": {"type": "unrestricted"}},
    betas=BETAS,
)

analyst = client.beta.agents.create(
    name="market_analyst",
    description="Writes sourced competitive landscape briefs.",
    model={"id": MODEL},
    system="""You write competitive landscape briefs for product teams.
Given a market, use web search and web fetch to find the notable vendors, their
positioning, and recent moves. Read primary sources rather than aggregators.
Write the brief to /mnt/session/outputs/brief.md with a Sources section listing every
URL you relied on. Keep researching until you are confident the brief is complete.""",
    tools=[
        {
            "type": "agent_toolset_20260401",
            "configs": [{"name": "web_search"}, {"name": "web_fetch"}],
        }
    ],
    betas=BETAS,
)
print(f"{analyst.name}: {analyst.id} v{analyst.version}")

from decimal import Decimal


def usd(money) -> str:
    """Render an integer minor-unit amount ("50" = fifty cents) as dollars."""
    return f"${Decimal(money.amount) / 100:.2f}"


session = client.beta.sessions.create(
    agent=analyst.id,
    environment_id=env.id,
    title="Landscape brief: observability platforms",
    budget={
        "type": "limit",
        "max_list_cost": {"currency": "USD", "amount": "10"},
    },
    initial_events=[
        {
            "type": "user.message",
            "content": [
                {
                    "type": "text",
                    "text": "Write a competitive landscape brief on the observability "
                    "platform market: the main vendors, how each positions itself, "
                    "and any notable moves in the last year.",
                }
            ],
        }
    ],
    betas=BETAS,
)
print(session.id, session.status)
print("budget:", usd(session.budget.max_list_cost))

managed_agents/CMA_pin_inference_geo.ipynb — 数据驻留 / 推理地理

GitHub 全文

researcher = client.beta.agents.create(
    name="us_records_analyst",
    description="Answers questions about internal records, pinned to US inference.",
    model={"id": MODEL, "inference_geo": "us"},
    system="You answer questions about the records you are given, concisely.",
    tools=[{"type": "agent_toolset_20260401"}],
    betas=BETAS,
)
print(f"{researcher.name}: {researcher.id} v{researcher.version}")
print("inference_geo:", researcher.model.inference_geo)

env = client.beta.environments.create(
    name="residency-demo",
    config={"type": "anthropic_cloud", "networking": {"type": "unrestricted"}},
    betas=BETAS,
)

session = client.beta.sessions.create(
    agent=researcher.id,
    environment_id=env.id,
    title="Records question, US-pinned",
    initial_events=[
        {
            "type": "user.message",
            "content": [
                {"type": "text", "text": "In one sentence: what is a data residency policy?"}
            ],
        }
    ],
    betas=BETAS,
)
print("session pin:", session.agent.model.inference_geo)

with client.beta.sessions.events.stream(session.id, betas=BETAS) as stream:
    for ev in stream:
        if ev.type == "agent.message":
            print("".join(b.text for b in ev.content if b.type == "text"), end="")
        elif ev.type == "session.status_idle":
            print(f"\n[idle] stop_reason={ev.stop_reason.type}")
            break

override = client.beta.sessions.create(
    agent={
        "type": "agent_with_overrides",
        "id": researcher.id,
        "model": {"id": MODEL, "inference_geo": "global"},  # this session only
    },
    environment_id=env.id,
    title="One-off, global geography",
    betas=BETAS,
)
print("override session pin:", override.agent.model.inference_geo)
print(
    "agent still pinned to:",
    client.beta.agents.retrieve(researcher.id, betas=BETAS).model.inference_geo,
)

managed_agents/CMA_operate_in_production.ipynb — idle / budget webhook 与资源生命周期

GitHub 全文

credential = client.beta.vaults.credentials.create(
    vault_id=vault.id,
    display_name="GitHub Copilot",
    auth={
        "type": "static_bearer",
        "mcp_server_url": "https://api.githubcopilot.com/mcp/",
        "token": GH_TOKEN,
    },
)
print(f"credential: {credential.id}")

agent = client.beta.agents.create(
    name="cookbook-operate",
    model=MODEL,
    system="You navigate GitHub repositories on behalf of the logged-in user.",
    mcp_servers=[
        {
            "type": "url",
            "name": "github",
            "url": "https://api.githubcopilot.com/mcp/",
        }
    ],
    tools=[
        {
            "type": "mcp_toolset",
            "mcp_server_name": "github",
            "default_config": {
                "enabled": True,
                "permission_policy": {"type": "always_allow"},
            },
        }
    ],
)

env = client.beta.environments.create(
    name="cookbook-operate-env",
    config={"type": "cloud", "networking": {"type": "unrestricted"}},
)

session = client.beta.sessions.create(
    environment_id=env.id,
    agent={"type": "agent", "id": agent.id, "version": agent.version},
    vault_ids=[vault.id],
    title="Operate demo",
)
print(f"session: {session.id}")

和 04-上下文工程、10-评测与成本 一起看：cache 和 compact 是在减账单；evals 是在证明减得有道理；这一章是在运行时看见账单，并在超限时停。

关键洞察

三层数字不要混：

1. 评测集上的 $/task（第 10 章）——改架构的依据
2. Admin API 的组织趋势——财务和容量
3. Session cap / webhook——单次运行的熔断

只做 (2) 的团队会在月报里发现异常，但拦不住已经跑完的 session。只做 (3) 没有 (1)，熔断阈值是拍的。Priority Tier 让 (2) 的 Cost 端点故意缺一块：分账逻辑必须同时读 Usage。

个人思考

Book 1 08-团队落地 讲的是人怎么用 harness；本章讲的是钱和合规怎么约束 harness。第 12 章第三方会再引入别家的账单（Pinecone、ElevenLabs）。可观测的边界是：凡是 token 离开 Anthropic 的，都要在你自己的账本上有一行。

相关

• 00-Cookbooks阅读导读
• 09-Managed-Agents
• 10-评测与成本
• 08-团队落地
• 04-上下文工程