目前我们对GraphRAG的四种查询模式（Basic、Local、Drift和Global）的设计和实现原理，虽然它们擅长从全局着眼来回答问题，但是很多情况下也不能一次就能提供你想要的答案。我们可以采用Agent常用的交谈方式，通过继续追问来获取更多信息。几乎所有这样的应用都会根据当前上下文提供一组候选的问题供你选择，GraphRAG为此专门设计一个名为Question Generation（以下简称QGen）的模块。

1. 将QGen应用到Chat Loop中

为了让读者对QGen针对上下文生成问题的功能有一个感官的认识，我们综合之前介绍的内容开发一款简单能够通过对话循环保持持续交流的应用。我们依然使用本系列一直使用的应用场景，使用如下这个四行的.csv提供上下文，并通过执行标准管道建立所需的索引表和基于lancedb的向量数据库。

id,title,text
doc1,选址,2024年3月，李明决定在杭州开一家咖啡店。他考察了三个地点：学校附近、写字楼商圈、老城区街角。学校附近人流量大，但租金每月高达两万元；写字楼商圈工作日客流稳定，周末冷清；老城区租金低，但年轻客群少。最终李明选择了写字楼商圈，月租一万五千元，签约两年。
doc2,产品,李明将核心产品定为拿铁和美式。拿铁定价二十八元，适合大众口味；美式定价二十二元，面向咖啡爱好者。2024年5月，他推出季节限定款桂花拿铁，首月销量占总额的三成。店长王芳建议增加会员折扣，提升复购率。
doc3,营销,开业初期，李明通过小红书和抖音引流，首周新增粉丝五千人。2024年6月，他建立会员体系，充值五百元赠送五十元，三个月内会员数达到八百人。李明还与周边书店、健身房合作互推，获客成本从每人三十元降到十二元。
doc4,供应链,咖啡豆由云南普洱的供应商张伟每月供货一次，单价每公斤一百二十元。牛奶由本地乳企每日配送。2024年7月，因雨季运输延误，门店断货两天，损失约四千元。李明随后增加了第二家供应商作为备份。

如下所示的演示程序由build_engines和chat_loop两个函数组成，前者用来创建作为搜索引擎LocalSearch对象和用来生成问题的LocalQuestionGen对象，后者提供了一个对话循环。在build_engines函数中，我们读取五个索引表（text_units、entities、relationships、communities和community_reports）生成对应的索引对象列表。加载配置文件settings.yaml创建了RagConfig配置，并在此基础上创建了作为Completion模型和嵌入模型的LLMCompletion和LLMEmbedding对象，以及针对实体描述向量存储的VectorStore。我们综合这些对象将作为Local search搜索引擎的LocalSearch创建出来。

import asyncio
import pandas as pd
from graphrag.config.load_config import load_config
from graphrag.query.indexer_adapters import (
    read_indexer_entities,
    read_indexer_relationships,
    read_indexer_reports,
    read_indexer_text_units,
)
from graphrag.query.question_gen.local_gen import LocalQuestionGen
from graphrag.query.structured_search.local_search.mixed_context import LocalSearchMixedContext
from graphrag.query.structured_search.local_search.search import LocalSearch
from graphrag_llm.completion.completion_factory import create_completion
from graphrag_llm.embedding.embedding_factory import create_embedding
from graphrag.utils.api import get_embedding_store
from graphrag.query.context_builder.conversation_history import ConversationHistory, ConversationRole


def build_engines():
    entity_df = pd.read_parquet(f"./output/entities.parquet")
    community_df = pd.read_parquet(f"./output/communities.parquet")
    report_df = pd.read_parquet(f"./output/community_reports.parquet")
    relationship_df = pd.read_parquet(f"./output/relationships.parquet")
    text_unit_df = pd.read_parquet(f"./output/text_units.parquet")

    entities = read_indexer_entities(entity_df, community_df, community_level=2)
    reports = read_indexer_reports(report_df, community_df, community_level=2)
    relationships = read_indexer_relationships(relationship_df)
    text_units = read_indexer_text_units(text_unit_df)

    config = load_config(root_dir="./")
    entity_text_embeddings = get_embedding_store(
        config=config.vector_store,
        embedding_name="entity_description",
    )
    chat_model = create_completion(
        model_config=config.completion_models[config.local_search.completion_model_id]
    )
    embedding_model = create_embedding(
        model_config=config.embedding_models[config.local_search.embedding_model_id]
    )

    context_builder = LocalSearchMixedContext(
        community_reports=reports,
        text_units=text_units,
        entities=entities,
        relationships=relationships,
        entity_text_embeddings=entity_text_embeddings,
        text_embedder=embedding_model,
    )
    local_search = LocalSearch(
        model=chat_model,
        context_builder=context_builder,
        response_type="Single Sentence",
    )
    question_gen = LocalQuestionGen(
        model=chat_model,
        context_builder=context_builder,
        model_params={"temperature": 0.5},
    )
    return local_search, question_gen

async def chat_loop():
    local_search, question_gen = build_engines()
    conversation_history = ConversationHistory()
    follow_ups = []

    print("GraphRAG 对话模式已启动（Local Search + Question Generation）")
    print("输入问题开始对话；输入编号选择引导问题；输入 exit 退出。")

    while True:
        if follow_ups:
            print("\n可以继续追问：")
            for i, q in enumerate(follow_ups, 1):
                print(f"  [{i}] {q}")
            print("  （也可直接输入新问题，或输入exit退出）")

        user_input = input("\nUser：").strip()
        if not user_input:
            continue
        if user_input.lower() in ("exit", "quit", "q"):
            break
        if user_input.isdigit() and follow_ups:
            idx = int(user_input) - 1
            if 0 <= idx < len(follow_ups):
                query = follow_ups[idx]
            else:
                print(f"[提示]编号超出范围（1–{len(follow_ups)}）。")
                continue
        else:
            query = user_input

        print("Assistant:", end="")
        chunks = []
        async for chunk in local_search.stream_search(query=query,conversation_history=conversation_history):
            chunks.append(chunk)
            print(chunk, end="")       
        conversation_history.add_turn(ConversationRole.USER, query)
        conversation_history.add_turn(ConversationRole.ASSISTANT, "".join(chunks))        

        result = await question_gen.agenerate(
            question_history= [item.content for item in  conversation_history.turns if item.role == ConversationRole.USER],
            context_data=None,
            question_count=3)
        follow_ups = result.response

asyncio.run(chat_loop())    

build_engines函数返回二元组提供的另一个LocalQuestionGen对象是这篇文章重点介绍的内容。顾名思义，它采用Local search模式来提供后续追问的问题。我们在创建LocalQuestionGen对象是提供了作为Completion模型的LLMCompletion对象和用来构建上下文的LocalSearchMixedContext对象。在chat_loop函数中，我们利用ConversationHistory对象那个来维护对话历史，并将每次迭代生成的追问问题存储在字符串列表follow_ups中。

第一轮对话循环会让用户提出初始问题，并利用build_engines函数提供的LocalSearch对象来回答问题。我们将用户问题和答案添加到对话历史中，并利用build_engines函数提供的LocalQuestionGen对象提供三个推荐的追问问题并输出，用户下一轮可以选择这些问题中的一个，也可以输入新的问题。值得一提的是，ConversationHistory对象维护的是完整的历史，但是LocalQuestionGen的agenerate方法的question_history是历史问题，即用户提出的问题，所以调用此方法是需要提取User消息的内容。以下是运行起来的对答效果：

GraphRAG 对话模式已启动（Local Search + Question Generation）
输入问题开始对话；输入编号选择引导问题；输入 exit 退出。

User：李明在经营咖啡店过程中都经历了哪些挑战？
Assistant:李明经营咖啡店过程中主要经历了选址权衡（在学校附近、老城区街角和写字楼商圈之间比较租金、客流和客群，最终选择写字楼商圈）、开业初期获客压力（通过小红书和抖音引流以快速积累粉丝）、本地获客成本控制（与周边书店和健身房合作互推以降低获客成本）、供应链稳定性问题（因雨季运输延误曾断货两天并造成损失，随后增加第二家供应商作为备份），以及通过会员体系和会员折扣提升复购率的经营挑战与优化需求 [Data: Sources (0, 2, 3); Reports (5); Entities (2, 3, 4, 17, 18, +more); Relationships (2, 3, 17, 18, 19, +more)]
可以继续追问：
  [1] - 李明在选址时为什么最终放弃学校附近和老城区街角，转而选择写字楼商圈？
  [2] - 李明在开业初期是如何通过小红书和抖音快速获取第一批顾客的？
  [3] - 李明后来又遇到了哪些供应和运营上的问题，并采取了什么措施来降低风险和提升复购率？
  （也可直接输入新问题，或输入exit退出）

User：2
Assistant:李明在开业初期通过小红书和抖音进行引流获客，并在首周新增粉丝五千人，随后又通过会员体系和周边书店、健身房的合作互推进一步提升获客效率与顾客转化 [Data: Sources (2); Entities (17, 18, 22, 21); Relationships (17, 18, 23, 19, 20)]
可以继续追问：
  [1] - 李明在选择咖啡店选址时，学校附近、写字楼商圈和老城区街角分别有哪些优劣，最终为什么选择了写字楼商圈？
  [2] - 李明在经营咖啡店过程中，会员体系和会员折扣是如何帮助提升复购率的，王芳在其中提出了什么建议？
  [3] - 李明在供应链管理上遇到了哪些问题，后来为什么要增加第二家供应商作为备份？
  （也可直接输入新问题，或输入exit退出）

User：3
Assistant:李明在供应链管理上主要遇到过雨季运输延误导致门店断货两天、损失约四千元的问题，因此后来增加第二家供应商作为备份，是为了降低单一供应风险、减少供货中断并提升门店运营稳定性 [Data: Sources (3); Reports (5); Relationships (25, 32); Entities (27)]
可以继续追问：
  [1] - 李明在选址过程中是如何比较学校附近、写字楼商圈和老城区街角的租金、客流和客群差异，并最终决定选择写字楼商圈的？
  [2] - 李明在开业初期是怎样通过小红书和抖音快速引流、并结合会员体系和周边书店、健身房合作来降低获客成本的？
  [3] - 李明在推出拿铁、美式和桂花拿铁之后，店长王芳提出了什么关于会员折扣和复购率的建议，李明后来是如何回应的？
  （也可直接输入新问题，或输入exit退出）

User：

2. 针对QGen的抽象

上面演示实例提供的LocalQuestionGen是BaseQuestionGen这个抽象类的子类，针对问题的生成体现在generate和agenerate两个抽象方法中。按照一般的API设计，generate和agenerate方法应该会定义成同步和异步方法，但是这里确定成了具有一致签名的异步方法，这可能是历史演进遗留导致的问题（可能generate原来是同步方法，但是升级之后涉及到异步调用，比如调用LLM）。

class BaseQuestionGen(ABC):
    def __init__(
        self,
        model: "LLMCompletion",
        context_builder: GlobalContextBuilder | LocalContextBuilder,
        tokenizer: Tokenizer | None = None,
        model_params: dict[str, Any] | None = None,
        context_builder_params: dict[str, Any] | None = None,
    ):
        self.model = model
        self.context_builder = context_builder
        self.tokenizer = tokenizer or model.tokenizer
        self.model_params = model_params or {}
        self.context_builder_params = context_builder_params or {}

    @abstractmethod
    async def generate(
        self,
        question_history: list[str],
        context_data: str | None,
        question_count: int,
        **kwargs,
    ) -> QuestionResult

    @abstractmethod
    async def agenerate(
        self,
        question_history: list[str],
        context_data: str | None,
        question_count: int,
        **kwargs,
    ) -> QuestionResult

构造函数参数说明如下：

• model：LLMCompletion 实例，用于生成问题。
• context_builder：GlobalContextBuilder 或 LocalContextBuilder 实例，负责为问题生成提供上下文数据。
• tokenizer：分词器，用于统计token数；未传时使用 model.tokenizer兜底。
• model_params：传给模型的额外参数字典；未传时为空字典。
• context_builder_params：传给上下文构建器的额外参数字典；未传时为空字典。

generate/agenerate方法的参数说明如下：

• question_history：历史问题列表（不是对话实例），用于在生成新问题时提供上下文与避免重复。
• context_data：可选的上下文数据字符串，作为问题生成的依据；为 None 时由实现决定是否自行构建上下文。
• question_count：期望生成的问题数量。
• kwargs：额外关键字参数，供具体实现扩展使用。

generate/agenerate方法返回一个QuestionResult对象，生成的问题包含在response字段返回的列表中。QuestionResult对象还提供了上下文数据和耗时、LLM调用数据以及提示词长度的统计。

@dataclass
class QuestionResult:
    response: list[str]
    context_data: str | dict[str, Any]
    completion_time: float
    llm_calls: int
    prompt_tokens: int

3. 针对Local search的实现

虽然BaseQuestionGen构造函数的context_builder参数的类型被标注成GlobalContextBuilder | LocalContextBuilder，但是GraphRAG目前只提供了LocalQuestionGen这个针对Local search的实现。

class LocalQuestionGen(BaseQuestionGen):
    def __init__(
        self,
        model: "LLMCompletion",
        context_builder: LocalContextBuilder,
        tokenizer: Tokenizer | None = None,
        system_prompt: str = QUESTION_SYSTEM_PROMPT,
        callbacks: list[BaseLLMCallback] | None = None,
        model_params: dict[str, Any] | None = None,
        context_builder_params: dict[str, Any] | None = None,
    ):
        super().__init__(
            model=model,
            context_builder=context_builder,
            tokenizer=tokenizer,
            model_params=model_params,
            context_builder_params=context_builder_params,
        )
        self.system_prompt = system_prompt
        self.callbacks = callbacks or []

除了将context_builder参数限制为LocalContextBuilder类型外，LocalQuestionGen的构造函数还可以提供系统提示词和一组BaseLLMCallback对象。如果没有对系统提示词进行显式设置，会使用如下的提示词文本作为兜底。格式化时需要利用LocalContextBuilder构建的上下文来替换占位符{context_data}。占位符{question_count}表示希望生成问题的数量。

---Role---

You are a helpful assistant generating a bulleted list of {question_count} questions about data in the tables provided.


---Data tables---

{context_data}


---Goal---

Given a series of example questions provided by the user, generate a bulleted list of {question_count} candidates for the next question. Use - marks as bullet points.

These candidate questions should represent the most important or urgent information content or themes in the data tables.

The candidate questions should be answerable using the data tables provided, but should not mention any specific data fields or data tables in the question text.

If the user's questions reference several named entities, then each candidate question should reference all named entities.

---Example questions---

LocalQuestionGens实现的generate/agenerate方法具有一致的逻辑，现在我们来简单梳理一下实现在agenerate中针对QGen的实现：

• 如果利用question_history参数以字符串列表的形式提供了历史问题列表，会将最后一条作为查询文本（question_text），之前的问题则转换成ConversationHistory对象；
• 如果没有利用context_data参数提供上下文数据，则调用LocalContextBuilder对象的build_context方法生成上下文（具体使用的是返回的ContextBuilderResult对象的context_chunks字段）；
• 将context_data和question_count作为参数格式化提示词模板生成完整的系统提示词；
• 利用CompletionMessagesBuilder将系统提示词和查询文本（question_text）转换成角色分别为system和user的消息；
• 将消息列表作为参数调用LLMCompletion对象的completion_async方法对LLM发起流式调用（stream=True），收集实时返回的响应内容，并最终生成返回的QuestionResult对象。

class LocalQuestionGen(BaseQuestionGen):
    async def agenerate(
        self,
        question_history: list[str],
        context_data: str | None,
        question_count: int,
        **kwargs,
    ) -> QuestionResult:
        start_time = time.time()
        if len(question_history) == 0:
            question_text = ""
            conversation_history = None
        else:
            question_text = question_history[-1]
            history = [
                {"role": "user", "content": query} for query in question_history[:-1]
            ]
            conversation_history = ConversationHistory.from_list(history)

        if context_data is None:
            result = cast(
                "ContextBuilderResult",
                self.context_builder.build_context(
                    query=question_text,
                    conversation_history=conversation_history,
                    **kwargs,
                    **self.context_builder_params,
                ),
            )
            context_data = cast("str", result.context_chunks)
            context_records = result.context_records
        else:
            context_records = {"context_data": context_data}

        system_prompt = ""
        try:
            system_prompt = self.system_prompt.format(
                context_data=context_data, question_count=question_count
            )

            messages_builder = (
                CompletionMessagesBuilder()
                .add_system_message(system_prompt)
                .add_user_message(question_text)
            )

            response = ""
            response_stream: AsyncIterator[LLMCompletionChunk] = await self.model.completion_async(
                messages=messages_builder.build(),
                stream=True,
                **self.model_params,
            ) 

            async for chunk in response_stream:
                response_text = chunk.choices[0].delta.content or ""
                response += response_text
                for callback in self.callbacks:
                    callback.on_llm_new_token(response_text)

            return QuestionResult(
                response=response.split("\n"),
                context_data={
                    "question_context": question_text,
                    **context_records,
                },
                completion_time=time.time() - start_time,
                llm_calls=1,
                prompt_tokens=self.tokenizer.num_tokens(system_prompt),
            )

        except Exception:
            return QuestionResult(
                response=[],
                context_data=context_records,
                completion_time=time.time() - start_time,
                llm_calls=1,
                prompt_tokens=self.tokenizer.num_tokens(system_prompt),
            )