这时候,你就需要跳出“应用端拼装”,进入“代码层级的自定义”。本项目就是为此而生——它不仅让你看到一个完整的“AI 研究员”是怎么从零到一搭建起来的,还能让你随时插拔、扩展、魔改每一个环节。
它适合:
核心亮点:
在接下来的内容中,我会像一位带你实战的讲师,带你从全局到细节,逐步拆解这个项目的每一处关键实现。
在正式读代码前,我们先快速了解一下项目的整体结构和技术选型。
目录结构简述:
frontend/:React 前端,负责 UI 展示、与后端 API 通信backend/:Python 后端,核心为 LangGraph 代理src/agent/:代理主流程、工具、状态定义examples/cli_research.py:命令行调用代理的最小示例技术栈:
核心流程:
你可以把它想象成一个“AI 研究员”,自动帮你查资料、归纳、补充、引用,最后给你一份有理有据的答案。
在项目根目录下:
setup.batinstall
或分别:
setup.bat install-backend
setup.bat install-frontendsetup.batdev
setup.batcli-example"今天武汉的天气怎么样?"
你可以先随便问一个问题,感受一下“processing...”之后,AI 是如何给你一份带引用的研究报告的。
接下来,让我们像课堂实战一样,带着问题、带着好奇心,一步步走进项目的核心实现。
我们先从最简单的命令行入口开始。
文件:backend/examples/cli_research.py
if __name__ == "__main__": main()(第42行)main()(第5行)你可以打开这个文件,看到如下关键代码:
fromagent.graphimportgraph # 第3行
...
result = graph.invoke(state) # 第36行这里的graph.invoke(state),就是整个“AI 研究员”流程的起点。你输入一个问题,所有的自动研究、搜索、推理、引用,都是从这里开始的。
小贴士:如果你想快速体验后端的效果,可以直接在命令行运行:
pythonbackend/examples/cli_research.py"今天武汉的天气怎么样?"
接下来,我们顺着from agent.graph import graph跳到核心代理的定义。
文件:backend/src/agent/graph.py
graph = builder.compile(name="pro-search-agent")(第293行)你会看到一段类似“流程图”的代码:
builder.add_node("generate_query", generate_query) # 第272行
builder.add_node("web_research", web_research) # 第273行
builder.add_node("reflection", reflection) # 第274行
builder.add_node("finalize_answer", finalize_answer) # 第275行
...
builder.add_edge(START,"generate_query") # 第279行
builder.add_conditional_edges("generate_query", continue_to_web_research, ["web_research"])# 第281行
builder.add_edge("web_research","reflection") # 第285行
builder.add_conditional_edges("reflection", evaluate_research, ["web_research","finalize_answer"])# 第287行
builder.add_edge("finalize_answer", END) # 第291行林生点评:
现在我们带着“AI 研究员”的视角,逐个节点来看它们都做了什么。
generate_query节点def generate_query(state: OverallState, config: RunnableConfig) -> QueryGenerationState:(第44行)llm = ChatGoogleGenerativeAI(...)
structured_llm = llm.with_structured_output(SearchQueryList)
formatted_prompt = query_writer_instructions.format(...)
result = structured_llm.invoke(formatted_prompt)
return{"search_query": result.query}SearchQueryList是在 tools_and_schemas.py 里定义的结构,专门用来描述“搜索词列表+理由”。continue_to_web_research节点def continue_to_web_research(state: QueryGenerationState):(第84行)return[
Send("web_research", {"search_query": search_query,"id":int(idx)})
foridx, search_queryinenumerate(state["search_query"])
]web_research节点def web_research(state: WebSearchState, config: RunnableConfig) -> OverallState:(第95行)response = genai_client.models.generate_content(...)
resolved_urls = resolve_urls(...)
citations = get_citations(response, resolved_urls)
modified_text = insert_citation_markers(response.text, citations)
sources_gathered = [itemforcitationincitationsforitemincitation["segments"]]
return{
"sources_gathered": sources_gathered,
"search_query": [state["search_query"]],
"web_research_result": [modified_text],
}reflection节点def reflection(state: OverallState, config: RunnableConfig) -> ReflectionState:(第139行)formatted_prompt = reflection_instructions.format(...)
llm = ChatGoogleGenerativeAI(...)
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,
...
}Reflection结构体也是在 tools_and_schemas.py 里定义的,描述“是否足够、知识盲区、补充查询”。evaluate_research节点def evaluate_research(state: ReflectionState, config: RunnableConfig) -> OverallState:(第183行)ifstate["is_sufficient"]orstate["research_loop_count"] >= max_research_loops:
return"finalize_answer"
else:
return[
Send("web_research", {...})
foridx, follow_up_queryinenumerate(state["follow_up_queries"])
]finalize_answer节点def finalize_answer(state: OverallState, config: RunnableConfig):(第220行)formatted_prompt = answer_instructions.format(...)
llm = ChatGoogleGenerativeAI(...)
result = llm.invoke(formatted_prompt)
# 替换短链为原始链接
forsourceinstate["sources_gathered"]:
ifsource["short_url"]inresult.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,
}在读节点实现时,你可能会好奇:SearchQueryList、Reflection这些结构体是怎么定义的?
文件:backend/src/agent/tools_and_schemas.py
SearchQueryList(第5行):定义了搜索词列表和 rationale。Reflection(第14行):定义了反思节点的输出结构。你可以随时跳到这个文件,看看每个字段的含义和注释。
到这里,我们已经带着“AI 研究员”的视角,完整走了一遍从命令行输入到最终答案输出的全链路。
cli_research.py第36行graph.invoke(state)graph.py第269-291行(StateGraph 构建)graph.py第36-265行(每个节点的具体逻辑)tools_and_schemas.py(第5-24行)你可以跟着这些跳转,边看边实验,体会每一步的数据流转和设计巧思。
最后,给大家一些实战建议:
SearchQueryList、Reflection),可跳到tools_and_schemas.py查看定义。tools_and_schemas.py或utils.py。本文章可作为写作、讲解、二次开发的基础材料,欢迎补充和完善!希望你能像带着学员实战一样,真正掌握 LangGraph 应用开发的精髓。
原先的 google-gemini 该项目的地址如下,大家可以跳转去看:https://github.com/google-gemini/gemini-fullstack-langgraph-quickstart
我在这个项目的基础上进行了部分修改,有完整的中文使用教程,以及本地 Windows 部署脚本调整,方便大家直接运行和二次开发。如果对这块感兴趣,可以关注我的微信公众号,直接发消息“langgraph”,即可获得该项目的完整代码和使用教程。
| 欢迎光临 链载Ai (https://www.lianzai.com/) | Powered by Discuz! X3.5 |