用 Python 构建一个可控的多 Agent 研究助手

用户问题
  ↓
Manager Agent：决定先快速回答还是继续检索
  ↓
Search / Scrape 工具：收集候选证据
  ↓
Judge Agent：判断证据是否足够，给出评分和缺口
  ↓ 如果不够，继续补证
Analyst Agent：生成结构化研究报告

mkdir -p multi_agent_research_assistant
cd multi_agent_research_assistant
python3 -m venv .venv
source .venv/bin/activate

multi_agent_research_assistant/
├── .env.example
├── README.md
├── app.py                  # 离线可跑的最小闭环
├── real_tools.py            # 真实 OpenAI / Olostep 接入骨架
├── requirements.txt
└── samples/
    └── evidence.json

cat > requirements.txt <<'EOF'
openai-agents
olostep
python-dotenv
pydantic
EOF

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Protocol


@dataclass
class Evidence:
    title: str
    url: str
    content: str
    source_type: str = "web"


@dataclass
class Judgment:
    is_good_enough: bool
    score: float
    reason: str
    missing_information: list[str] = field(default_factory=list)


class SearchTool(Protocol):
    def search(self, query: str, limit: int = 3) -> list[Evidence]: ...


class MockSearchTool:
    def search(self, query: str, limit: int = 3) -> list[Evidence]:
        records = [
            Evidence(
                title="AI agents in business research",
                url="https://example.com/agents-business-research",
                content=(
                    "AI agents can search sources, compare evidence, and draft reports. "
                    "Production systems need source tracking and quality checks."
                ),
            ),
            Evidence(
                title="Agent evaluation patterns",
                url="https://example.com/agent-evaluation",
                content=(
                    "A judge component can score evidence sufficiency. "
                    "Thresholds such as 0.85 help decide when to stop searching."
                ),
            ),
            Evidence(
                title="Failure modes of web research agents",
                url="https://example.com/research-agent-failures",
                content=(
                    "Search APIs may return empty pages, stale snippets, or duplicated sources. "
                    "Systems should enforce retry limits and budget caps."
                ),
            ),
        ]
        return records[:limit]


class JudgeAgent:
    def judge(self, question: str, evidence: list[Evidence]) -> Judgment:
        if not evidence:
            return Judgment(False, 0.0, "没有证据", ["至少需要 2 个来源"])

        unique_urls = {item.url for item in evidence}
        has_quality_check = any("score" in item.content.lower() or "quality" in item.content.lower() for item in evidence)
        has_failure_mode = any("empty" in item.content.lower() or "failure" in item.content.lower() for item in evidence)

        score = 0.35
        score += min(len(unique_urls), 3) * 0.15
        if has_quality_check:
            score += 0.20
        if has_failure_mode:
            score += 0.15
        score = min(score, 1.0)

        missing: list[str] = []
        if len(unique_urls) < 2:
            missing.append("需要至少 2 个独立来源")
        if not has_quality_check:
            missing.append("缺少质量评估机制")
        if not has_failure_mode:
            missing.append("缺少失败模式和降级策略")

        return Judgment(
            is_good_enough=score >= 0.85,
            score=score,
            reason=f"当前证据评分 {score:.2f}",
            missing_information=missing,
        )


class AnalystAgent:
    def write_report(self, question: str, evidence: list[Evidence], judgment: Judgment) -> str:
        sources = "\n".join(f"- {item.title}: {item.url}" for item in evidence)
        findings = "\n".join(f"- {item.content}" for item in evidence)
        return f"""# Research Report

## Executive Summary

问题：{question}

当前证据评分：{judgment.score:.2f}。结论：多 Agent 研究助手应采用 Manager 编排、Judge 评估、Analyst 生成报告的闭环结构。

## Key Findings

{findings}

## Source Notes

{sources}
"""


class ManagerAgent:
    def __init__(self, search_tool: SearchTool, judge: JudgeAgent, analyst: AnalystAgent) -> None:
        self.search_tool = search_tool
        self.judge = judge
        self.analyst = analyst

    def run(self, question: str, max_rounds: int = 2) -> str:
        all_evidence: list[Evidence] = []
        last_judgment = Judgment(False, 0.0, "尚未评估", [])

        for round_index in range(1, max_rounds + 1):
            query = question if round_index == 1 else question + " " + " ".join(last_judgment.missing_information)
            new_evidence = self.search_tool.search(query=query, limit=3)
            all_evidence.extend(new_evidence)

            last_judgment = self.judge.judge(question, all_evidence)
            print(f"round={round_index} score={last_judgment.score:.2f} good={last_judgment.is_good_enough}")

            if last_judgment.is_good_enough:
                break

        if not last_judgment.is_good_enough:
            return (
                "证据仍未达标，停止生成最终报告。\n"
                f"原因：{last_judgment.reason}\n"
                f"缺口：{', '.join(last_judgment.missing_information)}"
            )

        return self.analyst.write_report(question, all_evidence, last_judgment)


def main() -> None:
    manager = ManagerAgent(
        search_tool=MockSearchTool(),
        judge=JudgeAgent(),
        analyst=AnalystAgent(),
    )
    report = manager.run("How should a business research assistant use AI agents?")
    print("\n" + report)


if __name__ == "__main__":
    main()

python app.py

round=1 score=1.00 good=True

# Research Report

## Executive Summary
...
## Key Findings
...
## Source Notes
...

cat > .env.example <<'EOF'
OPENAI_API_KEY=your_openai_api_key
OLOSTEP_API_KEY=your_olostep_api_key
EOF
cp .env.example .env

pip install -r requirements.txt

from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Any

from dotenv import load_dotenv
from olostep import Olostep

load_dotenv()


@dataclass
class Evidence:
    title: str
    url: str
    content: str
    source_type: str = "web"


class ToolError(RuntimeError):
    pass


def require_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise ToolError(f"Missing environment variable: {name}")
    return value


def safe_text(value: Any, max_chars: int = 8000) -> str:
    text = "" if value is None else str(value)
    text = text.strip()
    if len(text) > max_chars:
        return text[:max_chars] + "\n... [truncated]"
    return text


class OlostepSearchTool:
    def __init__(self) -> None:
        self.client = Olostep(api_key=require_env("OLOSTEP_API_KEY"))

    def search(self, query: str, limit: int = 5) -> list[Evidence]:
        try:
            result = self.client.searches.create(
                query=query,
                limit=limit,
                scrape_options={"formats": ["markdown"], "timeout": 25},
            )
        except Exception as exc:
            raise ToolError(f"Olostep search failed: {exc}") from exc

        evidence: list[Evidence] = []
        for link in getattr(result, "links", []) or []:
            url = link.get("url", "")
            title = link.get("title") or url
            markdown = link.get("markdown_content") or link.get("description") or ""
            if not url or len(markdown.strip()) < 200:
                continue
            evidence.append(
                Evidence(
                    title=title,
                    url=url,
                    content=safe_text(markdown),
                )
            )
        return evidence

from dataclasses import dataclass, field


@dataclass
class Judgment:
    is_good_enough: bool
    score: float
    reason: str
    missing_information: list[str] = field(default_factory=list)

你是研究质量评估员。请判断当前证据是否足以回答用户问题。

评分标准：
- 0.85-1.00：证据充分，来源可信，无关键空白
- 0.75-0.84：信息较强，但缺少一个重要来源、细节或时效检查
- 0.50-0.74：只有局部证据，需要继续搜索
- 0.25-0.49：数据单薄、陈旧或弱相关
- <0.25：空数据或完全不相关

输出 JSON：
{
  "is_good_enough": true/false,
  "score": 0.0-1.0,
  "reason": "短解释",
  "missing_information": ["仍需补充的信息"]
}

# Research Report

## Executive Summary
一句话回答问题，并说明证据强度。

## Key Findings
用 bullet 列出核心发现，每条对应来源。

## Context
解释背景和问题边界。

## Evidence Review
说明证据来源、时间、可信度、冲突点。

## Detailed Analysis
展开分析，不把推论伪装成事实。

## Implications
说明对业务、技术或决策的影响。

## Source Notes
列出来源 URL 和抓取局限。

## References
列出引用链接。

你是研究流程编排器。

必须按以下步骤执行：
1. 先调用快速回答工具，获得初始答案或初始搜索方向。
2. 调用 judge_answer_quality 评估证据质量。
3. 如果 score >= 0.85 且 is_good_enough=true，调用 analyst 生成最终报告，然后停止。
4. 如果不达标，调用 search_with_scrape，并优先搜索 missing_information 指出的缺口。
5. 如果搜索结果中有高价值 URL 但内容不足，调用 scrape_url 进行 URL 级补抓。
6. 达到 max_rounds、max_sources、max_cost 或连续空结果时停止，不要无限循环。

def has_enough_text(evidence: list[Evidence], min_chars: int = 500) -> bool:
    return any(len(item.content.strip()) >= min_chars for item in evidence)

@dataclass
class ResearchBudget:
    max_rounds: int = 3
    max_sources: int = 8
    max_scrape_calls: int = 5

def dedupe_by_url(items: list[Evidence]) -> list[Evidence]:
    seen: set[str] = set()
    result: list[Evidence] = []
    for item in items:
        if item.url in seen:
            continue
        seen.add(item.url)
        result.append(item)
    return result

- 用户问题
- 搜索 query
- 使用的 URL
- Judge 分数
- missing_information
- 最终报告路径

编排器负责调度 → 工具负责取数 → Judge 负责质量闸门 → Analyst 负责表达 → 预算和日志负责生产可控