引言
在人工智能快速发展的今天，如何构建一个能够进行深度研究、自主学习和迭代优化的AI系统成为了技术前沿的重要课题。Gemini开源的DeepResearch一周收获7.9k Star，Google的开源项目Gemini DeepResearch技术通过结合LangGraph框架和Gemini大语言模型，实现了一个具备自主研究能力的智能代理系统。本文将深入分析这一技术的核心原理和具体实现方式。
开源项目 Gemini Fullstack LangGraph Quickstart
https://github.com/google-gemini/gemini-fullstack-langgraph-quickstart
技术架构概览
Gemini DeepResearch采用了基于状态图（StateGraph）的多节点协作架构，通过LangGraph框架实现了一个完整的研究工作流。整个系统包含以下核心组件：
1. 状态管理系统
系统定义了多种状态类型来管理不同阶段的数据流：
class OverallState(TypedDict):
messages: Annotated[list, add\_messages]
search\_query: Annotated[list, operator.add]
web\_research\_result: Annotated[list, operator.add]
sources\_gathered: Annotated[list, operator.add]
initial\_search\_query\_count: int
max\_research\_loops: int
research\_loop\_count: int
reasoning\_model: str
这种设计允许系统在不同节点间传递和累积信息，确保研究过程的连续性和完整性。
2. 核心工作流程
整个研究流程分为五个关键阶段：
阶段一：查询生成（Query Generation）
系统首先分析用户输入，使用Gemini 2.0 Flash模型生成多个优化的搜索查询：
def generate\_query(state: OverallState, config: RunnableConfig) -> QueryGenerationState:
llm = ChatGoogleGenerativeAI(
model=configurable.query\_generator\_model,
temperature=1.0,
max\_retries=2,
api\_key=os.getenv("GEMINI\_API\_KEY"),
)
structured\_llm = llm.with\_structured\_output(SearchQueryList)
formatted\_prompt = query\_writer\_instructions.format(
current\_date=current\_date,
research\_topic=get\_research\_topic(state["messages"]),
number\_queries=state["initial\_search\_query\_count"],
)
result = structured\_llm.invoke(formatted\_prompt)
return {"query\_list": result.query}
关键特点：
- 多样化查询生成 ：系统会生成多个不同角度的搜索查询，确保信息收集的全面性
- 结构化输出 ：使用Pydantic模型确保输出格式的一致性
- 时效性考虑 ：查询中包含当前日期信息，确保获取最新数据
阶段二：并行网络研究（Parallel Web Research）
系统使用LangGraph的Send机制实现并行搜索：
def continue\_to\_web\_research(state: QueryGenerationState):
return [
Send("web\_research", {"search\_query": search\_query, "id": int(idx)})
for idx, search\_query in enumerate(state["query\_list"])
]
每个搜索查询都会启动一个独立的web_research节点，实现真正的并行处理。
阶段三：智能网络搜索（Web Research）
这是系统的核心功能之一，集成了Google Search API和Gemini模型：
def web\_research(state: WebSearchState, config: RunnableConfig) -> OverallState:
response = genai\_client.models.generate\_content(
model=configurable.query\_generator\_model,
contents=formatted\_prompt,
config={
"tools": [{"google\_search": {}}],
"temperature": 0,
},
)
# 处理搜索结果和引用
resolved\_urls = resolve\_urls(
response.candidates[0].grounding\_metadata.grounding\_chunks, state["id"]
)
citations = get\_citations(response, resolved\_urls)
modified\_text = insert\_citation\_markers(response.text, citations)
return {
"sources\_gathered": sources\_gathered,
"search\_query": [state["search\_query"]],
"web\_research\_result": [modified\_text],
}
技术亮点：
- 原生Google Search集成 ：直接使用Google的搜索API获取实时信息
- 自动引用处理 ：系统自动提取和格式化引用信息
- URL优化 ：将长URL转换为短链接以节省token消耗
阶段四：反思与知识缺口分析（Reflection）
这是DeepResearch的核心创新之一，系统会自动评估已收集信息的充分性：
def reflection(state: OverallState, config: RunnableConfig) -> ReflectionState:
formatted\_prompt = reflection\_instructions.format(
current\_date=current\_date,
research\_topic=get\_research\_topic(state["messages"]),
summaries="\n\n---\n\n".join(state["web\_research\_result"]),
)
llm = ChatGoogleGenerativeAI(
model=reasoning\_model,
temperature=1.0,
max\_retries=2,
api\_key=os.getenv("GEMINI\_API\_KEY"),
)
result = llm.with\_structured\_output(Reflection).invoke(formatted\_prompt)
return {
"is\_sufficient": result.is\_sufficient,
"knowledge\_gap": result.knowledge\_gap,
"follow\_up\_queries": result.follow\_up\_queries,
"research\_loop\_count": state["research\_loop\_count"],
"number\_of\_ran\_queries": len(state["search\_query"]),
}
反思机制的核心功能：
- 知识缺口识别 ：自动分析当前信息是否足够回答用户问题
- 后续查询生成 ：针对发现的知识缺口生成新的搜索查询
- 迭代控制 ：决定是否需要进行下一轮研究
阶段五：答案综合（Answer Finalization）
最终阶段将所有收集的信息综合成完整的答案：
def finalize\_answer(state: OverallState, config: RunnableConfig):
formatted\_prompt = answer\_instructions.format(
current\_date=current\_date,
research\_topic=get\_research\_topic(state["messages"]),
summaries="\n---\n\n".join(state["web\_research\_result"]),
)
llm = ChatGoogleGenerativeAI(
model=reasoning\_model,
temperature=0,
max\_retries=2,
api\_key=os.getenv("GEMINI\_API\_KEY"),
)
result = llm.invoke(formatted\_prompt)
# 处理引用链接
unique\_sources = []
for source in state["sources\_gathered"]:
if source["short\_url"] in result.content:
result.content = result.content.replace(
source["short\_url"], source["value"]
)
unique\_sources.append(source)
return {
"messages": [AIMessage(content=result.content)],
"sources\_gathered": unique\_sources,
}
技术创新点
1. 自适应研究循环
系统通过evaluate\_research函数实现智能的研究循环控制：
def evaluate\_research(state: ReflectionState, config: RunnableConfig) -> OverallState:
configurable = Configuration.from\_runnable\_config(config)
max\_research\_loops = (
state.get("max\_research\_loops")
if state.get("max\_research\_loops") isnotNone
else configurable.max\_research\_loops
)
if state["is\_sufficient"] or state["research\_loop\_count"] >= max\_research\_loops:
return"finalize\_answer"
else:
return [
Send(
"web\_research",
{
"search\_query": follow\_up\_query,
"id": state["number\_of\_ran\_queries"] + int(idx),
},
)
for idx, follow\_up\_query in enumerate(state["follow\_up\_queries"])
]
这种设计确保了系统既能深入研究复杂问题，又能避免无限循环。
2. 智能引用管理
系统实现了完整的引用管理机制：
- URL解析 ：将复杂的搜索结果URL转换为简洁的引用格式
- 引用插入 ：自动在文本中插入引用标记
- 去重处理 ：确保最终答案中只包含实际使用的引用源
3. 多模型协作
系统巧妙地使用不同的Gemini模型处理不同任务：
-
Gemini 2.0 Flash ：用于查询生成和网络搜索，速度快
-
Gemini 2.5 Flash ：用于反思分析，平衡速度和质量
-
Gemini 2.5 Pro ：用于最终答案生成，确保高质量输出
系统架构图
用户输入 → 查询生成 → 并行网络搜索 → 反思分析 → 评估决策
↓ ↓ ↓ ↓
多个搜索查询 收集网络信息 知识缺口分析 继续研究/结束
↓ ↓
生成后续查询 答案综合
↓ ↓
返回搜索 最终答案
添加微信，备注” LLM “进入大模型技术交流群
如果你觉得这篇文章对你有帮助，别忘了点个赞、送个喜欢
/ 作者：致Great
/ 作者：欢迎转载，标注来源即可