Mercury 3 教程：构建实时事故分析助手

这只是一个概念性示例，并非 Mercury 3 的精确内部表示。

Inception 的 API 可以暴露这些中间状态。使用普通流式输出时，Mercury 返回可追加到回复中的精修文本块。启用 diffusing=True 后，每个分片则包含模型在优化过程中当前版本的完整答案。

我们稍后会看到差异。

演示：构建实时支持升级助手

在构建演示之前，我们先将模型设置与支持应用本身分开。

支持案例只是我们用来运行 Mercury 3 的输入。重要的模型侧控制包括推理预算、完成预算以及流式模式。

这些是本实验的工作参数，并非 Mercury 3 的全局最优设置。

步骤 1：先决条件

Mercury 3 上手只需几步。要跟进操作，您需要：

• 一个 Inception API 密钥
• Python 3
• inceptionai 包
• 将支持案例保存为 case_support.json

首先注册 Inception 平台并获取 API 密钥。点击 Get Started 并填写信息：

注意： 截至撰写时，Inception 为新账户提供 1 亿免费 tokens 以便试用 API。

您可以为 API 密钥命名，或跳过此步骤直接点击 Create Key。

接着，复制 API 密钥并进行您的第一次 API 调用。

步骤 2：试用 Mercury 3

在编写任何代码之前，您可以直接在 Inception 平台上测试 Mercury 3。创建 API 密钥后，点击 Next，在 Try It Out 选项卡中选择 Run it now 按钮。

这有助于在将提示词和模型设置迁移到应用之前快速进行测试。

步骤 3：准备环境并发起 API 调用

在本教程的其余部分，我们将从 Python 调用 Mercury 3。首先安装 inception SDK，并按如下方式导出 API 密钥：

!pip install -q inceptionai技术支持

然后导入演示所需的包：

import getpass, json, os, time, difflib
from pathlib import Path
from inceptionai import Inception
from IPython.display import clear_output技术支持

getpass 用于在不回显的情况下请求 API 密钥，os 将其放到 SDK 查找的位置，Path 与 json 加载 case_support.json 并将其序列化进提示词，time.perf_counter() 记录每个响应与优化状态到达的时间。

difflib 用于比较相邻的扩散状态，clear_output() 则替换单元输出，以便在同一位置回放扩散过程而不是在页面向下堆叠。

如果尚未设置 API 密钥，您可以交互式输入：

if not os.environ.get("INCEPTION_API_KEY"):
    os.environ["INCEPTION_API_KEY"] = getpass.getpass(
        "Inception API key: "
    )
client = Inception()
MODEL = "mercury-3"
print("ready")技术支持

Inception() 会从环境变量中读取 API 密钥。其余导入中，例如 time 用于记录优化状态的到达时间，difflib 帮助比较中间文本，clear_output() 则让我们在同一位置回放扩散过程。

步骤 4：构建模型消息

支持案例单独存储在 case_support.json 中。该案例文件包含客户消息、订单信息、支付记录、履约日志、支持政策，以及我们要用于此场景的提示词。

您可以自行构建，或直接使用：

{
  "scenario": "Customer reports two charges for one order",
  "system_prompt": "You are a customer-support escalation assistant.\nUse only the evidence in the supplied case file.\nSeparate what the evidence shows from what you infer.\nDo not state that a refund has been issued unless the case file says so.\nIf the evidence is insufficient to decide, say so explicitly.",
  "task_prompt": "Analyze this customer-support case.\n\nReturn:\n1. The most likely explanation.\n2. Three pieces of evidence from the case file that support it.\n3. Two alternative explanations, and what is still uncertain.\n4. The next three actions for the agent, in priority order.\n\nKeep the answer concise. Write it so the customer could read it.",
  "packet_label": "CASE FILE",
  "customer_message": "I ordered one pair of headphones yesterday evening and my card now shows two charges of $129 each. I have one confirmation email and one tracking number. I only pressed buy once. Please refund the second charge today, I cannot afford to be out $258.",
  "order": {
    "order_id": "ORD-1048",
    "placed_at": "2026-10-28T19:41:08Z",
    "status": "shipped",
    "shipped_at": "2026-10-29T08:12:00Z",
    "item": "wireless headphones",
    "quantity": 1,
    "amount_usd": 129,
    "orders_on_account_that_day": 1
  },
  "payment_records": [
    {
      "entry": 1,
      "at": "2026-10-28T19:41:11Z",
      "amount_usd": 129,
      "status": "completed",
      "authorization_code": "A7X41K",
      "card_last_four": "4417"
    },
    {
      "entry": 2,
      "at": "2026-10-28T19:42:45Z",
      "amount_usd": 129,
      "status": "pending",
      "authorization_code": "B2M09T",
      "card_last_four": "4417"
    }
  ],
  "fulfillment_log": [
    "19:41:12 reserved 1 unit for ORD-1048",
    "19:42:47 reserved 1 unit (no order id attached)",
    "19:58:30 released 1 unreserved unit back to stock",
    "08:12:00 shipped 1 unit against ORD-1048, one tracking number issued"
  ],
  "customer_history": [
    {
      "date": "2026-10-07",
      "contact_reason": "reported two charges for one order",
      "outcome": "both charges settled; agent issued a refund for one of them"
    },
    {
      "date": "2026-06-19",
      "contact_reason": "delivery address change",
      "outcome": "resolved, no payment involved"
    }
  ],
  "support_policy": [
    "A pending card authorization can appear on a customer statement as a second charge.",
    "A pending authorization that is never captured usually disappears within 3-5 business days.",
    "Escalate to the payments team if both charges settle.",
    "Agents may not issue a refund against an authorization that has not settled.",
    "Tell the customer what you have confirmed and what you have not."
  ],
  "available_agent_actions": [
    "explain the pending authorization to the customer",
    "set a follow-up for a named number of business days",
    "escalate to the payments team",
    "issue a refund for a settled charge",
    "request the authorization expiry date from the payment processor"
  ]
}技术支持

现在，加载该 JSON 文件：

def load_incident(path: str) -> dict:
    return json.loads(Path(path).read_text())
case = load_incident("case_support.json")技术支持

将其发送给 Mercury 3 之前，我们把提示配置与证据分离：

PROMPT_KEYS = (
    "system_prompt",
    "task_prompt",
    "packet_label",
)
def build_messages(case: dict):
    system = case.get("system_prompt", SYSTEM)
    task = case.get("task_prompt", TASK)
    label = case.get(
        "packet_label",
        "INCIDENT PACKET",
    )
    evidence = {
        k: v
        for k, v in case.items()
        if k not in PROMPT_KEYS
    }
    packet = json.dumps(evidence, indent=2)
    return [
        {
            "role": "system",
            "content": system,
        },
        {
            "role": "user",
            "content": (
                f"{task}\n\n"
                f"{label}:\n{packet}"
            ),
        },
    ]技术支持

键 system_prompt、task_prompt 和 packet_label 控制我们如何指示 Mercury，因此在序列化案例本身之前，它们会从证据中移除。现在我们可以准备最终的消息：

messages = build_messages(case)技术支持

此处没有检索系统或外部工具参与。Mercury 所需的全部分析信息都直接包含在请求中。

步骤 5：向 Mercury 3 发起请求

在测试流式之前，先发起一个普通的非流式请求。以下辅助函数封装了 Mercury 的 API 调用并记录响应元数据：

def usage_dict(response) -> dict:
    usage = getattr(response, "usage", None)

    if usage is None:
        return {}

    if hasattr(usage, "model_dump"):
        return usage.model_dump()

    return dict(usage)技术支持

现在定义一次模型调用：

def run_nonstream(
    client,
    messages,
    effort: str,
    max_tokens: int,
):
    start = time.perf_counter()
    response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        reasoning_effort=effort,
        temperature=0.75,
        max_completion_tokens=max_tokens,
    )
    elapsed = time.perf_counter() - start
    choice = response.choices[0]
    text = choice.message.content or ""
    usage = usage_dict(response)
    return {
        "effort": effort,
        "elapsed_s": elapsed,
        "finish_reason": choice.finish_reason,
        "usage": usage,
        "text": text,
    }技术支持

主要的模型侧控制包括：

reasoning_effort=effort
temperature=0.75
max_completion_tokens=max_tokens技术支持

现在可以进行基线调用：

result = run_nonstream(
    client,
    messages,
    effort="low",
    max_tokens=8192,
)
print(result["text"])技术支持

Mercury 接收完整的支持案例，并返回所需的分析，包括最可能的解释、支持该解释的证据、备选解释与不确定之处，以及推荐的后续行动。

该函数还记录了 elapsed_s 和 usage 字段。前者衡量端到端请求时间，后者包含 token 统计，包括报告的推理 token 用量。

在本教程使用的多次运行中，Mercury 一致地将第二条支付记录判定为待处理的预授权，而非已确认的第二个订单的证据。不过，该案例刻意缺少足够的信息来判断该预授权最终会自动消失还是会结算。

步骤 6：添加标准流式

普通请求会等 Mercury 完成后再返回完整答案。要增量接收响应，我们需要启用 stream=True。标准流式函数如下：

def run_standard_stream(
    client,
    messages,
    effort: str,
    max_tokens: int,
):
    start = time.perf_counter()
    first_output = None
    finish_reason = None
    chunks = []
    stream = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        reasoning_effort=effort,
        temperature=0.75,
        max_completion_tokens=max_tokens,
        stream=True,
    )
    for chunk in stream:
        choices = getattr(
            chunk,
            "choices",
            None,
        ) or []
        if (
            choices
            and getattr(
                choices[0],
                "finish_reason",
                None,
            )
        ):
            finish_reason = choices[0].finish_reason
        content = extract_delta_text(chunk)
        if content:
            now = time.perf_counter()
            if first_output is None:
                first_output = now
            chunks.append(content)
            print(
                content,
                end="",
                flush=True,
            )
    end = time.perf_counter()
    return {
        "time_to_first_output_s": (
            first_output - start
            if first_output
            else None
        ),
        "total_s": end - start,
        "num_chunks": len(chunks),
        "finish_reason": finish_reason,
        "final_text": "".join(chunks),
    }技术支持

此处的重要行为是标准流式中的 chunks.append(content)：每个分片包含回复的下一段内容，因此可以将分片累积，并通过 "".join(chunks) 重建最终答案。

注意：首次可见输出的中位时间为 2.903 秒，总延迟中位数为 3.375 秒。

步骤 7：切换到扩散流式

要启用扩散流式，请将以下参数设为 True：

stream=True
diffusing=True技术支持

与标准流式相比，API 调用仅多了一个参数，即 diffusing=True。

def run_diffusion_stream(
    client,
    messages,
    effort: str,
    max_tokens: int,
    live: bool = True,
):
    start = time.perf_counter()
    first_output = None
    finish_reason = None
    states = []
    empty_chunks = 0
    stream = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        reasoning_effort=effort,
        temperature=0.75,
        max_completion_tokens=max_tokens,
        stream=True,
        diffusing=True,
    )
    for chunk in stream:
        choices = getattr(
            chunk,
            "choices",
            None,
        ) or []
        if (
            choices
            and getattr(
                choices[0],
                "finish_reason",
                None,
            )
        ):
            finish_reason = choices[0].finish_reason
        content = extract_delta_text(chunk)
        if not content:
            empty_chunks += 1
            continue
        now = time.perf_counter()
        if first_output is None:
            first_output = now
        states.append({
            "t_s": now - start,
            "text": content,
        })
        if live:
            clear_output(wait=True)
            print(content, flush=True)
    end = time.perf_counter()
    return {
        "time_to_first_output_s": (
            first_output - start
            if first_output
            else None
        ),
        "total_s": end - start,
        "num_refinement_states": len(states),
        "finish_reason": finish_reason,
        "final_text": (
            states[-1]["text"]
            if states
            else ""
        ),
        "states": states,
    }技术支持

在标准流式中，分片会被追加；而在扩散流式中，最新的非空分片代表当前的完整答案版本。因此不应将其追加，而应替换先前显示的内容。

我们也保留返回的状态，用于衡量优化状态的数量，并检查后续状态是否确实修订了先前出现的文本。

在五次运行中，扩散流式返回的可见优化状态中位数为 9，范围为 7–12。首次可见状态的中位时间为 1.896 秒，总延迟中位数为 2.560 秒。

步骤 8：比较标准与扩散流式

理解两种代码路径后，我们可以比较测得的行为。

对于这个特定支持提示，扩散的延迟中位数更低，但区间重叠显著。因此，不应得出扩散流式本质上更快的结论。

图：12 个状态的运行，逐步过渡。棕色条表示 Mercury 3 已经写过但随后被替换的文本，蓝色条表示先前末尾之后的新文本。若是纯追加，将不会有棕色部分。

在我的运行中，返回的扩散状态会修订先前可见的文本，而非表现为仅追加的分片。在上述运行中，11 次状态过渡都不是纯追加，并且每次都在前 14 个字符内与前一状态产生差异。

步骤 9：测试完成预算

接下来，我们看看在推理 token 与完成预算交互下，Mercury 3 的行为。前文的普通调用使用 max_completion_tokens=8192，但为了更清晰地暴露该行为，我们需要将 max_completion_tokens 降至 1600。

现在，在低与高推理强度下分别运行同一请求。测得结果如下：

LOW
completion_tokens: 964
reasoning_tokens:  620
finish_reason:     stop
visible answer:    returned

HIGH
completion_tokens: 1539
reasoning_tokens:  1539
finish_reason:     length
visible answer:    none技术支持

在高强度下，1,539 个完成 token 全部报告为推理 token。因此模型在返回可见答案文本之前就达到了完成上限。

图：在 1,600 与 8,192 token 上限下的高强度完成预算

在 1,600 的上限下，全部预算都用于推理，已无剩余可输出。在 8,192 的情况下，高强度运行的中位数为 2,292 个推理 token，仍有空间输出 412 个 token 的答案。由于 8,192 上限超出此图刻度的右边界，因此上图中只有第一行显示了虚线上限。

高强度失败请求的重要信号是 response.choices[0].finish_reason 返回了 length。这意味着请求达到了最大完成 token 预算（max_completion_tokens）。该预算同时包含推理 token 与可见输出 token。在该次运行中，Mercury 3 报告的 1,539 个完成 token 均为推理 token，因此在产生可见答案前就到达了完成上限。

注意：对本演示而言，max_completion_tokens=8192 足以覆盖剩余运行，但不应将其视为通用的最低值。

步骤 10：比较低与高推理强度

现在回到正常的 8192 完成预算，仅改变 reasoning_effort 在低与高之间切换。我们将使用步骤 5 的 run_nonstream() 辅助函数，分别在低推理强度与高推理强度下各发起一次请求：

results = []

for effort in ["low", "high"]:
    result = run_nonstream(
        client,
        messages,
        effort=effort,
        max_tokens=8192,
    )
    results.append(result)

    print(
        effort,
        result["elapsed_s"],
        result["usage"],
    )技术支持

对于报告的结果，我对每种配置进行了五次重复，而不是依赖单次请求。

中位数如下：

推理用量的中位数从 828 增加到 2,292 个 token，而延迟中位数从 2.936 秒增加到 8.146 秒（约 2.77 倍）。

图：在各推理强度下，完成预算用于何处？

可见答案的长度变化并不大：

• 低：中位 468 个 token
• 高：中位 412 个 token

且观测的范围有重叠：

• 低：393–494
• 高：356–469

因此，这些运行并不支持“提高推理强度会让可见答案更短”的说法。延迟范围也说明了为什么单次运行不够：

• 低：1.602–4.993 s
• 高：4.102–9.068 s

注意：最快的高强度请求实际上比最慢的低强度请求还要快。

图：十次推理强度请求的延迟散点图

因此，在本实验中，reasoning_effort="high" 显著增加了报告的推理 token 用量和延迟（符合预期），而可见答案长度大体相近。

结语

关于 Mercury 3，有三点实际行为值得注意：

• 扩散流式采用不同的输出契约。 标准流式是追加新文本，而扩散流式返回连续的完整回复状态，应替换先前状态而非追加。

• 推理 tokens 与完成预算共享。同一 max_completion_tokens 若过小，可能在 Mercury 3 产出可见答案前就被推理耗尽。

• 更高的推理强度用更多计算换取更高延迟。在我的运行中，高强度使用了显著更多的报告推理 tokens，耗时也更长（符合预期），而可见答案长度总体相近。

这些结果来自一个合成支持案例上的少量运行，应视为本次演示的观察，而非对 Mercury 3 的通用性能断言。