如何用 Python 构建多智能体研究助手

pip install -q -U openai-agents olostep python-dotenv pydantic

OPENAI_API_KEY=your_openai_api_key
OLOSTEP_API_KEY=your_olostep_api_key

import json
import os
from datetime import datetime
from typing import Any
from dotenv import load_dotenv
from IPython.display import Markdown, display
from olostep import Olostep
from pydantic import BaseModel, Field
from agents import (
    Agent,
    Runner,
    custom_span,
    flush_traces,
    function_tool,
    gen_trace_id,
    trace,
)

load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
OLOSTEP_API_KEY = os.getenv("OLOSTEP_API_KEY")

client = Olostep(api_key=OLOSTEP_API_KEY)
search = client.searches.create(
    query="What are the most important recent developments in AI agents for business research?",
    limit=5,
    scrape_options={"formats": ["markdown"], "timeout": 25},
)

for link in search.links:
    print(link["url"], "-", len(link.get("markdown_content") or ""), "chars")

class OlostepError(RuntimeError):
    """当 Olostep SDK 请求失败时抛出。"""

def require_olostep_key() -> str:
    if not OLOSTEP_API_KEY:
        raise OlostepError(
            "未设置 OLOSTEP_API_KEY。请将其添加至 .env 文件，并重新运行设置单元格。"
        )
    return OLOSTEP_API_KEY

def get_olostep_client() -> Olostep:
    return Olostep(api_key=require_olostep_key())

def sdk_result_to_dict(result: Any) -> dict[str, Any]:
    if hasattr(result, "model_dump"):
        return result.model_dump()
    if hasattr(result, "__dict__"):
        return {
            key: value
            for key, value in vars(result).items()
            if not key.startswith("_")
        }
    return {"value": str(result)}

def compact_json(data: Any, max_chars: int = 8000) -> str:
    text = json.dumps(data, ensure_ascii=False, indent=2, default=str)
    if len(text) <= max_chars:
        return text
    return text[:max_chars] + "\n... [已截断]"

def current_date_context() -> str:
    return datetime.now().strftime("%Y年%m月%d日")

def current_year_context() -> str:
    return str(datetime.now().year)

def normalize_search_links(
    links: list[dict[str, Any]], limit: int = 8
) -> list[dict[str, Any]]:
    rows = []
    for link in links[:limit]:
        markdown = link.get("markdown_content") or ""
        rows.append(
            {
                "title": link.get("title") or "无标题",
                "url": link.get("url") or "",
                "description": link.get("description") or "",
                "markdown_chars": len(markdown),
                "markdown_preview": markdown[:1500] if markdown else "",
            }
        )
    return rows

class Judgment(BaseModel):
    is_good_enough: bool = Field(
        description="该答案是否足以满足用户查询需求（即评分 ≥ 0.85）。"
    )
    score: float = Field(ge=0, le=1, description="质量评分，范围为 0 至 1。")
    reason: str = Field(description="对该决策的简要说明。")
    missing_information: list[str] = Field(
        default_factory=list, description="需要补充的重要信息缺口。"
    )

class MarkdownResearchReport(BaseModel):
    markdown_report: str = Field(
        description="完整的 Markdown 报告，包含规范的标题、清晰的分析、读者友好的结构以及引用来源。"
    )

@function_tool
async def answer_query(query: str) -> str:
    """使用 Olostep Answer API 回答用户的自然语言研究问题。"""
    try:
        with custom_span("olostep.answer_query", {"query": query}):
            result = get_olostep_client().answers.create(task=query)
            return compact_json(sdk_result_to_dict(result))
    except Exception as exc:
        raise OlostepError(f"Olostep Answer API 调用失败：{exc}") from exc

@function_tool
async def search_web(query: str, limit: int = 8) -> str:
    """使用 Olostep Search 进行网页搜索，并返回规范化结果。"""
    try:
        with custom_span("olostep.search_web", {"query": query, "limit": limit}):
            search = get_olostep_client().searches.create(
                query=query,
                limit=limit,
            )

data = sdk_result_to_dict(search)

return compact_json(
    {
        "query": query,
        "results": normalize_search_links(
            data.get("links", []),
            limit=limit,
        ),
    }
)
except Exception as exc:
    raise OlostepError(f"Olostep Search API failed: {exc}") from exc

@function_tool
async def search_with_scrape(query: str, limit: int = 5) -> str:
    """使用 Olostep Search with Scrape 在网页中搜索并抓取每个返回链接。"""
    scrape_options = {
        "formats": ["markdown"],
        "timeout": 25,
    }
    try:
        with custom_span(
            "olostep.search_with_scrape",
            {
                "query": query,
                "limit": limit,
                "scrape_options": scrape_options,
            },
        ):
            search = get_olostep_client().searches.create(
                query=query,
                limit=limit,
                scrape_options=scrape_options,
            )
        data = sdk_result_to_dict(search)
        return compact_json(
            {
                "query": query,
                "results": normalize_search_links(
                    data.get("links", []),
                    limit=limit,
                ),
            },
            max_chars=12000,
        )
    except Exception as exc:
        raise OlostepError(f"Olostep Search with Scrape failed: {exc}") from exc

@function_tool
async def scrape_url(url: str) -> str:
    """使用 Olostep 抓取单个 URL 并返回精简后的页面内容。"""
    try:
        with custom_span(
            "olostep.scrape_url",
            {
                "url": url,
                "formats": ["markdown"],
            },
        ):
            scrape = get_olostep_client().scrapes.create(
                url=url,
                formats=["markdown"],
            )
        return compact_json(
            {
                "url": url,
                "scrape": sdk_result_to_dict(scrape),
            },
            max_chars=10000,
        )
    except Exception as exc:
        raise OlostepError(f"Olostep Scrape API failed: {exc}") from exc

MODEL = "gpt-5.4-mini"

judge_agent = Agent(
    name="Judge agent",
    model=MODEL,
    instructions=(
        "你需判断所提供的答案是否足以回应原始研究问题。"
        "优先奖励直接、具体、有可靠来源支撑的答案；拒绝模糊、陈旧或缺乏支撑的答案。"
        "要求严格：仅当评分 ≥ 0.85 且证据能直接、具体地回答问题（含具体来源内容、领域特定细节及适当时效性）时，才可标记 is_good_enough 为 true。"
        "对于时事、产品状态、政策、定价或可能变动的事实性主张，必须引用近期的一手或高度权威来源。"
        "若存在任何关键缺失，不得标记证据为充分。"
        "评分标准如下："
        "0.85–1.0：证据充分，可终止研究，具备强来源支撑且无关键缺失；"
        "0.75–0.84：证据较强，但仍缺失一项重要来源、细节、时效性验证或覆盖范围；"
        "0.50–0.74：存在部分相关证据，但需进一步研究；"
        "0.25–0.49：证据薄弱、模糊、陈旧或关联性弱；"
        "低于 0.25：仅适用于空内容、不可用或基本无关的证据。"
        "即使证据看似合理或方向正确，也不应标记为充分。"
        "仅返回结构化判断结果。"
    ),
    output_type=Judgment,
)

analyst_agent = Agent(
    name="Analyst agent",
    model=MODEL,
    instructions=(
        "你需基于证据撰写一份规范的 Markdown 格式研究报告。"
        "面向希望获得任何主题下清晰、精炼研究简报的专业读者。"
        "报告应根据用户问题进行适配，markdown_report 内容须充实、易于扫描，仅使用以下通用章节："
    ),
)

judge_tool = judge_agent.as_tool(
    tool_name="judge_answer_quality",
    tool_description="评估某个答案或证据是否足以回应原始研究问题。",
)

analyst_tool = analyst_agent.as_tool(
    tool_name="write_markdown_research_report",
    tool_description="基于所收集的证据，撰写最终结构化 Markdown 研究报告。",
)

manager_agent = Agent(
    name="Manager research agent",
    model=MODEL,
    instructions=(
        f"Current date: {current_date_context()}\n"
        f"Current year: {current_year_context()}\n\n"
        "You are the orchestrator for a multi-agent research assistant. You must manage the workflow, "
        "not answer from your own memory. Follow this policy exactly:\n"
        "1. Always call answer_query first to get a simple initial answer for the user's question.\n"
        "2. Immediately call judge_answer_quality on the original question plus the answer_query result. "
        "If the judge returns is_good_enough=true and score >= 0.85, stop researching and call "
        "write_markdown_research_report with the question, answer result, and judgment.\n"
        "3. If the first judgment is weak, call search_with_scrape for the original question. "
        "Immediately call judge_answer_quality again on the original question plus the answer_query result, "
        "first judgment, and search_with_scrape result. If this second judge returns is_good_enough=true "
        "and score >= 0.85, stop researching and call write_markdown_research_report with all evidence.\n"
        "4. If the second judgment is still weak, do not call the judge again. Run multiple targeted "
        "search_web calls first, using the judge's missing_information to form the searches. Inspect the "
        "search results, choose at least the top 3 relevant source URLs most likely to answer the missing "
        "points, then call scrape_url on each of those top 3 pages. Scrape more than 3 only if clearly needed.\n"
        "5. Call write_markdown_research_report exactly once at the end, using every answer, judgment, "
        "search result, and scraped page. The analyst must produce the final MarkdownResearchReport.\n"
        "6. Return only the final MarkdownResearchReport. Do not return a casual chat answer, tool transcript, or plan."
    ),
    tools=[
        answer_query,
        judge_tool,
        search_with_scrape,
        search_web,
        scrape_url,
        analyst_tool,
    ],
    output_type=MarkdownResearchReport,
)

51 def openai_trace_url(trace_id: str) -> str:
52     return f"https://platform.openai.com/logs/trace?trace_id={trace_id}"

async def run_research_assistant(query: str) -> MarkdownResearchReport:
    if not OPENAI_API_KEY:
        raise RuntimeError(
            "OPENAI_API_KEY 未设置。请将其添加至 .env 文件并重新运行设置单元格。"
        )
    require_olostep_key()
    trace_id = gen_trace_id()
    print("OpenAI trace ID:", trace_id)
    print("OpenAI trace URL:", openai_trace_url(trace_id))
    current_date = current_date_context()
    current_year = current_year_context()
    prompt = f"""
当前日期：
{current_date}

当前年份：
{current_year}

研究问题：
{query}

请针对用户的特定问题，返回一份内容详实、易于阅读的 Markdown 格式研究报告。请严格遵循以下工作流程：

- 首先使用 answer_query 获取一个初步的简要答案；
- 紧接着使用 judge agent 判断是否应停止或继续；
- 若首次判断认为答案不够充分，则运行 search_with_scrape；
- 在 search_with_scrape 完成后，立即再次使用 judge agent 判断是否应停止或继续；
- 若第二次判断仍认为证据不足，则不再重复判断，而是执行多次有针对性的 search_web 调用，从搜索结果中至少选取前 3 个最相关的源 URL，并对这 3 个页面进行内容抓取以获取上下文；
- 最后由 analyst agent 综合所有 answer、judge、search 和 scrape 的结果，撰写最终的 Markdown 报告。注意：报告中无需包含 Limitations（局限性）或 Next Steps（后续步骤）部分。
"""

    with trace(
        workflow_name="multi_agent_research_assistant_olostep",
        trace_id=trace_id,
        metadata={
            "query": query,
            "notebook": "multi_agent_research_assistant_openai_agents_olostep",
        },
    ):
        with custom_span("manager.run", {"query": query}):
            result = await Runner.run(manager_agent, prompt, max_turns=30)
        flush_traces()
        print(
            "Trace 已刷新。请打开上方 URL 查看 manager、专业 agent、工具及 Olostep 的 span 数据。"
        )
        return result.final_output

example_query = "What's going on with OpenAI's Sora shutting down?"
report = await run_research_assistant(example_query)
display(Markdown(report.markdown_report))

OpenAI trace ID: trace_45f3333363da420dbcefc8bb8819224c
OpenAI trace URL: https://platform.openai.com/logs/trace?trace_id=trace_45f3333363da420dbcefc8bb8819224c

git clone https://github.com/kingabzpro/Multi-Agent-Research-Assistant.git

cd Multi-Agent-Research-Assistant
pip install -r requirements.txt

OPENAI_API_KEY=your_openai_api_key
OLOSTEP_API_KEY=your_olostep_api_key
OPENAI_MODEL=gpt-5.4-mini

reflex run