个人研究助理Agent开发实践

一、研究助理Agent的核心价值与应用场景

在AI Agent的落地实践中，研究助手是最经典且实用的场景之一。为什么这么说？因为研究任务的流程天然适合状态机建模：搜索→分析→总结，每个步骤都有明确的输入输出，且需要多角色协作完成。

1.1 为什么研究助手是Agent入门的最佳项目？

• 流程清晰：搜索→分析→总结，天然适合状态机建模
• 多Agent协作：研究员、审查员、撰稿人各司其职
• 工具调用密集：搜索API、内容提取、数据处理
• 实用性强：可用于竞品分析、市场调研、学术研究

1.2 典型应用场景

1.3 最终效果演示

输入一个研究主题（如"2026年AI编程工具市场分析"），系统自动完成：

1. 生成研究大纲和关键子问题
2. 并行搜索多个资料来源
3. 审查信息准确性和完整性
4. 生成结构化的研究报告

二、研究助理Agent的系统架构设计

2.1 总体架构

一个完整的研究助理Agent通常包含以下组件：

2.2 技术栈选择

2.3 多Agent团队设计

参考GPT-Researcher项目的7 Agent架构，一个标准的研究团队应包含：

三、LangGraph核心实现

3.1 状态定义——整个系统的"数据库"

LangGraph的核心是状态（State）。所有Agent通过共享状态传递信息：

# state.py
from typing import TypedDict, Annotated, List, Dict, Optional
from langgraph.graph.message import add_messages
import operator

class ResearchState(TypedDict):
    """研究助手的全局状态"""
    
    # 研究主题和配置
    topic: str                              # 研究主题
    language: str                           # 输出语言（zh/en）
    format: str                             # 输出格式（markdown/html）

    # 编辑阶段
    outline: Optional[List[str]]            # 研究大纲
    sub_questions: Optional[List[str]]      # 子问题列表

    # 研究阶段（按子问题组织）
    research_findings: Annotated[
        Dict[str, List[str]], 
        operator.add  # 并行写入时内容合并，而非覆盖
    ]

    # 审查阶段
    review_results: Annotated[
        Dict[str, str], 
        operator.add
    ]

    # 修订阶段
    revised_findings: Annotated[
        Dict[str, List[str]], 
        operator.add
    ]

    # 最终输出
    final_report: Optional[str]             # 最终报告

    # 控制流
    messages: Annotated[list, add_messages] # 对话消息记录
    current_step: str                        # 当前执行步骤
    error_count: int                        # 错误计数

3.2 编辑Agent——生成研究大纲

# agents/editor.py
from langchain_core.messages import SystemMessage, HumanMessage

def editor_agent(state: ResearchState) -> dict:
    """编辑智能体：生成研究大纲和子问题"""
    
    system_prompt = """你是一位资深研究编辑。你的任务是：
    1. 将研究主题分解为3-5个关键子问题
    2. 生成研究大纲
    3. 确保子问题之间不重叠，覆盖主题的各个维度

    输出格式：
    ## 大纲
    - 子问题1
    - 子问题2
    - 子问题3

    ## 子问题
    1. 具体的子问题1
    2. 具体的子问题2
    3. 具体的子问题3"""
    
    messages = [
        SystemMessage(content=system_prompt),
        HumanMessage(content=f"研究主题：{state['topic']}")
    ]
    
    response = llm.invoke(messages)
    
    # 解析大纲和子问题
    outline, sub_questions = parse_outline(response.content)
    
    return {
        "outline": outline,
        "sub_questions": sub_questions,
        "current_step": "research",
        "messages": [response]
    }

def parse_outline(text: str) -> tuple:
    """解析LLM返回的大纲和子问题"""
    outline = []
    questions = []
    
    in_outline = False
    in_questions = False
    
    for line in text.split("\n"):
        if "## 大纲" in line:
            in_outline = True
            in_questions = False
            continue
        elif "## 子问题" in line:
            in_outline = False
            in_questions = True
            continue
        
        line = line.strip()
        if not line:
            continue
        
        if in_outline and line.startswith("-"):
            outline.append(line[1:].strip())
        elif in_questions and (line[0].isdigit()):
            questions.append(line.split(".", 1)[1].strip())
    
    return outline, questions

3.3 研究员Agent——并行搜索（核心）

这是最核心的步骤。使用LangGraph的 Send API实现并行处理：

# agents/researcher.py
from langgraph.constants import Send

def research_orchestrator(state: ResearchState) -> list:
    """研究调度器：为每个子问题分配研究员（并行执行）"""
    return [
        Send(
            "researcher_worker",
            {
                "sub_question": q,
                "topic": state["topic"],
            }
        )
        for q in state["sub_questions"]
    ]

def researcher_worker(state: dict) -> dict:
    """研究员工作节点：搜索特定子问题的资料"""
    sub_question = state["sub_question"]
    
    # 调用搜索工具
    search_results = tavily_search(
        query=f"{state['topic']}: {sub_question}",
        max_results=5
    )
    
    # 提取网页内容
    contents = []
    for result in search_results:
        content = extract_web_content(result["url"])
        if content:
            contents.append(content)
    
    # 用LLM总结搜索结果
    summary = llm.invoke([
        SystemMessage(content="你是研究员。请从搜索结果中提取关键信息，每条不超过100字。"),
        HumanMessage(content=f"子问题：{sub_question}\n\n搜索结果：\n" + 
                         "\n---\n".join(contents))
    ]).content
    
    return {
        "research_findings": {sub_question: [summary]},
        "messages": [HumanMessage(content=f"完成研究：{sub_question}")]
    }

3.4 审查Agent——质量控制

# agents/reviewer.py
def review_orchestrator(state: ResearchState) -> list:
    """审查调度器：为每份研究发现分配审查员"""
    return [
        Send("reviewer_worker", {
            "sub_question": q,
            "findings": state["research_findings"].get(q, []),
        })
        for q in state["sub_questions"]
    ]

def reviewer_worker(state: dict) -> dict:
    """审查员工作节点：评估研究质量"""
    sub_question = state["sub_question"]
    findings = state["findings"]
    
    review_prompt = f"""你是研究审查员。请评估以下研究内容的质量。

    子问题：{sub_question}

    研究内容：
    {chr(10).join(findings)}

    请从以下维度评估：
    1. 准确性：信息是否准确可靠？
    2. 完整性：是否覆盖了关键方面？
    3. 时效性：信息是否最新？
    4. 建议：需要补充什么？

    输出格式：
    评分：X/10
    审查意见：...
    需要补充：..."""
    
    review = llm.invoke([
        HumanMessage(content=review_prompt)
    ]).content
    
    return {
        "review_results": {sub_question: review},
        "messages": [HumanMessage(content=f"完成审查：{sub_question}")]
    }

3.5 组装完整工作流

# main.py
from langgraph.graph import StateGraph, START, END

def build_research_workflow() -> StateGraph:
    """构建完整的研究助手工作流"""
    
    workflow = StateGraph(ResearchState)
    
    # ===== 添加节点 =====
    workflow.add_node("editor", editor_agent)
    workflow.add_node("research_orchestrator", research_orchestrator)
    workflow.add_node("researcher_worker", researcher_worker)
    workflow.add_node("review_orchestrator", review_orchestrator)
    workflow.add_node("reviewer_worker", reviewer_worker)
    workflow.add_node("reviser", reviser_agent)
    workflow.add_node("writer", writer_agent)
    workflow.add_node("publisher", publisher_agent)

    # ===== 添加边 =====
    workflow.add_edge(START, "editor")
    workflow.add_edge("editor", "research_orchestrator")
    workflow.add_edge("research_orchestrator", "researcher_worker")
    workflow.add_edge("researcher_worker", "review_orchestrator")
    workflow.add_edge("review_orchestrator", "reviewer_worker")
    workflow.add_edge("reviewer_worker", "reviser")
    workflow.add_edge("reviser", "writer")
    workflow.add_edge("writer", "publisher")
    workflow.add_edge("publisher", END)
    
    return workflow.compile()

# ===== 运行 =====
if __name__ == "__main__":
    workflow = build_research_workflow()
    
    result = workflow.invoke({
        "topic": "2026年AI编程工具市场分析",
        "language": "zh",
        "format": "markdown",
        "research_findings": {},
        "review_results": {},
        "revised_findings": {},
        "messages": [],
        "error_count": 0,
    })
    
    print(result["final_report"])

四、CrewAI实现方案对比

对于快速原型开发，CrewAI提供了更简洁的语法。以同样的研究任务为例：

# crewai_version.py
from crewai import Agent, Task, Crew, Process, LLM

# 初始化LLM
llm = LLM(model="ollama/qwen3:8b", base_url="http://localhost:11434")

# 定义角色
researcher = Agent(
    role="技术研究员",
    goal="深入调研技术话题，输出结构化报告",
    backstory="你是一个有10年经验的技术分析师",
    llm=llm,
    verbose=True
)

writer = Agent(
    role="技术作家",
    goal="将技术报告转化为通俗易懂的文章",
    backstory="你是知名科技博客的资深撰稿人",
    llm=llm,
    verbose=True
)

editor = Agent(
    role="内容编辑",
    goal="审校文章质量，确保准确性和可读性",
    backstory="你是严谨的技术出版编辑",
    llm=llm,
    verbose=True
)

# 定义任务
research_task = Task(
    agent=researcher,
    description="深入调研Python 3.14的新特性",
    expected_output="结构化的技术报告"
)

write_task = Task(
    agent=writer,
    description="将研究报告转化为通俗博客文章",
    expected_output="面向普通读者的博客文章"
)

edit_task = Task(
    agent=editor,
    description="审校文章并给出修改意见",
    expected_output="修改意见列表"
)

# 组建团队并执行
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, write_task, edit_task],
    process=Process.sequential  # 顺序执行
)

result = crew.kickoff(inputs={"topic": "Python 3.14 新特性"})
print(result.raw)

4.1 LangGraph vs CrewAI 选型决策

选型建议：

• 需要快速验证想法 → 选择CrewAI
• 需要生产级稳定性 → 选择LangGraph
• 需要复杂条件分支 → 选择LangGraph
• 团队成员背景多样 → 选择CrewAI

五、关键技术要点详解

5.1 并行处理的原理

LangGraph的 Send API是实现并行处理的关键。它的原理是：

# 串行处理（慢）
for q in sub_questions:
    result = research(q)  # 每个耗时10秒，3个问题=30秒

# 并行处理（快）
results = [Send("researcher_worker", {"sub_question": q}) 
           for q in sub_questions]  # 3个问题同时执行≈10秒

Send 不是一个普通的边，而是一个动态生成的任务列表。LangGraph会同时启动多个工作节点处理这些任务，结果通过 operator.add 合并到共享状态中。

5.2 错误处理与重试机制

def researcher_worker_with_retry(state: dict) -> dict:
    """带重试的研究员工作节点"""
    max_retries = 3
    
    for attempt in range(max_retries):
        try:
            result = do_research(state)
            return result
        except SearchAPIError as e:
            if attempt == max_retries - 1:
                # 最后一次重试失败，返回空结果而不是崩溃
                return {
                    "research_findings": {
                        state["sub_question"]: [f"搜索失败：{str(e)}"]
                    }
                }
            time.sleep(2 ** attempt)  # 指数退避

# 关键原则：Agent系统不应该因为单个节点的失败而整体崩溃
# 用优雅降级代替硬中断

5.3 性能优化清单

5.4 质量提升技巧

# 技巧1：Few-shot示例提升输出质量
editor_prompt = """
好的研究大纲示例：
主题："大模型在金融领域的应用"
大纲：
1. 大模型在风险评估中的应用
2. 大模型在智能投顾中的应用
3. 大模型在反欺诈中的应用
4. 监管政策与合规挑战

现在请为以下主题生成大纲：{topic}
"""

# 技巧2：Chain-of-Thought引导复杂推理
review_prompt = """
请按以下步骤审查：
步骤1：列出研究发现中的关键论点
步骤2：逐个验证论点的可信度
步骤3：检查是否有遗漏的重要方面
步骤4：给出总体评分和改进建议
"""

# 技巧3：结构化输出（使用Pydantic）
from pydantic import BaseModel

class ReviewResult(BaseModel):
    score: int          # 1-10分
    accuracy: str       # 准确性评估
    completeness: str   # 完整性评估
    suggestions: list   # 改进建议

六、生产环境部署方案

方案一：命令行工具（个人使用）

# 安装依赖
pip install langgraph langchain tavily-python python-dotenv

# 配置环境变量
echo 'OPENAI_API_KEY=sk-xxx' > .env
echo 'TAVILY_API_KEY=tvly-xxx' >> .env

# 运行
python main.py --topic "2026年AI编程工具市场分析" --lang zh --format markdown

方案二：Web界面（团队共享）

# webui.py
import streamlit as st
from main import build_research_workflow

st.title(" AI研究助手")
st.caption("输入研究主题，自动生成研究报告")

topic = st.text_input("研究主题", placeholder="例如：2026年AI编程工具市场分析")
language = st.selectbox("输出语言", ["中文", "English"])
report_format = st.selectbox("输出格式", ["Markdown", "HTML"])

if st.button("开始研究", type="primary"):
    with st.spinner("正在进行研究，请稍候..."):
        workflow = build_research_workflow()
        result = workflow.invoke({
            "topic": topic,
            "language": "zh" if language == "中文" else "en",
            "format": report_format.lower(),
            "research_findings": {},
            "review_results": {},
            "revised_findings": {},
            "messages": [],
            "error_count": 0,
        })
    st.markdown(result["final_report"])

# 运行：streamlit run webui.py

方案三：API服务（生产环境推荐）

# api.py
from fastapi import FastAPI
from pydantic import BaseModel
from main import build_research_workflow

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

class ResearchRequest(BaseModel):
    topic: str
    language: str = "zh"
    format: str = "markdown"

class ResearchResponse(BaseModel):
    report: str
    outline: list
    sub_questions: list

@app.post("/research", response_model=ResearchResponse)
async def research(request: ResearchRequest):
    result = await workflow.ainvoke({
        "topic": request.topic,
        "language": request.language,
        "format": request.format,
        "research_findings": {},
        "review_results": {},
        "revised_findings": {},
        "messages": [],
        "error_count": 0,
    })
    
    return ResearchResponse(
        report=result["final_report"],
        outline=result["outline"],
        sub_questions=result["sub_questions"],
    )

# 运行：uvicorn api:app --host 0.0.0.0 --port 8000

6.1 成本估算

七、深度研究Agent架构（企业级）

7.1 NVIDIA AI-Q蓝图架构

NVIDIA提出的企业级深度研究Agent采用模块化子Agent设计，具有严格的上下文隔离机制：

7.2 分层Agent结构

# 企业级Agent架构示例
class DeepResearchAgent:
    """
    深度研究Agent - 支持浅层和深层研究流程
    使用LangGraph实现，支持流式追踪
    """
    
    def create_deep_agent(self, config: AgentConfig):
        # 1. Planner Agent - 生成研究计划
        planner = self.create_planner(
            model=self.llm_provider.get(LLMRole.PLANNER),
            tools=[web_search, document_retrieval]
        )
        
        # 2. Researcher Agent - 执行研究
        researcher = self.create_researcher(
            model=self.llm_provider.get(LLMRole.RESEARCHER),
            tools=[tavily_search, content_extraction]
        )
        
        # 3. Synthesizer Agent - 整合结果
        synthesizer = self.create_synthesizer(
            model=self.llm_provider.get(LLMRole.SYNTHESIZER)
        )
        
        return StateGraph(DeepResearchState) \
            .add_node("planner", planner) \
            .add_node("researcher", researcher) \
            .add_node("synthesizer", synthesizer) \
            .add_edge(START, "planner") \
            .add_edge("planner", "researcher") \
            .add_edge("researcher", "synthesizer") \
            .add_edge("synthesizer", END) \
            .compile()

7.3 添加企业内部数据源

# 使用NeMo Agent Toolkit扩展数据源
# 1. 实现自定义工具
from nemo_agent_toolkit import tool

@tool
def internal_kb_search(query: str) -> list:
    """连接内部知识库"""
    # Azure AI Search集成
    return azure_search(query)

# 2. 在配置中引用工具
functions:
  - internal_kb_tool  # 自定义知识库
  - web_search_tool   # 网页搜索
  - document_reader    # 文档读取

# 3. 分配给Agent
researcher:
  tools:
    - internal_kb_search
    - web_search

八、实战项目：DeepSeek研究Agent

项目简介

这是一个基于DeepSeek LLM和Tavily搜索的自主研究代理，使用LangGraph构建。核心功能包括：

• 智能查询生成：将研究主题分解为多个聚焦的搜索查询
• 并行搜索：使用Tavily同时搜索多个查询
• 智能总结：LLM驱动的搜索结果摘要
• 自我反思：评估研究完整性（0-10分）
• 📄 报告生成：创建包含来源的完整Markdown报告

快速开始

# 1. 安装依赖
uv sync  # 或 pip install -r requirements.txt

# 2. 配置API密钥
# 编辑 .env 文件
DEEPSEEK_API_KEY=sk-your-key-here
TAVILY_API_KEY=tvly-your-key-here

# 3. 运行（Streamlit界面）
streamlit run app.py

# 4. 或使用命令行测试
python test_agent.py

程序化调用

from agent import create_research_agent, run_research

# 创建Agent
agent = create_research_agent()

# 运行研究
result = run_research("What are the latest AI developments?")

# 访问结果
print(result["report_draft"])
print(result["confidence_score"])
print(result["sources"])

配置参数