构建可审计的深度推理 Agent 系统

我们要构建一个深度推理研究 Agent，它能够将复杂的技术问题分解为明确的子论断，对自身的逻辑进行批判性审视，并返回结构化、可审计的回答。这种能力对工程团队很有价值：评估架构权衡、审视事故复盘报告，或者在不必阅读数百页文档的情况下快速上手陌生的系统。

依赖安装：

pip install openai pydantic

还需要从 https://portal.oxlo.ai 获取一个 Oxlo.ai API Key。

建议将 Key 导出为环境变量，这样它永远不会触及磁盘。

第一步：搭建 Oxlo.ai 客户端

我习惯把客户端初始化集中在一处，这样业务逻辑里就不会出现硬编码的 URL。Oxlo.ai 提供了一个完全兼容 OpenAI 的端点，所以直接用这个插入式客户端就够了。

from openai import OpenAI
import os

client = OpenAI(
    base_url="https://api.oxlo.ai/v1",
    api_key=os.environ.get("OXLO_API_KEY")
)

第二步：编写 System Prompt

System Prompt 是整个技术栈中最重要的可调参数。它强制模型在得出结论前外化其思维链，这让调试推理失败变得容易得多。同时我将输出锁定为 JSON，这样下游代码无需解析自然语言。

SYSTEM_PROMPT = """You are a deep reasoning engine. Your job is to answer complex technical questions through explicit, verifiable reasoning.

Follow these rules:
1. Break the user's question into 2 to 5 sub-questions or claims.
2. For each sub-question, provide a short analysis based on first principles.
3. If you lack certainty, state your confidence level and assumptions.
4. Before concluding, run a quick sanity check on your own logic.
5. Return your entire reasoning as a JSON object with two keys: "steps" (a list of strings) and "final_answer" (a string).

Be concise. Avoid speculation beyond what the reasoning supports."""

第三步：用 Pydantic 强制结构化

我用 Pydantic 来定义 LLM 与应用其余部分之间的契约。Oxlo.ai 支持 JSON 模式，所以传入 response_format={"type": "json_object"} 可以保证返回的 JSON 有效且符合该 schema。

import json
from typing import List
from pydantic import BaseModel, Field

class ReasoningOutput(BaseModel):
    steps: List[str] = Field(description="Chain of thought steps")
    final_answer: str = Field(description="Synthesized final answer")

def generate_reasoning(question: str) -> ReasoningOutput:
    response = client.chat.completions.create(
        model="kimi-k2.6",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": question},
        ],
        response_format={"type": "json_object"},
        temperature=0.2,
    )
    content = response.choices[0].message.content
    return ReasoningOutput(**json.loads(content))

第四步：添加自我批判循环

原始的思维链有帮助，但第二轮审查能 catch 住那些懒散的泛化。我把草稿输出反馈给同一个模型，但赋予它评审者身份，让它收紧逻辑。由于 Oxlo.ai 采用按请求计费的定价模式，这额外的往返不会随 token 数量增加而涨价，这让多步推理保持在可承受的成本范围内。

CRITIQUE_PROMPT = """You are a logic reviewer. Review the following reasoning steps and final answer.

Identify any logical gaps, unstated assumptions, or alternative interpretations. Then produce an improved JSON object with the same schema: "steps" and "final_answer". Preserve only sound reasoning."""

def critique_and_refine(draft: ReasoningOutput) -> ReasoningOutput:
    payload = json.dumps(draft.model_dump(), indent=2)
    response = client.chat.completions.create(
        model="kimi-k2.6",
        messages=[
            {"role": "system", "content": CRITIQUE_PROMPT},
            {"role": "user", "content": payload},
        ],
        response_format={"type": "json_object"},
        temperature=0.1,
    )
    data = json.loads(response.choices[0].message.content)
    return ReasoningOutput(**data)

第五步：串联完整流水线

最后，我把两个阶段连接成一个函数，打印中间输出以便在推理展开过程中追踪。

def deep_reason(question: str) -> str:
    print(f"Question: {question}\n")
    
    draft = generate_reasoning(question)
    print("Initial reasoning:")
    for step in draft.steps:
        print(f"  - {step}")
    print()
    
    refined = critique_and_refine(draft)
    print("Refined reasoning:")
    for step in refined.steps:
        print(f"  - {step}")
    print()
    
    return refined.final_answer

将完整脚本保存为 reasoning_agent.py，设置好环境变量，然后运行它。

export OXLO_API_KEY="sk-oxlo.ai-..."
python reasoning_agent.py

以下是我用来测试的入口：

if __name__ == "__main__":
    question = (
        "I am building a distributed task queue. "
        "Should I use at-least-once delivery with idempotent workers, "
        "or exactly-once delivery with stronger coordination? "
        "Consider latency, operational complexity, and failure modes."
    )
    answer = deep_reason(question)
    print("Final Answer:")
    print(answer)

Question: I am building a distributed task queue. Should I use at-least-once delivery with idempotent workers, or exactly-once delivery with stronger coordination? Consider latency, operational complexity, and failure modes.

Initial reasoning:
  - Sub-question 1: What are the latency implications of each model?
  - Sub-question 2: How do failure modes differ under network partitions?
  - Sub-question 3: What operational overhead does idempotency require versus distributed transactions?
  - Sanity check: At-least-once plus idempotency is the default in most large-scale systems, which suggests it is the simpler path.

Refined reasoning:
  - Latency: At-least-once requires only an ack, while exactly-once needs a consensus round or deduplication store lookup.
  - Failure modes: Exactly-once systems can stall if the coordinator fails; at-least-once systems continue but risk duplicate work.
  - Operational complexity: Idempotent workers push complexity to application code, which teams already own, rather than to infrastructure.
  - Sanity check: The claimed latency advantage holds only if the deduplication store is hot; otherwise it is a wash.

Final Answer:
Choose at-least-once delivery with idempotent workers unless you have a strict regulatory requirement for exactly-once semantics. The operational surface area is smaller, recovery from partitions is automatic, and latency remains predictable because you avoid distributed coordination on every enqueue.

后续扩展方向

我会从两个方向扩展这个系统。首先，添加一个检索步骤，把内部 wiki 页面或 API 文档注入到上下文中，让 Agent 的推理扎根于你自己的系统而不是通用知识。其次，通过流式端点暴露中间步骤，这样前端可以在推理到达时实时渲染思维链。这两项扩展都很容易实现，因为 Oxlo.ai 原生支持流式输出、Function Calling 和长上下文窗口。