深度研究 Agent 落地生产的实战教训

import json
import os
from datetime import datetime
from typing import Literal

from openai import OpenAI

client = OpenAI()  # 默认读取 OPENAI_API_KEY 环境变量

STATE_FILE = "research_state.json"
MAX_ITERATIONS = 3

# ── 状态管理 ──────────────────────────────────────────────

def load_state() -> dict:
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE) as f:
            return json.load(f)
    return {"sub_questions": [], "results": {}, "iteration": 0, "status": "planning"}

def save_state(state: dict):
    with open(STATE_FILE, "w") as f:
        json.dump(state, f, indent=2, ensure_ascii=False)

# ── Agent 定义 ────────────────────────────────────────────

def planner_agent(question: str, feedback: str | None = None) -> list[str]:
    """拆解研究问题为子问题列表"""
    prompt = f"""你是一个研究规划师。把以下研究问题拆解为 3-5 个具体可检索的子问题。
只输出 JSON 数组，不要其他内容。

研究问题：{question}
"""
    if feedback:
        prompt += f"\n上一轮审阅反馈（请据此调整子问题）：{feedback}"

    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
    )
    raw = json.loads(resp.choices[0].message.content)
    # 期望 {"sub_questions": ["q1", "q2", ...]}
    return raw.get("sub_questions", [])

def searcher_agent(sub_question: str) -> dict:
    """对单个子问题执行模拟检索与摘要"""
    prompt = f"""你是一个信息检索与摘要专家。针对以下子问题，
模拟多跳检索过程，输出：
1. findings: 你找到的关键信息（2-3 条）
2. sources: 信息来源（URL 或文档名）
3. confidence: 高/中/低
4. gaps: 还有什么信息缺失

只输出 JSON，不要其他内容。

子问题：{sub_question}
"""
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
    )
    return json.loads(resp.choices[0].message.content)

def reviewer_agent(question: str, report: str) -> dict:
    """审阅报告完整性，输出 pass/fail + 具体反馈"""
    prompt = f"""你是一个研究审阅者。检查以下报告是否完整回答了原始问题。
输出 JSON：
- decision: "pass" 或 "fail"
- feedback: 如果 fail，给出具体缺失方向（不要笼统说"再研究一下"）

原始问题：{question}

报告：
{report}
"""
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
    )
    return json.loads(resp.choices[0].message.content)

# ── 主循环 ────────────────────────────────────────────────

def run_research(question: str) -> str:
    state = load_state()
    feedback = None

    for i in range(MAX_ITERATIONS):
        state["iteration"] = i + 1
        print(f"\n=== 第 {i+1} 轮迭代 ===")

        # 1. 规划
        sub_questions = planner_agent(question, feedback)
        state["sub_questions"] = sub_questions
        print(f"子问题：{sub_questions}")

        # 2. 检索（每个子问题一个 searcher）
        results = {}
        for sq in sub_questions:
            result = searcher_agent(sq)
            results[sq] = result
            print(f"  [{sq}] → confidence={result.get('confidence')}, gaps={result.get('gaps')}")
        state["results"] = results

        # 3. 合成报告
        findings_text = ""
        for sq, r in results.items():
            findings_text += f"\n## {sq}\n"
            for f in r.get("findings", []):
                findings_text += f"- {f}\n"
            findings_text += f"来源: {r.get('sources', [])}\n"
            findings_text += f"置信度: {r.get('confidence')}\n"
            if r.get("gaps"):
                findings_text += f"信息缺口: {r['gaps']}\n"

        report = f"# 研究报告：{question}\n{findings_text}"
        state["report"] = report

        # 4. 审阅
        review = reviewer_agent(question, report)
        decision = review.get("decision", "fail")
        print(f"审阅结果：{decision}")

        save_state(state)

        if decision == "pass":
            print("✅ 报告通过审阅")
            return report
        else:
            feedback = review.get("feedback", "")
            print(f"❌ 需要改进：{feedback}")

    print("⚠️ 达到最大迭代次数，返回当前报告")
    return state.get("report", "未能生成完整报告")

# ── 运行 ──────────────────────────────────────────────────

if __name__ == "__main__":
    question = "2024-2025年主流大模型厂商的推理能力对比，包括价格、延迟和准确率"
    final_report = run_research(question)
    print("\n" + final_report)