项目地址

Gemini-fullstack-langgraph-quickstart 是一个开源的全栈应用程序项目，应用能针对用户查询进行全面研究，动态生成搜索词，通过 Google Search API 搜索结果，并进行反思分析，迭代优化搜索，最终提供有引用支持的回答。架构图：

项目使用langgraph搭建主框架，需要对langgraph有一定的了解，可以看一下这篇文章
代码解析：

Generate_queries

这个函数的作用是使用大模型基于用户的问题生成搜索查询，其prompt的意思大概如下：

query_writer_instructions = '''
你的目标是生成复杂且多样的网络搜索查询。这些查询是为一个先进的自动化网络研究工具设计的，该工具能够分析复杂的结果、跟踪链接并综合信息。

指令：
- 尽量优先使用单一知识库查询语句，只有在原始问题涉及多个不同方面或元素且一个查询语句无法充分涵盖时，才添加另一个查询语句。
- 每个查询语句应聚焦于原始问题的一个特定方面，精准定位知识库中的相关知识条目。
- 生成的查询语句数量不要超过 {number_queries} 个。
- 查询语句应具有多样性，如果主题宽泛，应从不同角度生成多个查询语句。
- 避免生成多个相似的查询语句，一个精心设计的查询语句就足够了。
- 查询应确保收集到最新的信息。当前日期是 {current_date}

格式：
- 将你的回复格式化为一个 JSON 对象，必须包含以下三个完全匹配的键：
   - "rationale"：简要解释这些查询语句的相关性原因，说明它们如何能够从知识库中检索到有用的信息
   - "query"：知识库查询语句列表

示例：

主题：我目前的工作是上班族，同时还要带孩子，这对学习课程会有影响吗？
```json
{{
    "rationale": "为了帮助上班族家长了解在工作和带孩子的双重压力下学习课程的可行性和建议，这些查询语句聚焦于时间管理、家庭与工作的平衡、以及家长学员的学习经验等实际问题，能够精准检索到与用户处境高度相关的知识库内容。",
    "query": [
        "上班族如何平衡工作、带孩子与学习课程",
        "家长学员在学习课程时常见的挑战与应对方法",
        "时间管理技巧：上班族家长如何高效利用碎片时间学习",
        "有孩子的学员在课程学习中的真实经验分享",
        "家庭和工作压力对学习课程的影响及建议"
    ]
}}


历史对话上下文: {research_topic}
'''

代码部分：

def generate_query(state: OverallState, config: RunnableConfig) -> QueryGenerationState:
    """LangGraph node that generates a search queries based on the User's question.

    Uses Gemini 2.0 Flash to create an optimized search query for web research based on
    the User's question.

    Args:
        state: Current graph state containing the User's question
        config: Configuration for the runnable, including LLM provider settings

    Returns:
        Dictionary with state update, including search_query key containing the generated query
    """
    configurable = Configuration.from_runnable_config(config)

    # check for custom initial search query count
    if state.get("initial_search_query_count") is None:
        state["initial_search_query_count"] = configurable.number_of_initial_queries

    # init Gemini 2.0 Flash
    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)

    # Format the prompt
    current_date = get_current_date()
    formatted_prompt = query_writer_instructions.format(
        current_date=current_date,
        research_topic=get_research_topic(state["messages"]),
        number_queries=state["initial_search_query_count"],
    )
    # Generate the search queries
    result = structured_llm.invoke(formatted_prompt)
    return {"query_list": result.query}

简单来说就是定义了一个大模型，将prompt需要的参数填入后进行回复生成，根据用户的输入返回生成的问题列表给下一节点。此处的字典是定义的state中的一种，因为langgraph是通过state来更新获取对应的数据。

Web_research

这个函数的作用是进行网络查询并返回查询的结果
prompt的意思大概是：

web_searcher_instructions = """进行针对性的搜索，收集关于 "{research_topic}" 的最新、可信信息，并将其整合成一个可验证的文本作品。
指令：
查询应确保收集到最新的信息。当前日期是 {current_date}。
进行多次多样化的搜索，以收集全面的信息。
整理关键发现，同时仔细跟踪每条具体信息的来源。
输出内容应是基于搜索发现的精心撰写的总结或报告。
仅包含搜索结果中的信息，不要编造任何信息。
研究主题：
{research_topic}
"""

• research_topic这个state就是聊天记录，最新的输入也就是research_topic的最新一条，所以上面的prompt说的是历史上下文，这边搜索的prompt就是最新的输入

代码部分：

def web_research(state: WebSearchState, config: RunnableConfig) -> OverallState:
    """LangGraph node that performs web research using the native Google Search API tool.

    Executes a web search using the native Google Search API tool in combination with Gemini 2.0 Flash.

    Args:
        state: Current graph state containing the search query and research loop count
        config: Configuration for the runnable, including search API settings

    Returns:
        Dictionary with state update, including sources_gathered, research_loop_count, and web_research_results
    """
    # Configure
    configurable = Configuration.from_runnable_config(config)
    formatted_prompt = web_searcher_instructions.format(
        current_date=get_current_date(),
        research_topic=state["search_query"],
    )

    # Uses the google genai client as the langchain client doesn't return grounding metadata
    response = genai_client.models.generate_content(
        model=configurable.query_generator_model,
        contents=formatted_prompt,
        config={
            "tools": [{"google_search": {}}],
            "temperature": 0,
        },
    )
    # resolve the urls to short urls for saving tokens and time
    resolved_urls = resolve_urls(
        response.candidates[0].grounding_metadata.grounding_chunks, state["id"]
    )
    # Gets the citations and adds them to the generated text
    citations = get_citations(response, resolved_urls)
    modified_text = insert_citation_markers(response.text, citations)
    sources_gathered = [item for citation in citations for item in citation["segments"]]

    return {
        "sources_gathered": sources_gathered,
        "search_query": [state["search_query"]],
        "web_research_result": [modified_text],
    }

定义大模型并且配备了谷歌搜索的tools，并不单单是简单的url搜索引擎，所以可以传递进整个prompt，让大模型根据要求和搜索内容去进行搜索，并且根据放回的格式进行处理获取到对应的信息。

• 这里搜索的内容是：state[“search_query”]，而上面generate_query返回的内容是”query_list“，这是因为中间还有一个步骤，就是continue_to_web_research。

Continue_to_web_research

这个函数的作用是实现异步查询，因为生成了query_list，也就是多个问题，如果直接让web_research去查询的话会浪费时间，其实也可以将web_research写成异步，但是效果不如分发任务多路执行的结构好，也就是并发多查询结构。
代码部分：

def continue_to_web_research(state: QueryGenerationState):
    """LangGraph node that sends the search queries to the web research node.

    This is used to spawn n number of web research nodes, one for each search query.
    """
    return [
        Send("web_research", {"search_query": search_query, "id": int(idx)})
        for idx, search_query in enumerate(state["query_list"])
    ]

这里将state[“query_list”]进行提取了，也就解释了上面说的搜索的内容是：state[“search_query”]

Reflection

这个函数是来判断web_research的结果是否满足回答用户的问题，如果不满足还会生成问题继续一轮web_research，是整个架构的核心所在。如果满足的话就会根据搜索到的内容总结成回复。
prompt的意思大概是：

reflection_instructions = """你是一位分析关于 "{research_topic}" 的总结的专家研究助理。

指令：
- 识别知识空白或需要深入探索的领域，并生成后续查询。（一个或多个）
- 如果提供的总结足以回答用户的问题，则不要生成后续查询。
- 如果存在知识空白，生成一个后续查询，以帮助扩展你的理解。
- 重点在于未充分涵盖的技术细节、实施细节或新兴趋势。。

要求：
- 确保后续查询是自包含的，并包含知识库查询所需的必要背景信息。

输出格式：
- 将你的回复格式化为一个 JSON 对象，必须包含以下完全匹配的键：
   - "is_sufficient"：true 或 false
   - "knowledge_gap"：描述缺失的信息或需要澄清的内容
   - "follow_up_queries"：编写一个具体问题以解决这一空白

示例：
```json
{{
    "is_sufficient": true, // 或 false
    "knowledge_gap": "总结中缺少关于性能指标和基准测试的信息", // 如果 is_sufficient 为 true 则为空
    "follow_up_queries": ["用于评估 [特定技术] 的典型性能基准和指标是什么？"] // 如果 is_sufficient 为 true 则为空列表
}}


请仔细反思总结，识别知识空白并生成后续查询。然后，按照此 JSON 格式生成你的输出：

总结:
{summaries}
"""

代码部分：

def reflection(state: OverallState, config: RunnableConfig) -> ReflectionState:
    """LangGraph node that identifies knowledge gaps and generates potential follow-up queries.

    Analyzes the current summary to identify areas for further research and generates
    potential follow-up queries. Uses structured output to extract
    the follow-up query in JSON format.

    Args:
        state: Current graph state containing the running summary and research topic
        config: Configuration for the runnable, including LLM provider settings

    Returns:
        Dictionary with state update, including search_query key containing the generated follow-up query
    """
    configurable = Configuration.from_runnable_config(config)
    # Increment the research loop count and get the reasoning model
    state["research_loop_count"] = state.get("research_loop_count", 0) + 1
    reasoning_model = state.get("reasoning_model") or configurable.reasoning_model

    # Format the prompt
    current_date = get_current_date()
    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"]),
    )
    # init Reasoning Model
    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"]),
    }

将网络搜索的结果和用户的问题传递给大模型，让大模型判断并输出对应的结果。

Evaluate_research

这个函数的作用与continue_to_web_research的作用差不多，负责把reflection生成的follow_up_query传递给web_research实现并行多查询。其中还加了一个循环轮次的判断，防止一直在循环中。

def evaluate_research(
    state: ReflectionState,
    config: RunnableConfig,
) -> OverallState:
    """LangGraph routing function that determines the next step in the research flow.

    Controls the research loop by deciding whether to continue gathering information
    or to finalize the summary based on the configured maximum number of research loops.

    Args:
        state: Current graph state containing the research loop count
        config: Configuration for the runnable, including max_research_loops setting

    Returns:
        String literal indicating the next node to visit ("web_research" or "finalize_summary")
    """
    configurable = Configuration.from_runnable_config(config)
    max_research_loops = (
        state.get("max_research_loops")
        if state.get("max_research_loops") is not None
        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"])
        ]

写法与continue_to_web_research类似。

Finalize_answer

这个函数是做最终总结的
prompt的意思大概是：

answer_instructions = """
根据提供的摘要，为用户的问题生成高质量的回答。

指南：
- 当前日期为 {current_date}。
- 你是多步研究流程的最后一步，不要提及你处于最后一步。
- 你可以访问前几步收集的所有信息。
- 你可以访问用户的问题。
- 根据提供的摘要和用户的问题，生成高质量的回答。
- 你必须正确地在回答中包含摘要中的所有引用。

用户背景：
- {research_topic}

摘要：
{summaries}
"""

结合前几部分的输出进行最终的回答。

Graph

根据结构进行连接即可，此项目可以根据自己的需要改变部分组件，例如连接知识库作为补充等功能。

# Create our Agent Graph
builder = StateGraph(OverallState, config_schema=Configuration)

# Define the nodes we will cycle between
builder.add_node("generate_query", generate_query)
builder.add_node("web_research", web_research)
builder.add_node("reflection", reflection)
builder.add_node("finalize_answer", finalize_answer)

# Set the entrypoint as `generate_query`
# This means that this node is the first one called
builder.add_edge(START, "generate_query")
# Add conditional edge to continue with search queries in a parallel branch
builder.add_conditional_edges(
    "generate_query", continue_to_web_research, ["web_research"]
)
# Reflect on the web research
builder.add_edge("web_research", "reflection")
# Evaluate the research
builder.add_conditional_edges(
    "reflection", evaluate_research, ["web_research", "finalize_answer"]
)
# Finalize the answer
builder.add_edge("finalize_answer", END)

graph = builder.compile(name="pro-search-agent")