Agentic RAG流程Chroma检索失败问题排查与修复请求

核心问题分析 #

1. 嵌入函数不匹配：检索Agent使用Chroma默认嵌入函数，而向量库是用Azure OpenAI Embeddings构建的，嵌入格式不一致导致检索无结果。
2. 分解器提示模糊：未明确要求生成适合向量检索的精准子查询，可能导致子查询与向量库内容不匹配。
3. 缺少向量库连接验证：未确认是否成功连接到指定集合，无法排查路径/集合名称错误。

修复方案 #

1. 统一嵌入函数 #

修改ChromaRetrieverAgent，使用与向量库一致的Azure OpenAI嵌入函数，替换默认嵌入函数：

# --- Chroma Retriever Agent ---
class ChromaRetrieverAgent(AssistantAgent):
    def __init__(self, name, chroma_path, collection_name, model_client, embedding_function, system_message=None):
        super().__init__(name, model_client=model_client, system_message=system_message)
        self.chroma_path = chroma_path
        # 初始化Chroma客户端时指定匹配的嵌入函数
        self.client = PersistentClient(path=chroma_path)
        self.collection = self.client.get_or_create_collection(
            collection_name,
            embedding_function=embedding_function
        )
        self.embed_fn = embedding_function

    def retrieve(self, query, top_k=5):
        # 先检查集合是否有数据
        if self.collection.count() == 0:
            return "COLLECTION EMPTY"
        results = self.collection.query(query_texts=[query], n_results=top_k)
        docs = results.get("documents", [[]])[0]
        return docs

    def on_message(self, message, context):
        query = message.content.strip()
        if not query:
            return "INVALID QUERY"
        docs = self.retrieve(query)
        if isinstance(docs, list) and docs:
            return f"RETRIEVED: {json.dumps(docs, indent=2)}"
        elif docs == "COLLECTION EMPTY":
            return "COLLECTION EMPTY"
        else:
            return "NO DATA"

2. 优化分解器系统提示 #

让分解器生成更精准的检索子查询：

decomposer = AssistantAgent(
    "decomposer", model_client=client,
    system_message=(
        "将用户查询拆分为1-2个适合向量检索的具体子查询，子查询要直接对应员工休假资格相关的规则内容，"
        "只输出子查询内容，不要添加其他解释。"
    )
)

3. 验证向量库连接与数据 #

在初始化检索Agent后，添加验证代码，确认集合存在且有数据：

# 验证向量库数据量
print(f"Creta集合数据量: {creta_vector_retriever.collection.count()}")
print(f"Seltos集合数据量: {seltos_vector_retriever.collection.count()}")

4. 修正检索Agent初始化 #

传入之前定义的Azure嵌入函数：

creta_vector_retriever = ChromaRetrieverAgent(
    "creta_vector_retriever",
    chroma_path=VECTOR_STORE1_PATH,
    collection_name=VS1_COLLECTION_NAME,
    model_client=client,
    embedding_function=embedding_fn,  # 传入匹配的嵌入函数
    system_message="从向量库中检索与查询相关的文档，返回检索到的内容"
)
seltos_vector_retriever = ChromaRetrieverAgent(
    "seltos_vector_retriever",
    chroma_path=VECTOR_STORE2_PATH,
    collection_name=VS2_COLLECTION_NAME,
    model_client=client,
    embedding_function=embedding_fn,  # 传入匹配的嵌入函数
    system_message="从向量库中检索与查询相关的文档，返回检索到的内容"
)

5. 优化评估器上下文处理 #

确保评估器能整合所有检索结果：

context_evaluator = AssistantAgent(
    "context_evaluator", model_client=client,
    system_message=(
        "结合用户原始查询和所有检索到的上下文内容，判断是否足够回答问题。"
        "如果不足，回复'NOT ENOUGH'并说明需要补充的检索方向；"
        "如果足够，回复'ENOUGH'并汇总所有上下文信息。"
    )
)

完整修改后的关键代码片段 #

# ------------------- Embedding function -------------------
embedding_fn = AzureOpenAIEmbeddings(
    azure_endpoint=embedding_model_endpoint,
    api_key=embedding_model_api_key,
    deployment=embedding_model_deployment,
    api_version=EMBEDDING_API_VERSION
)

# --- Chroma Retriever Agent ---
class ChromaRetrieverAgent(AssistantAgent):
    def __init__(self, name, chroma_path, collection_name, model_client, embedding_function, system_message=None):
        super().__init__(name, model_client=model_client, system_message=system_message)
        self.chroma_path = chroma_path
        self.client = PersistentClient(path=chroma_path)
        self.collection = self.client.get_or_create_collection(
            collection_name,
            embedding_function=embedding_function
        )
        self.embed_fn = embedding_function

    def retrieve(self, query, top_k=5):
        if self.collection.count() == 0:
            return "COLLECTION EMPTY"
        results = self.collection.query(query_texts=[query], n_results=top_k)
        docs = results.get("documents", [[]])[0]
        return docs

    def on_message(self, message, context):
        query = message.content.strip()
        if not query:
            return "INVALID QUERY"
        docs = self.retrieve(query)
        if isinstance(docs, list) and docs:
            return f"RETRIEVED: {json.dumps(docs, indent=2)}"
        elif docs == "COLLECTION EMPTY":
            return "COLLECTION EMPTY"
        else:
            return "NO DATA"

# ------------------- AGENTS -------------------
decomposer = AssistantAgent(
    "decomposer", model_client=client,
    system_message=(
        "将用户查询拆分为1-2个适合向量检索的具体子查询，子查询要直接对应员工休假资格相关的规则内容，"
        "只输出子查询内容，不要添加其他解释。"
    )
)

creta_vector_retriever = ChromaRetrieverAgent(
    "creta_vector_retriever",
    chroma_path=VECTOR_STORE1_PATH,
    collection_name=VS1_COLLECTION_NAME,
    model_client=client,
    embedding_function=embedding_fn,
    system_message="从向量库中检索与查询相关的文档，返回检索到的内容"
)
seltos_vector_retriever = ChromaRetrieverAgent(
    "seltos_vector_retriever",
    chroma_path=VECTOR_STORE2_PATH,
    collection_name=VS2_COLLECTION_NAME,
    model_client=client,
    embedding_function=embedding_fn,
    system_message="从向量库中检索与查询相关的文档，返回检索到的内容"
)

context_evaluator = AssistantAgent(
    "context_evaluator", model_client=client,
    system_message=(
        "结合用户原始查询和所有检索到的上下文内容，判断是否足够回答问题。"
        "如果不足，回复'NOT ENOUGH'并说明需要补充的检索方向；"
        "如果足够，回复'ENOUGH'并汇总所有上下文信息。"
    )
)

# 验证向量库数据
print(f"Creta集合数据量: {creta_vector_retriever.collection.count()}")
print(f"Seltos集合数据量: {seltos_vector_retriever.collection.count()}")

内容的提问来源于stack exchange，提问作者Sushruth Kamarushi