Advisor：让工作中的智能体在回合中咨询更强模型

本文介绍 Anthropic 的 Advisor 机制，允许工作模型在会话中按需咨询更强大的模型，以平衡成本与决策质量。通过 multiagent 配置声明顾问，平台自动暴露 advisor 工具，咨询以独立线程运行并单独计费。文章涵盖配置方法、事件流观察、成本核算、内容编辑处理及与专家团队的集成，并提供完整代码示例。适合有编程基础、希望掌握 AI 工程实践的开发者。

技术背景

在构建多智能体系统时，开发者常面临模型选择困境：使用最强模型虽安全但成本高昂，使用中档模型则可能在复杂决策时缺乏判断力。传统解决方案是自建升级机制，但实现复杂且难以维护。Advisor 机制将这一过程内置于会话中，允许工作模型在需要时动态咨询更强模型，既保证决策质量又控制成本。适用于需要平衡性能与成本的场景，如 API 设计、代码审查、复杂推理等。主要限制包括：每个名册仅一个顾问、模型配对需平台允许、咨询内容可能被策略编辑。

核心原理

Advisor 的核心原理是在会话中创建一个特殊的顾问线程。工作模型通过 advisor 工具触发咨询，平台将当前对话上下文传递给顾问模型，顾问模型返回指导，工作模型继续采样。咨询作为短期线程运行，独立计费，其生命周期事件（thread_created、thread_status_running、thread_message_received 等）在主数据流中可见。顾问模型不接收工作模型编写的查询，而是读取整个对话，因此系统提示用于控制咨询策略。每次咨询都是新线程，传递建议后自行终止。

架构分析

系统架构采用协调器模式，multiagent 配置中声明顾问条目，平台将其作为 advisor 工具暴露给工作模型。顾问线程与主线程并行运行，但不受子线程并发限制。事件流包含线程生命周期事件和消息事件，用于监控咨询过程。成本核算通过线程 usage 字段实现，按顾问模型公开价格计费。内容编辑由顾问模型策略控制，工作模型始终能读取完整建议。

完整实战教程

在每个回合都使用最强大的模型是安全但昂贵的选择。使用中档模型虽然便宜，但智能体可能在任务中遇到需要更多判断力的决策时，已经做出了承诺。通常的解决方法是构建自己的升级机制：检测困难情况，打包上下文，调用更大的模型，并将答案拼接回去。

Advisor在会话内部完成这一过程。你在智能体的multiagent配置中指定一个更强大的模型作为{"type": "advisor", "model": ...}条目。主线程随后拥有一个advisor工具，可以在回合中调用：平台将当前对话上下文展示给顾问模型，将其指导返回给工作模型，然后继续采样。每次咨询作为短期线程运行，独立计费，因此你可以看到咨询发生的时间并精确计算其成本。

本笔记本将介绍：

向智能体添加advisor条目
告诉工作模型何时咨询
从事件流读取咨询结果及其线程成本
当顾问模型的输出被编辑时会发生什么变化

1. 设置客户端

顾问条目位于 multiagent 块中，位于标准的 managed-agents-2026-04-01 beta 标头下。类型化的名册形状需要 anthropic>=0.121.0。

%%capture
%pip install -qU "anthropic>=0.121.0" python-dotenv

import os

import anthropic
from dotenv import load_dotenv

load_dotenv()

BETAS = ["managed-agents-2026-04-01"]
WORKER_MODEL = os.environ.get("COOKBOOK_MODEL", "claude-sonnet-5")
ADVISOR_MODEL = os.environ.get("COOKBOOK_ADVISOR_MODEL", "claude-opus-5")
client = anthropic.Anthropic()

2. 给 Sonnet 智能体配备 Opus 顾问

你在名册中声明顾问，而不是作为工具定义，也不是单独创建的智能体。该条目与任何专家一起放在 multiagent.agents 中，仅包含顾问条目的名册也是有效的：智能体会咨询但不会生成子智能体。服务器在保存时强制执行以下几条规则：

每个名册最多一个顾问条目。
model 必须被允许作为顾问，并且智能体自身模型与它的配对必须是平台允许的。不合适的配对会在 agents.create 时返回 400，而不是运行时意外。
该条目在名册中占用名称 anthropic.advisor，因此任何专家都不能使用该名称。

在运行时，平台将该条目作为 advisor 工具暴露给工作模型。该工具不接受输入，因为顾问会读取直到调用为止的整个对话，而不是工作模型编写的查询。因此，系统提示是你设置咨询策略的地方：它控制的是模型何时使用该工具。

SYSTEM = """You are a backend engineer designing HTTP APIs.
You have an advisor: a more capable model that can review your conversation so far and
send back guidance. Consult it with the advisor tool before you commit to any decision
that would be expensive to reverse once clients depend on it: identifier and idempotency
schemes, pagination contracts, error semantics, versioning. Do routine drafting yourself.
When you consult, act on the guidance you get back and say what you changed.
Write your final design to /mnt/session/outputs/design.md."""

designer = client.beta.agents.create(
    name="api_designer",
    description="Designs HTTP APIs, escalating irreversible decisions to an advisor.",
    model={"id": WORKER_MODEL},
    system=SYSTEM,
    tools=[{"type": "agent_toolset_20260401"}],
    multiagent={
        "type": "coordinator",
        "agents": [
            {"type": "advisor", "model": ADVISOR_MODEL},
        ],
    },
    betas=BETAS,
)
print(f"{designer.name}: {designer.id} v{designer.version}")
print("roster:", [entry.to_dict() for entry in designer.multiagent.agents])

3. 交给它一个内部包含硬调用的任务

对于资金转移端点的幂等性方案，提示词告诉代理要升级处理：编写成本低，但在客户端基于其开发后修改成本高。设计的其余部分是常规工作，因此一个校准良好的代理会就此咨询，并自行起草其余部分。

咨询费用仅受咨询轮次的限制：顾问令牌除了工作模型的令牌外还要计费，并且没有每次咨询的上限。会话带有 budget，因此过度升级的习惯会在已知上限处暂停，而不是让账单不断攀升（机制见 CMA_cap_session_spend.ipynb）。

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

session = client.beta.sessions.create(
    agent=designer.id,
    environment_id=env.id,
    title="Design: refunds API",
    budget={"type": "limit", "max_list_cost": {"currency": "USD", "amount": "500"}},
    initial_events=[
        {
            "type": "user.message",
            "content": [
                {
                    "type": "text",
                    "text": "Design the REST surface for a refunds resource on a payments "
                    "API: create, retrieve, list. Refund creation is retried by mobile "
                    "clients on flaky networks, and a double refund costs real money. "
                    "Cover the request/response shapes, idempotency, pagination on list, "
                    "and error codes.",
                }
            ],
        }
    ],
    betas=BETAS,
)
print(session.id, session.status)

4. 在主数据流上观察咨询过程

咨询并不是一个 agent.tool_use 事件。它作为主数据流上的线程生命周期事件出现，其中 agent_name 设置为 anthropic.advisor：

session.thread_created            agent_name: anthropic.advisor
session.thread_status_running     agent_name: anthropic.advisor
agent.thread_message_received     from_agent_name: anthropic.advisor, content: [{type: "text", ...}]
session.thread_status_idle        agent_name: anthropic.advisor
session.thread_status_terminated  agent_name: anthropic.advisor

咨询线程在整个过程中保持 running 状态。建议在调用时传递，采样继续进行，因此从外部看，主线程暂停，一个顾问线程出现并消失，主线程在融入指导后恢复。每次咨询都是一个新线程，在传递建议后自行终止。

def advice_text(content):
    """Render an advisor delivery: joined text, or a marker for the redacted arm."""
    if any(block.type == "redacted" for block in content):
        return "[redacted advice: content withheld by the advisor model's policy]"
    return "".join(block.text for block in content if block.type == "text")


consultations = 0

with client.beta.sessions.events.stream(session.id, betas=BETAS) as stream:
    for ev in stream:
        if ev.type == "session.thread_created" and ev.agent_name == "anthropic.advisor":
            consultations += 1
            print(f"\n--- consulting advisor #{consultations} ({ev.session_thread_id}) ---")
        elif (
            ev.type == "agent.thread_message_received" and ev.from_agent_name == "anthropic.advisor"
        ):
            print(advice_text(ev.content)[:800])
            print("--- advice received, primary thread resumes ---\n")
        elif 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

print(f"\nconsultations this turn: {consultations}")

5. 为咨询定价

由于每次咨询都是一个真实的会话线程，其成本归属于该线程，而不是隐式地计入主线程的数字中。通过每个顾问线程的 session.thread_created 事件中的 id 拉取该线程，并读取其 usage。令牌计数就在那里，还有 usage.list_cost：该线程的令牌按顾问模型的公开列表价格计价，以美元最小单位的整数字符串表示（"131" 表示 $1.31），因此成本来自 API，而不是你维护的费率表。这些令牌在主线程的基础上额外计费，并且它们已经计入会话级别的 usage 聚合中，该聚合带有自己的 list_cost。

一次完成的咨询是一个 terminated 线程。它仍然可以检索，但其 agent 不是通常的快照代理：没有 Agent 资源支持平台顾问，因此线程本身携带名册条目，{"type": "advisor", "model": ...}。agent.type 为 "advisor" 是区分平台顾问线程与普通子线程的可靠方法，比名称更可靠。

from decimal import Decimal


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


advisor_thread_ids = [
    ev.session_thread_id
    for ev in client.beta.sessions.events.list(session.id, limit=1000, betas=BETAS)
    if ev.type == "session.thread_created" and ev.agent_name == "anthropic.advisor"
]

advisor_in = advisor_out = 0
advisor_cents = Decimal("0")
for thread_id in advisor_thread_ids:
    thread = client.beta.sessions.threads.retrieve(thread_id, session_id=session.id, betas=BETAS)
    usage = thread.usage
    advisor_in += usage.input_tokens or 0
    advisor_out += usage.output_tokens or 0
    advisor_cents += Decimal(usage.list_cost.amount)
    print(
        f"{thread.id}  agent.type={thread.agent.type}  model={thread.agent.model}  "
        f"status={thread.status}  "
        f"in={usage.input_tokens} out={usage.output_tokens}  {usd(usage.list_cost)}"
    )

total = client.beta.sessions.retrieve(session.id, betas=BETAS).usage
print(f"\nadvisor tokens: in={advisor_in} out={advisor_out}  cost=${advisor_cents / 100:.2f}")
print(
    f"session total:  in={total.input_tokens} out={total.output_tokens}  "
    f"cost={usd(total.list_cost)}  (advisor included)"
)

6. 阅读设计文档

智能体使用 write 工具撰写了最终文档。从事件日志中获取该文档，并查看幂等性部分，这正是咨询内容所在之处。

design = ""
for ev in client.beta.sessions.events.list(session.id, limit=1000, betas=BETAS):
    if (
        ev.type == "agent.tool_use"
        and ev.name == "write"
        and ev.input.get("file_path", "").endswith("design.md")
    ):
        design = ev.input["content"]  # keep the last write, in case of revisions

for line in design.splitlines():
    if line.startswith("#"):
        print(line)
print()
print(design[:1200])

7. 当建议被编辑时

你是否能阅读咨询内容取决于顾问模型的属性，该属性由管理 Messages API 顾问工具的同一 Anthropic 策略设置。对于输出被策略扣留的模型，交付事件携带 [{"type": "redacted"}] 占位符块而不是文本，并且顾问线程自身的 agent.message 事件也是占位符。

编辑改变的是你观察到的内容，而不是代理使用的内容。工作模型无论哪种方式都能完整读取建议，因此设计仍然得到改进。所有结构性的内容仍然可见：线程、其状态转换、其时间安排以及其 usage，这就是为什么第 4 步中的 advice_text 辅助函数会渲染一个标记而不是崩溃。顾问自身的推理在任何策略下都不会被传递，只有其最终文本会被传递。

由于策略可以在不更改 API 的情况下更改，因此在消费这些事件的任何内容中，将 text 和 redacted 分支视为同样正常。

8. 顾问与专家团队并行

顾问条目与普通名册成员共享 agents 数组，因此协调器可以在同一会话中既委派任务给专家，又咨询更强的模型。只有主线程会进行咨询。它派生的专家线程运行时没有自己的顾问，并且顾问线程不受子线程并发限制的约束，因此即使协调器已达到其子线程上限，仍然可以进行咨询。

client.beta.agents.create(
    name="research_coordinator",
    model={"id": WORKER_MODEL},
    system="Delegate research to your specialists, then synthesize their reports.",
    tools=[{"type": "agent_toolset_20260401"}],
    multiagent={
        "type": "coordinator",
        "agents": [
            "agent_01Res...",                                # a specialist, by id
            {"type": "agent", "id": "agent_01Aud...", "version": 3},
            {"type": "advisor", "model": ADVISOR_MODEL},     # the advisor
        ],
    },
    betas=BETAS,
)

协调器模式本身、按角色划分的工具作用域以及委派事件类型，在 CMA_coordinate_specialist_team.ipynb 中有所涵盖。顾问在创建会话时固定不变：在代理上更改顾问会影响之后创建的会话，而不会影响已经运行的会话。

9. 清理

from utilities import wait_for_idle_status

wait_for_idle_status(client, session.id)
client.beta.sessions.archive(session.id, betas=BETAS)
client.beta.environments.archive(env.id, betas=BETAS)
print("archived")

常见问题

常见错误：1) 未安装 anthropic>=0.121.0 导致类型化名册不可用；2) 模型配对不被允许，agents.create 返回 400；3) 顾问条目名称冲突，占用 anthropic.advisor；4) 未处理 redacted 内容导致崩溃。排查方法：检查 betas 标头、模型版本、事件流中的错误信息。生产优化：设置会话预算限制咨询次数，监控咨询频率，调整系统提示优化咨询策略。

企业应用扩展

可扩展方向：1) 结合专家团队，协调器同时委派任务和咨询顾问；2) 自定义咨询触发条件，通过系统提示精细控制；3) 集成成本监控，实时分析咨询开销；4) 探索不同模型配对的效果，优化性能与成本平衡。

总结

内容来源与验证

内容等级：PREMIUM_A
质量等级：A
难度：高级
验证等级：V1
原文作者：cj-ant
原文日期：2026-07-23
来源许可：MIT
来源版本：ced21c10eab122981d922f19cbef8e81555adb74
查看原文

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员