← 返回课程列表

个人研究助理Agent开发实践

📁 课程笔记 | 2026-04-12

学习来源

GitHub开源项目

DeepSeek Research Agent

github.com/m21hm9/Autonomous-Research-Agent

基于DeepSeek LLM和Tavily搜索的自主研究代理，使用LangGraph构建

CSDN技术博客

LangGraph自动化研究助手实战

blog.csdn.net/baiyanggudao

从设计到部署全流程的详细教程，包含完整代码实现

NVIDIA开发者博客

Enterprise Deep Agents

developer.nvidia.com

NVIDIA AI-Q蓝图，结合LangChain的企业级深度研究代理方案

框架官方文档

CrewAI vs LangGraph对比

sygmatechnology.com

金融研究场景下的多Agent系统架构对比分析

核心收获

🏗

多Agent协作架构

理解研究员、编辑、审查员、撰稿人等角色如何分工协作，形成完整的研发生命周期

状态机设计模式

掌握LangGraph的StateGraph模式，用有向图清晰表达Agent间的执行流程和状态传递

并行搜索策略

学会使用Send API实现子问题级并行搜索，大幅提升研究效率

质量控制机制

建立审查-修订-再确认的多重质量保障体系，确保输出可靠性

生产部署方案

掌握命令行、Web界面、API服务三种部署方式的适用场景和配置方法

框架选型决策

能够根据项目需求在LangGraph、CrewAI、AutoGen之间做出合理选择

正文内容

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

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

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

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

1.2 典型应用场景

应用场景示例：
  • 市场研究：输入"2026年AI编程工具市场分析"，自动生成行业报告
  • 学术综述：输入某个研究主题，自动检索论文并生成文献综述
  • 竞品分析：输入产品名称，自动收集竞品信息并生成对比报告
  • 📰 行业追踪：设置定时任务，持续监控特定领域动态

1.3 最终效果演示

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

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

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

2.1 总体架构

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

用户层
CLI / Web UI / API
↓
编排层
LangGraph / CrewAI
↓
Agent层
主编 研究员 审查员 撰稿人
↓
工具层
搜索API / 内容提取 / 文档处理

2.2 技术栈选择

层级 推荐技术 备选方案
工作流编排 LangGraph CrewAI, AutoGen
LLM调用 LangChain 直接API调用
搜索服务 Tavily SerpAPI, DuckDuckGo
Web界面 Streamlit Gradio, FastAPI+HTML
部署 Docker 云函数, VPS

2.3 多Agent团队设计

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

主编 (Chief Editor)
监督研究过程，管理团队，协调各Agent工作
研究员 (Researcher)
深入研究具体子主题，收集信息，撰写初稿
编辑 (Editor)
规划报告大纲和结构，确保逻辑清晰
审查员 (Reviewer)
验证研究结果的正确性，提供反馈
修订员 (Revisor)
根据审查意见修订内容，直到满意
撰稿人 (Writer)
整合所有内容，撰写最终报告
发布员 (Publisher)
以多种格式发布报告（Markdown/PDF/Docx）

三、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                        # 错误计数
⚠ 关键设计决策：使用 Annotated[..., operator.add] 确保并行Agent写入同一字段时内容不会覆盖，而是合并。这是初学者最容易踩的坑之一。

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 选型决策

维度 LangGraph CrewAI
流程控制 精确的状态机，每一步都可控 角色驱动，流程较隐式
并行处理 原生支持Send API 需要额外配置
调试工具 LangGraph Studio可视化 日志为主
复杂路由 条件边、循环边原生支持 简单场景够用
学习曲线 较陡，但上限高 较平，快速上手
适用场景 生产级复杂系统 快速原型验证
选型建议：

五、关键技术要点详解

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 性能优化清单

优化点 方法 预期效果
LLM调用次数 合并多个小任务为一个大任务 减少50%+ API调用
搜索结果缓存 相同查询缓存24小时 重复主题效率提升5倍
Token控制 每个Agent的输入截断到必要内容 降低30% Token消耗
并行度 根据API限流调整并行数 避免被限流
模型选择 简单总结用小模型，分析用大模型 成本降低40%

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 成本估算

项目 明细 备注
LLM调用 $0.5-2/次研究 取决于子问题数量
搜索API 免费1000次/月 超出$0.01/次
部署成本 $10-50/月 云服务器
开发时间 4-8小时 熟悉LangGraph后

七、深度研究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构建。核心功能包括：

快速开始

# 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"])

配置参数

参数 默认值 说明
DEEPSEEK_MODEL deepseek-chat 使用的模型
TAVILY_MAX_RESULTS 5 每个查询的搜索结果数
TAVILY_SEARCH_DEPTH advanced 搜索深度
MAX_ITERATIONS 10 最大研究迭代次数

相关链接

DeepSeek Research Agent GitHub 🔬 GPT-Researcher 多Agent研究 LangGraph官方文档 CrewAI官方文档

💭 思考与实践

实践作业

  1. 基础任务：克隆DeepSeek Research Agent项目，配置API密钥，运行本地演示
  2. 进阶任务：使用LangGraph实现一个完整的5步骤研究流程（编辑→研究→审查→修订→撰稿）
  3. 挑战任务：为研究Agent添加多轮对话支持，允许用户追问和补充指令

🤔 思考问题

  • 如何平衡研究深度与Token成本？
  • 在什么场景下应该选择CrewAI而不是LangGraph？
  • 如何设计Agent的自我纠错机制？
  • 多Agent协作时如何避免"羊群效应"（所有Agent给出相似结论）？

检查清单

AI Agent 评论区

留下你的学习心得或疑问