传统调研方式的伪代码示意

def traditional_research(query):
     步骤1: 手动 → 打开浏览器 → 输入关键词 → 筛选结果
    search_results = manual_search(query)
    
     步骤2: 逐个打开网页 → 人工阅读 → 手动摘录
    notes = []
    for url in search_results:
        page = open_in_browser(url)
        relevant_content = manual_read_and_extract(page)
        notes.append(relevant_content)
    
     步骤3: 人工整理汇总 → 编写报告
    report = manual_summarize(notes)
    
     痛点：大量重复劳动，信息可能遗漏，缺乏系统性整合
    return report

 RAG标准实现示意（基于LangChain伪代码）
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.llms import ChatOpenAI

 步骤1: 索引（预处理阶段）
documents = load_documents("knowledge_base/")         加载文档
chunks = split_into_chunks(documents, chunk_size=512)  文本分块
embeddings = OpenAIEmbeddings().embed_documents(chunks)  向量化
vector_db = Chroma.from_documents(chunks, embeddings)  存入向量库

 步骤2: 检索（查询阶段）
query = "Spring Boot 3.x 有哪些新特性？"
query_embedding = OpenAIEmbeddings().embed_query(query)
retrieved_chunks = vector_db.similarity_search(query_embedding, k=5)

 步骤3: 增强（构造上下文）
context = "\n\n".join([chunk.page_content for chunk in retrieved_chunks])
enhanced_prompt = f"""
基于以下参考资料回答用户问题：
参考资料：
{context}
用户问题：{query}
请给出准确、有依据的回答：
"""

 步骤4: 生成
llm = ChatOpenAI(model="gpt-4")
response = llm.invoke(enhanced_prompt)

 ReAct循环示意
Thought: 用户问的是Spring Boot 3.x，我需要先检索相关文档
Action: search("Spring Boot 3.x 新特性")
Observation: 检索到3篇相关文章：A、B、C

Thought: 这些文章的信息还不够完整，我需要再一下官方文档
Action: search("Spring Boot 3.x official documentation")
Observation: 找到官方发布说明

Thought: 信息收集完毕，可以开始整理回答了
Action: generate_answer()

 ReAct循环的简化实现（基于LangGraph）
from langgraph.graph import StateGraph, END

def agent_loop(state):
    while not state["is_complete"]:
         Thought阶段：Agent决定下一步行动
        thought = llm.invoke(
            f"当前状态：{state['context']}\n用户问题：{state['query']}\n下一步该做什么？"
        )
        
         Action阶段：执行决定的操作
        if "need_search" in thought:
            new_results = search(state["next_query"])
            state["context"].extend(new_results)
        elif "need_generate" in thought:
            state["answer"] = llm.invoke(f"基于{state['context']}生成回答")
            state["is_complete"] = True
    return state

标准RAG（单一检索+生成）
   ↓
Agentic RAG（多轮检索+推理+自主决策）
   ↓
Deep Research Agent（多智能体协作+战略规划+长时记忆）

 环境准备（参考2026年主流技术栈）
 pip install langgraph langchain chromadb openai crewai

from langgraph.graph import StateGraph, END
from langchain.tools import Tool
from langchain_openai import ChatOpenAI
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

 1. 定义状态结构
class ResearchState(TypedDict):
    query: str
    findings: List[str]
    current_step: str
    report: str
    iteration_count: int

 2. 定义智能体可使用的工具
search_tool = Tool(
    name="web_search",
    func=lambda q: search_web(q),   封装API
    description="互联网获取最新信息"
)

retrieve_tool = Tool(
    name="knowledge_retrieve", 
    func=lambda q: vector_db.similarity_search(q, k=3),
    description="从内部知识库检索"
)

 3. 构建Agent图工作流
def create_research_agent():
    workflow = StateGraph(ResearchState)
    
     定义节点
    workflow.add_node("planner", planner_node)       规划节点
    workflow.add_node("searcher", searcher_node)     节点
    workflow.add_node("analyzer", analyzer_node)     分析节点
    workflow.add_node("writer", writer_node)         写作节点
    
     定义边（控制流）
    workflow.set_entry_point("planner")
    workflow.add_edge("planner", "searcher")
    workflow.add_conditional_edges("searcher", should_continue)
    workflow.add_edge("analyzer", "writer")
    workflow.add_edge("writer", END)
    
    return workflow.compile()

def planner_node(state):
    """规划节点：分解任务，制定检索策略"""
    plan = llm.invoke(f"分析问题'{state['query']}'，给出检索关键词和优先级")
    state["current_step"] = "searching"
    state["findings"] = []
    return state

def searcher_node(state):
    """节点：执行多轮检索"""
     第一轮：宽泛检索
    broad_results = search_tool.run(state["query"])
    state["findings"].extend(broad_results)
    
     第二轮：基于初步发现进行定向检索
    follow_up_queries = extract_keywords(broad_results)
    for q in follow_up_queries[:3]:   最多3个跟进查询
        deep_results = search_tool.run(q)
        state["findings"].extend(deep_results)
    
    state["iteration_count"] += 1
    return state

def analyzer_node(state):
    """分析节点：整合检索结果，提取关键洞察"""
    analysis = llm.invoke(f"""
    分析以下调研发现，提取3-5个核心要点：
    {state["findings"]}
    按重要性排序，并说明每个要点的依据。
    """)
    state["analysis"] = analysis
    state["current_step"] = "writing"
    return state

def writer_node(state):
    """写作节点：生成结构化报告"""
    report = llm.invoke(f"""
    基于以下调研发现和分析结论，生成一份调研报告：
    
    用户问题：{state["query"]}
    核心发现：{state["analysis"]}
    
    报告应包含：摘要、详细分析、总结建议、参考资料。
    请确保格式规范、逻辑清晰。
    """)
    state["report"] = report
    return state