AI Agent开发实战⑰多跳检索与知识图谱让RAG突破单跳限制用户问马斯克创立的公司有哪些员工超过10000人传统RAG检索两次就断了——先找马斯克创立的公司再找这些公司的员工数。多跳检索就是解决这个问题的让Agent像人一样一步一步查下去。一、单跳检索的局限性传统RAG是一问一查一答用户马斯克创立的公司有哪些员工超过10000人 传统RAG检索 Query: 马斯克创立的公司 员工超过10000人 检索结果混合了马斯克、公司、员工数的片段但都是碎片化信息 问题 1. 没有一篇文档同时包含马斯克创立的公司和员工数 2. 需要先查公司列表再查每个公司的员工数 3. 这是一个两跳问题传统RAG只能处理一跳多跳问题的特征问题包含多个子问题子问题之间有依赖关系需要用前一步的结果作为下一步的输入二、多跳检索的三种架构2.1 迭代检索Iterative Retrieval最直观的方案查一步、用结果再查一步。classIterativeRetriever:迭代检索def__init__(self,llm,retriever,max_hops:int3):self.llmllm self.retrieverretriever self.max_hopsmax_hopsdefretrieve(self,query:str)-list[dict]:多跳检索all_docs[]current_queryqueryforhopinrange(self.max_hops):print(f第{hop1}跳查询{current_query})# 检索docsself.retriever.search(current_query,k5)all_docs.extend(docs)# 判断是否需要继续ifself._is_complete(docs,query):break# 生成下一跳查询next_queryself._generate_next_query(query,docs)ifnotnext_query:breakcurrent_querynext_queryreturnall_docsdef_is_complete(self,docs:list,original_query:str)-bool:判断是否已经找到答案promptf 原始问题{original_query}检索到的文档{self._format_docs(docs)}请判断这些文档是否足以回答原始问题 回答是/否 responseself.llm.invoke(prompt)return是inresponse.contentdef_generate_next_query(self,original_query:str,docs:list)-str:生成下一跳查询promptf 原始问题{original_query}已检索到的文档{self._format_docs(docs)}这些文档还不足以回答原始问题。请生成下一步需要查询的内容。 要求 1. 基于已检索到的信息 2. 补充回答原始问题所需的信息 3. 简洁明了一句话 下一步查询 responseself.llm.invoke(prompt)returnresponse.content.strip()def_format_docs(self,docs:list)-str:return\n.join([f-{doc[content][:200]}fordocindocs])# 使用示例retrieverIterativeRetriever(llm,vector_retriever)query马斯克创立的公司有哪些员工超过10000人docsretriever.retrieve(query)# 输出# 第1跳查询马斯克创立的公司有哪些员工超过10000人# 第2跳查询SpaceX Tesla Twitter员工人数# 第3跳查询SpaceX员工数 Tesla员工数 Twitter员工数2.2 分解检索Decomposition Retrieval先拆解问题再分别检索。classQueryDecomposer:查询分解器def__init__(self,llm):self.llmllmdefdecompose(self,query:str)-list[str]:将复杂问题分解为子问题promptf 用户问题{query}请将这个问题分解为多个独立的子问题。 要求 1. 每个子问题可以独立回答 2. 子问题的答案组合起来可以回答原问题 3. 每行一个子问题不要编号 示例 原问题马斯克创立的公司有哪些员工超过10000人 子问题 马斯克创立了哪些公司 SpaceX有多少员工 Tesla有多少员工 Twitter有多少员工 responseself.llm.invoke(prompt)sub_queries[line.strip()forlineinresponse.content.split(\n)ifline.strip()]returnsub_queriesclassDecompositionRetriever:分解检索def__init__(self,llm,retriever):self.llmllm self.retrieverretriever self.decomposerQueryDecomposer(llm)defretrieve(self,query:str)-dict:分解并检索# 分解问题sub_queriesself.decomposer.decompose(query)print(f分解为{len(sub_queries)}个子问题)fori,sqinenumerate(sub_queries,1):print(f{i}.{sq})# 分别检索results{}forsub_queryinsub_queries:docsself.retriever.search(sub_query,k3)results[sub_query]docs# 合并结果all_docs[]fordocsinresults.values():all_docs.extend(docs)return{sub_queries:sub_queries,results:results,all_docs:self._deduplicate(all_docs)}def_deduplicate(self,docs:list)-list:去重seenset()unique[]fordocindocs:ifdoc[id]notinseen:seen.add(doc[id])unique.append(doc)returnunique2.3 知识图谱增强检索KG-Enhanced Retrieval利用知识图谱存储实体关系进行结构化查询。classKnowledgeGraphRetriever:知识图谱增强检索def__init__(self,neo4j_driver,llm):self.driverneo4j_driver self.llmllmdefretrieve(self,query:str)-list[dict]:知识图谱检索# 提取实体entitiesself._extract_entities(query)# 在图谱中查询关系factsself._query_graph(entities)# 用图谱信息补充检索docsself._retrieve_with_facts(query,facts)returndocsdef_extract_entities(self,query:str)-list[str]:提取实体promptf 从以下文本中提取实体人名、公司名、地点等{query}每行一个实体不要编号。 responseself.llm.invoke(prompt)return[line.strip()forlineinresponse.content.split(\n)ifline.strip()]def_query_graph(self,entities:list[str])-list[dict]:在知识图谱中查询facts[]withself.driver.session()assession:forentityinentities:# 查询与实体相关的关系resultsession.run( MATCH (e:Entity {name: $name})-[r]-(related) RETURN e.name, type(r) as relation, related.name LIMIT 10 ,nameentity)forrecordinresult:facts.append({subject:record[e.name],relation:record[relation],object:record[related.name]})returnfactsdef_retrieve_with_facts(self,query:str,facts:list[dict])-list[dict]:用图谱事实增强检索# 构建扩展查询fact_text\n.join([f{f[subject]}{f[relation]}{f[object]}forfinfacts])enhanced_queryf{query}\n相关信息{fact_text}# 这里可以调用向量检索returnvector_retriever.search(enhanced_query)# 知识图谱数据结构 节点 - Entity: {name: 马斯克, type: 人物} - Company: {name: Tesla, type: 公司} 关系 - (:Entity {name: 马斯克})-[:FOUNDED]-(:Company {name: Tesla}) - (:Company {name: Tesla})-[:HAS_EMPLOYEES]-(:Value {count: 127855}) # 构建知识图谱defbuild_knowledge_graph(documents:list[str],llm):从文档构建知识图谱fordocindocuments:# 用LLM提取三元组promptf 从以下文档中提取实体和关系{doc}输出格式每行一个三元组 实体1, 关系, 实体2 示例 马斯克, 创立, SpaceX SpaceX, 员工数, 13000 responsellm.invoke(prompt)triples[line.split(, )forlineinresponse.content.split(\n)if,inline]# 写入图谱fortripleintriples:iflen(triple)3:subject,relation,obj[t.strip()fortintriple]write_to_graph(subject,relation,obj)三、三种方案对比方案优势劣势适用场景迭代检索简单易实现依赖LLM判断可能走偏通用场景分解检索逻辑清晰子问题可能遗漏依赖问题可明确分解知识图谱结构化查询、准确需要构建图谱、维护成本高特定领域、实体关系明确四、实测效果对比4.1 测试设置测试数据-文档5000篇公司新闻-查询50个多跳查询-评估Answer Accuracy答案准确率4.2 效果对比方案准确率平均检索次数平均耗时单跳检索基线42.3%1120ms迭代检索68.7%2.4580ms分解检索71.2%3.1620ms知识图谱78.5%1图谱查询350ms关键发现多跳检索比单跳提升26-36%知识图谱效果最好但需要前期构建成本五、选型决策第一步问题类型判断 │ ├── 问题可明确分解为子问题 │ → 【分解检索】 │ ├── 问题需要逐步探索 │ → 【迭代检索】 │ └── 涉及大量实体关系查询 → 【知识图谱】 第二步资源评估 │ ├── 无图谱需要快速实现 │ → 【迭代检索】 │ ├── 有图谱或愿意构建 │ → 【知识图谱】 │ └── 问题模式固定 → 【分解检索】六、实战案例公司信息查询AgentclassCompanyInfoAgent:公司信息查询Agent多跳检索示例def__init__(self,llm,retriever,kg_retrieverNone):self.llmllm self.retrieverretriever self.kg_retrieverkg_retrieverdefanswer(self,query:str)-str:回答多跳问题# 判断是否需要多跳ifself._needs_multi_hop(query):# 使用迭代检索docsself._multi_hop_retrieve(query)else:# 单跳检索docsself.retriever.search(query,k5)# 生成答案contextself._format_docs(docs)answerself._generate_answer(query,context)returnanswerdef_needs_multi_hop(self,query:str)-bool:判断是否需要多跳multi_hop_keywords[哪些,分别,每个,所有]returnany(kwinqueryforkwinmulti_hop_keywords)def_multi_hop_retrieve(self,query:str)-list[dict]:多跳检索# 第1跳检索初步信息docs_1self.retriever.search(query,k10)# 提取关键实体entitiesself._extract_entities_from_docs(docs_1)# 第2跳针对每个实体深入检索docs_2[]forentityinentities[:5]:# 限制实体数量entity_docsself.retriever.search(f{entity}员工数,k2)docs_2.extend(entity_docs)# 合并去重all_docsself._deduplicate(docs_1docs_2)returnall_docsdef_extract_entities_from_docs(self,docs:list)-list[str]:从文档中提取实体text .join([d[content][:200]fordindocs])promptf 从以下文本中提取公司名称{text}每行一个公司名不要编号。 responseself.llm.invoke(prompt)return[line.strip()forlineinresponse.content.split(\n)ifline.strip()]# 使用示例agentCompanyInfoAgent(llm,retriever)query马斯克创立的公司有哪些员工超过10000人answeragent.answer(query)print(answer)七、总结场景推荐方案准确率提升实现难度通用多跳问题迭代检索26%中可分解问题分解检索29%中实体关系查询知识图谱36%高多跳检索是复杂问题的必选项但会增加延迟需要权衡效果和性能。下篇预告「上下文压缩与选择让LLM看到最有价值的信息」——检索到50篇文档但LLM只能看5篇如何选择需要完整多跳检索代码的同学可以看我主页的付费资源专栏。有问题欢迎评论区留言大家一起讨论