实战项目——AI 研究助手

项目架构设计

研究助手采用四 Agent 协作架构：入口 Agent 负责接收研究请求并协调，三个专家 Agent 分别负责搜索、分析和写作：

项目结构

research-assistant/
├── app.py              # FastAPI 主应用
├── agents/
│   ├── __init__.py
│   ├── orchestrator.py  # 研究协调 Agent
│   ├── search.py        # 搜索 Agent
│   ├── analyst.py       # 分析 Agent
│   └── writer.py        # 写作 Agent
├── tools/
│   ├── web_search.py    # Tavily 搜索工具
│   ├── pdf_parser.py    # PDF 解析工具
│   └── vector_search.py # 向量检索工具
├── models.py            # Pydantic 数据模型
└── requirements.txt

工具实现

# tools/web_search.py
from agents import function_tool, RunContextWrapper
from dataclasses import dataclass, field
import httpx

@dataclass
class ResearchContext:
    topic: str
    search_results: list[dict] = field(default_factory=list)
    analysis_notes: list[str] = field(default_factory=list)
    sources: list[str] = field(default_factory=list)

@function_tool
async def tavily_search(
    ctx: RunContextWrapper[ResearchContext],
    query: str,
    max_results: int = 5
) -> str:
    """使用 Tavily API 搜索最新信息。适合需要实时数据的研究任务。"""
    import os
    api_key = os.getenv("TAVILY_API_KEY")

    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.tavily.com/search",
            json={
                "api_key": api_key,
                "query": query,
                "max_results": max_results,
                "search_depth": "advanced",
                "include_answer": True
            },
            timeout=30
        )

    data = response.json()

    # 存储到 Context，供后续 Agent 使用
    for result in data.get("results", []):
        ctx.context.search_results.append(result)
        ctx.context.sources.append(result.get("url", ""))

    # 格式化返回给 LLM
    formatted = f"搜索查询：{query}\n"
    formatted += f"综合答案：{data.get('answer', '无')}\n\n"
    formatted += "详细结果：\n"
    for i, r in enumerate(data.get("results", [])[:max_results], 1):
        formatted += f"{i}. [{r.get('title')}]({r.get('url')})\n{r.get('content', '')[:300]}...\n\n"

    return formatted[:4000]  # 限制长度

@function_tool
async def parse_pdf(pdf_url: str) -> str:
    """下载并解析 PDF 文档，提取文字内容。适合学术论文和报告。"""
    import pdfplumber, io
    async with httpx.AsyncClient() as client:
        response = await client.get(pdf_url, timeout=30)
        response.raise_for_status()

    with pdfplumber.open(io.BytesIO(response.content)) as pdf:
        text = ""
        for page in pdf.pages[:10]:  # 最多读 10 页
            text += page.extract_text() or ""

    return text[:5000]  # 截取前 5000 字符

多 Agent 定义

# agents/__init__.py
from agents import Agent, handoff
from pydantic import BaseModel
from tools.web_search import tavily_search, parse_pdf

# ── 输出模型 ───────────────────────────────────────────────
class ResearchReport(BaseModel):
    title: str
    executive_summary: str
    key_findings: list[str]
    detailed_analysis: str
    sources: list[str]
    word_count: int

# ── 搜索 Agent ────────────────────────────────────────────
search_agent = Agent(
    name="搜索专家",
    instructions="""你是信息收集专家。
    接到研究主题后，拆解为 3-5 个关键搜索查询，
    使用 tavily_search 工具逐一搜索，
    遇到 PDF 链接时使用 parse_pdf 工具提取内容。
    完成搜索后，整理出结构化的原始资料摘要。
    """,
    model="gpt-4o-mini",
    tools=[tavily_search, parse_pdf]
)

# ── 分析 Agent ────────────────────────────────────────────
analyst_agent = Agent(
    name="分析专家",
    instructions="""你是数据分析专家。
    基于搜索专家收集的资料，进行深度分析：
    1. 识别核心趋势和关键数据点
    2. 发现不同来源之间的共识与分歧
    3. 评估信息的可靠性和时效性
    4. 提炼出 5-8 个关键发现
    输出结构化的分析笔记，供写作专家使用。
    """,
    model="gpt-4o"   # 分析任务用高质量模型
)

# ── 写作 Agent ────────────────────────────────────────────
writer_agent = Agent(
    name="写作专家",
    instructions="""你是专业研究报告写作专家。
    基于分析专家的笔记，生成高质量研究报告：
    - 执行摘要：200 字以内，点出核心价值
    - 关键发现：5-8 条，每条一句话
    - 详细分析：分节展开，使用数据支撑
    - 引用来源：列出所有参考链接

    报告风格：专业、客观、数据驱动
    """,
    model="gpt-4o",
    output_type=ResearchReport
)

# ── 研究协调 Agent（入口）──────────────────────────────────
research_orchestrator = Agent(
    name="研究协调员",
    instructions="""你是研究项目协调员。收到研究主题后：
    1. 首先转交给搜索专家收集信息
    2. 搜索完成后，转交分析专家进行深度分析
    3. 分析完成后，转交写作专家生成最终报告

    确保每个阶段都完整完成后再进行下一步。
    """,
    model="gpt-4o-mini",
    handoffs=[
        handoff(search_agent),
        handoff(analyst_agent),
        handoff(writer_agent),
    ]
)

FastAPI WebSocket 流式接口

# app.py
from fastapi import FastAPI, WebSocket
from agents import Runner
from agents.stream_events import RunItemStreamEvent, AgentUpdatedStreamEvent
from agents_module import research_orchestrator, ResearchReport, ResearchContext
import json

app = FastAPI(title="AI 研究助手")

@app.websocket("/research/stream")
async def research_websocket(websocket: WebSocket):
    await websocket.accept()

    try:
        # 接收研究主题
        data = await websocket.receive_json()
        topic = data.get("topic", "")

        if not topic:
            await websocket.send_json({"error": "请提供研究主题"})
            return

        # 初始化研究上下文
        ctx = ResearchContext(topic=topic)

        # 发送开始消息
        await websocket.send_json({
            "type": "start",
            "message": f"开始研究：{topic}"
        })

        # 流式运行研究工作流
        stream = Runner.run_streamed(
            research_orchestrator,
            f"请对以下主题进行深度研究并生成报告：{topic}",
            context=ctx
        )

        async for event in stream.stream_events():

            # Agent 切换：通知前端当前执行阶段
            if isinstance(event, AgentUpdatedStreamEvent):
                await websocket.send_json({
                    "type": "agent_switch",
                    "agent": event.new_agent.name,
                    "message": f"进入阶段：{event.new_agent.name}"
                })

            # 工具调用：通知前端正在搜索什么
            elif isinstance(event, RunItemStreamEvent):
                item = event.item
                if item.type == "tool_call_item":
                    tool_input = item.raw_item.arguments if hasattr(item.raw_item, "arguments") else ""
                    await websocket.send_json({
                        "type": "tool_call",
                        "tool": item.raw_item.name,
                        "message": f"调用工具：{item.raw_item.name}"
                    })

        # 发送最终报告
        final_output = stream.final_output
        if isinstance(final_output, ResearchReport):
            await websocket.send_json({
                "type": "complete",
                "report": final_output.model_dump()
            })

    except Exception as e:
        await websocket.send_json({"type": "error", "message": str(e)})
    finally:
        await websocket.close()

与 Ollama 集成：本地模型替代

# 使用 Ollama 运行本地模型（隐私数据场景）
# 前提：本地运行 ollama pull qwen2.5:7b
from openai import AsyncOpenAI
from agents import Agent, set_default_openai_client

# Ollama 兼容 OpenAI API 格式
ollama_client = AsyncOpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama"  # Ollama 不需要真实 API Key
)

# 可以为不同 Agent 指定不同的客户端/模型
# 搜索 Agent 用 OpenAI（需要强推理），分析 Agent 用本地模型（保护数据）
local_analyst = Agent(
    name="本地分析专家",
    instructions="分析搜索结果，提炼关键信息。",
    model="qwen2.5:7b",  # Ollama 本地模型名称
    model_settings=ModelSettings(
        openai_client=ollama_client  # 指定使用 Ollama 客户端
    )
)

# 混合使用：公开数据用 OpenAI，私有数据用本地
# 这个架构在金融、医疗等合规场景非常实用

部署注意事项

未来展望：SDK 路线图与 MCP 融合