简介：《Building AI Agents with LLMs, RAG, and Knowledge Graphs》是一本面向具备机器学习与自然语言处理基础、熟悉Python，并希望将AI系统落地于实际业务场景的开发者和研究人员的英文专著。书中从Transformer架构与注意力机制讲起，梳理LLM的演化脉络与主要局限；随后深入剖析RAG技术的发展路径，既说明朴素RAG的不足，也系统介绍高级RAG的分层索引、查询转换、重排序等优化策略，以及模块化RAG的设计思路。更进一步，作者引导读者使用LLM自动构建知识图谱，并将其与AI Agent集成，以增强语义理解与复杂推理能力，适合医疗、法律、金融等需要可解释、可追溯问答的场景。资源为单个PDF文件，压缩包大小约46.39MB，便于移动阅读与检索；目前已有163人学习浏览，对探索HybridRAG架构、多跳查询与关系挖掘的实践者有较高参考价值。

1. LLM、RAG、知识图谱组合的智能体，先从一次多跳翻车说起

把 LLM、RAG 和 Knowledge Graphs 组合成 AI Agents，第一步要承认：RAG 不是万能药。同一个多跳问题——「某科技公司的创始人，后来投资的另一家公司，核心业务是什么」——纯向量 RAG 会从知识库里捞出好几段相似度 0.8 以上的碎片，却拼不出横跨两个实体、三跳关系的完整链路。原因不难理解：向量检索擅长找语义相似的段落，但段落与段落之间的关系没有被显式建出来。

再加上知识图谱之后，事情就变了：实体之间的关系是一条条可遍历的边，多跳问题变成路径查询，答案可以溯源。LLM 负责理解意图和生成回答，RAG 负责把开放文档里的背景信息喂进来，图谱负责把关系链钉死，Agent 负责判断每一步该用哪个工具。这篇笔记按「建模 → 检索 → 编排 → 排错」的顺序讲一套可复现的链路，适合正在做知识库问答、智能客服和企业搜索的工程师。

2. 知识图谱建模先行：schema 设计、三元组抽取与「先粗后细」原则

很多团队在搭 Agent 时，第一步就急着写 ReAct 循环，结果知识没组织好，工具再多也答不对。常见做法是先问一句：我要回答的问题，底层靠的是文本语义，还是实体之间的关系？如果是后者，那就值得在向量库旁边补一个知识图谱。这一步的产出不是一个宏大本体，而是一张能跑起来的最小 schema。

2.1 纯向量 RAG 的三个瓶颈：图谱补在哪一环

第一个瓶颈是多跳。向量相似度只能告诉你「这段文本和问题像不像」，它不会主动把「张一鸣创立了字节跳动」和「张一鸣投资了某公司」这两条信息串成一条路径。模型需要把所有相关段落一起喂进去，靠注意力机制隐式拼接，一旦中间某段没被召回到，答案就断了。

第二个瓶颈是知识更新。文档改了一个事实，向量库需要重新切块、重新计算 embedding，成本很高；图谱只需要找到对应的那一条关系，做一次局部 MERGE。第三个瓶颈是证据链不透明。纯 RAG 给出来的依据是一整段原文，读者很难定位到「到底是哪个实体、哪条关系支撑了这个结论」。图谱把证据压缩成三元组，任何答案都可以反查到一条路径。

这也解释了 RAG 知识库和结构知识库的典型分界：开放域总结、语义相近的问答，靠 RAG 文本块就够了；实体密集、逻辑链固定、需要精确回答「谁和谁什么关系」的场景，图谱更合适。两者不是替代关系，而是同一套 Agent 里的两条腿。

2.2 最小可行图谱：实体类型、关系类型和属性怎么定

先粗后细是图谱建模里最值得遵守的原则。很多人一上来就设计二十种实体、三十种关系，结果 LLM 抽三元组时错误率飙升，图变得稀疏，查询经常断链。我一般会控制在 10 种关系以内，用属性保存细节，而不是拆出新的关系类型。

用一个科技企业投资图谱举例。实体类型只需要三个：Company、Person、Product。关系类型先定三种：FOUNDED_BY（公司由谁创立）、INVESTED_IN（公司投资了哪家公司）、DEVELOPED（公司研发了什么产品）。至于「A 轮投资还是 B 轮投资」「从哪一年开始任职」，全部放进关系属性里，不单独建关系。这样的粒度足够支撑大多数多跳问答，又不会让抽取任务难到模型胡编。

在 Neo4j 里先建好约束和索引，保证后续写入不会产生重复实体：

CREATE CONSTRAINT company_name IF NOT EXISTS FOR (c:Company) REQUIRE c.name IS UNIQUE; CREATE CONSTRAINT person_name IF NOT EXISTS FOR (p:Person) REQUIRE p.name IS UNIQUE; CREATE INDEX company_business IF NOT EXISTS FOR (c:Company) ON (c.business);

这段 Cypher 的意图很直接：Company 和 Person 的 name 是唯一键，同名实体在 MERGE 时只会被复用，不会被重复创建。第三个索引给 Company 的 business 属性建普通索引，因为后续会有大量「按业务类型找公司」的过滤查询，没有索引会全库扫描。注意这是 Neo4j 5.x 的语法，旧版本需要用CREATE CONSTRAINT ON (c:Company) ASSERT c.name IS UNIQUE的写法。

2.3 用 LLM 抽三元组：提示词模板、JSON 校验与去重写入

本地文本拆解这一步，不一定要买商业抽取服务。用本地模型跑一个高约束提示词，配合严格的 JSON 校验，效果完全够用。提示词里要把实体类型、关系类型、输出格式全部锁死，禁止模型发挥。

triple_extract_prompt = """你是知识图谱构建助手，从文本中抽取实体关系三元组。 只允许使用以下实体类型：Company, Person, Product。 只允许使用以下关系类型：FOUNDED_BY, INVESTED_IN, DEVELOPED。 规则： 1. 只抽取文本明确表达的事实，禁止推断 2. 同名实体必须使用相同名称，公司全称统一为简称 3. 输出 JSON 数组，格式： [{"source": "字节跳动", "relation": "FOUNDED_BY", "target": "张一鸣", "evidence": "字节跳动由张一鸣创立"}] 文本： {text} """

调用时把 temperature 设成 0，优先使用支持 JSON 输出格式的模型接口，解析失败就重试一次，再失败就把这段文本跳过并记入日志。很多人忽略了 evidence 字段，它在后面排错时非常有用——当图谱关系和原文对不上时，翻 evidence 就能定位是哪次抽取写错了。

写入 Neo4j 时，关系类型来自白名单，不要直接拼接用户输入。实体标签也需要按关系方向映射，否则所有节点都变成同一种标签，查询性能会退化。

LABEL_MAP = { "FOUNDED_BY": ("Company", "Person"), "INVESTED_IN": ("Company", "Company"), "DEVELOPED": ("Company", "Product"), } def write_triples(triples, driver): for t in triples: rel_type = t["relation"] s_label, t_label = LABEL_MAP.get(rel_type, ("Entity", "Entity")) cypher = f""" MERGE (s:{s_label} {{name: $source}}) MERGE (t:{t_label} {{name: $target}}) MERGE (s)-[r:{rel_type}]->(t) SET r.evidence = coalesce(r.evidence, []) + [$evidence] """ with driver.session() as session: session.run( cypher, source=t["source"], target=t["target"], evidence=t["evidence"], )

这里的 MERGE 是关键操作：实体不存在就创建，存在就复用；关系不存在就建立，存在就追加 evidence 到数组。coalesce 的作用是兼容首次写入时 r.evidence 为 null 的情况。整个函数是幂等的，同一个文档重复跑不会产生重复实体，只会追加证据来源。

3. 把 RAG 从「查段落」升级成「查关系」：GraphRAG 检索管线与 Cypher 生成

图谱建好之后，下一步是把检索管线搭起来。GraphRAG 不是简单地把向量检索换成 Cypher，而是三路召回同时跑：向量负责语义相似，Cypher 负责关系路径，全文索引负责兜底实体名匹配。三路结果按来源标签拼在一起，交给 LLM 做最终判断。

3.1 混合检索：向量召回、Cypher 查询与全文索引并行

先说为什么不能只靠 Cypher。用户的问题往往是自然语言，实体名可能带别名、简称，比如「字节」「ByteDance」都能指同一家公司。Cypher 的等值匹配遇到这种情况直接返回空。所以图谱查询之前，先建一个全文索引让它做模糊匹配。

CREATE FULLTEXT INDEX entity_index IF NOT EXISTS FOR (n:Person | Company) ON EACH [n.name];

混合检索的 Python 实现，我一般把三路结果放进一个列表，每一路都带上来源标签：

def hybrid_retrieve(question, entities, top_k=5): results = [] # 路1：向量召回，找语义相关的文本块 hits = vector_index.query(question, top_k=top_k) results.append(("vector", hits)) # 路2：图查询，按实体名找一跳关系 if entities: with driver.session() as session: rows = session.run( """ MATCH (n)-[r]-(m) WHERE n.name IN $entities RETURN n.name AS s, type(r) AS rel, m.name AS t, properties(r) AS props LIMIT 50 """, entities=entities, ).data() results.append(("graph", rows)) # 路3：全文索引，处理别名和不完整名称 fulltext_hits = fulltext_search(question, top_k=3) results.append(("text", fulltext_hits)) return results

三个参数需要根据场景调：top_k 在 5 到 10 之间，太小召回不足，太大上下文会爆炸；图查询的 LIMIT 设成 50，防止一个热门实体带出几百条关系把提示词冲垮；全文索引只取前 3 条，因为它是兜底，命中一个实体就够了。三路结果不需要合并去重，保持独立，LLM 看到「graph:」和「vector:」前缀时自然知道怎么对齐。

3.2 自然语言转 Cypher：意图分类和 few-shot 模板

查询生成是 GraphRAG 里最容易翻车的一环。我的做法是先用一个轻量分类把问题分成三类：实体属性型、关系路径型、开放总结型。前两类走图谱，第三类直接走向量 RAG，不要硬生成 Cypher。

Cypher 生成的提示词里，关键约束是「拿不准就返回 None」。这一步宁可少查，不能乱查，一条错误 Cypher 会把整个上下文污染。

cypher_gen_prompt = """你是 Neo4j 查询生成器。把问题转成 Cypher。 约束： 1. 实体名不做精确匹配，优先用 CONTAINS 或全文索引 2. 只读查询，禁止写操作 3. 拿不准时返回 None，不要编造查询 示例： 问题：字节跳动的创始人是谁？ 查询：MATCH (c:Company)-[:FOUNDED_BY]->(p:Person) WHERE c.name CONTAINS '字节' RETURN p.name 问题：{question} 查询： """

执行生成的 Cypher 时，要捕获所有异常。语法错误、字段不存在、约束冲突，任何一种都不该让整个流程中断。正确的做法是 try-except 包住执行，任何异常都降级为纯向量 RAG，同时把错误记录到日志里，后续用它来改进提示词。

3.3 端到端最小链路：Ollama 本地模型 + Neo4j 跑通 GraphRAG

整套链路在 Mac 或者 Linux 上都能跑通。Neo4j 用 Docker 起一个社区版，模型用 Ollama 拉一个 qwen2.5:7b，embedding 用 nomic-embed-text。资源占用可控，零基础也能跟着复现。

from ollama import chat, embed def graph_rag_answer(question: str) -> str: entities = extract_entities(question) # 用 LLM 抽取问题里的实体名 cypher = generate_cypher(question) # 3.2 的 few-shot 模板 ctx_blocks = [] if cypher and cypher.strip() not in ("None", ""): with driver.session() as session: rows = session.run(cypher).data() if rows: ctx_blocks.append("图谱证据:\n" + format_graph_rows(rows)) vec_rows = vector_store.query(question, top_k=5) ctx_blocks.append("文档证据:\n" + format_vec_rows(vec_rows)) final = chat("qwen2.5:7b", [ {"role": "system", "content": "结合图谱和文档证据作答。优先采用图谱证据，文档证据只作背景补充。"}, {"role": "user", "content": f"问题：{question}\n\n证据：\n" + "\n".join(ctx_blocks)}, ]) return final["message"]["content"]

这一段把前面所有东西串起来了：先抽实体、再生成 Cypher、执行图查询，同时跑向量召回，最后把两路证据拼进提示词让 LLM 生成。整套链路的成败不单看代码，更看格式函数怎么处理空结果——graph_rag_answer 里如果 rows 为空，图谱证据块干脆不出现，LLM 就不会被空数据干扰。

4. 让 Agent 自己决定查什么：工具注册、ReAct 循环与状态管理

检索链路稳定之后，Agent 的作用是把选择权交给模型：这个问题该查图，还是该查文档，还是两路都要。这一章不推荐直接用 LangChain 的 AgentExecutor，黑匣子问题太严重，工具返回格式、循环终止条件、错误恢复都不容易控制。手写一个轻量 ReAct 循环，整套逻辑都摊在自己眼前。

4.1 Agent 的四层结构：内核、工具集、记忆与策略

常见做法是把 Agent 拆成四个层次。内核是 LLM，负责推理和生成，参数不多，但 temperature 和 max_tokens 直接决定行为稳定性。工具集是一个注册表，每个工具包含描述、执行函数、最大调用次数三个字段。记忆分短期和长期，短期是对话历史，长期是图谱里已经确认的实体状态。策略层是 ReAct 循环加上路由规则，决定「什么时候调用哪个工具，什么时候必须停」。

工具注册表用一个 dict 就能搞定：

TOOLS = { "graph_query": { "desc": "查询知识图谱中的实体关系", "run": graph_query, "max_calls": 3, }, "vector_rag": { "desc": "检索文档片段补充背景", "run": vector_rag, "max_calls": 3, }, "final": { "desc": "给出最终答案", "run": None, "max_calls": 1, }, }

每个工具都带 max_calls，是为了防止同一个操作反复执行。手写的好处是这些计数逻辑完全可见，不像框架那样把循环行为藏在抽象层里。

4.2 手写一个轻量 ReAct 循环：行动解析与终止条件

ReAct 循环的本质是让 LLM 输出「Thought + Action + Action Input」，代码负责解析并执行，把结果作为 Observation 喂回去。循环一定要有硬终止条件，否则模型陷入死循环时没有任何保护。

def agent_loop(question, max_steps=5): messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": question}, ] for step in range(max_steps): reply = llm_chat(messages, temperature=0.2) action = parse_action(reply) # 解析 LLM 输出的 JSON 或 XML if action["type"] == "final": return action["answer"] if action["type"] not in TOOLS: messages.append({ "role": "user", "content": "无效工具，请从注册表中选择", }) continue try: obs = TOOLS[action["type"]]["run"](action.get("args", {})) except Exception as e: obs = f"工具执行失败：{e}" messages.append({ "role": "user", "content": f"Observation: {obs}", }) return "未能完成，需要拆分成更具体的问题"

两个参数很关键。max_steps 设成 5，超过直接结束，而不是返回兜底答案——这么做是为了暴露出流程问题。temperature 设成 0.2，太低模型不会探索，太高行动格式会开始漂移。parse_action 失败时，要把原始回复追加到上下文里，让模型看到自己刚才输出了什么。

4.3 多轮对话的状态管理：实体状态池与上下文压缩

多轮对话里最常见的坑是代词解析。「那家公司后来怎么样了？」——「那家」指代上一轮提到的公司，如果不做状态管理，LLM 只能靠对话历史猜，猜错的概率很高。我一般会维护一个实体状态池，每轮把确认过的实体名和类型放进去，下一轮优先用它补全指代。

state_pool = {} def resolve_entities(question, state_pool): if any(w in question for w in ("它", "那家", "这个公司")): return state_pool.get("last_company") entities = extract_entities(question) if entities: state_pool["last_company"] = entities[0] return entities

上下文超过六轮之后，直接把全部历史塞给 LLM 会稀释注意力。我一般会用另一个 LLM 调用把历史压成两段式摘要：一段保存事实性结论，另一段保存待办意图。实体状态池不参与压缩，它始终保持结构化，这样图谱查询永远有干净的输入。

5. 避坑：检索空、答案冲突、死循环与图片入库的 5 个排查记录

这套链路跑起来之后，真正耗时间的不是写代码，是排错。这里整理五条血泪经验，都是实际跑链路时反复踩过的问题，按「现象 → 原因 → 解决」的方式来写。

5.1 图谱里明明有关系，查询却返回空

现象：在 Neo4j Browser 里手动查，关系清清楚楚；Agent 一跑，返回空。

原因九成是实体名不匹配。「字节跳动科技有限公司」和「字节跳动」在数据库里是两个节点，「ByteDance」又是第三个节点。Cypher 的等值匹配完全失效。还有一种是关系方向写反了，FOUNDED_BY 的箭头指向反了，查询结果自然为空。

解决：不要依赖精确匹配，查询前先做一次实体归一化，维护别名表；拿不准方向时，先跑一个无方向探针确认关系确实存在：

MATCH (a)-[r]-(b) WHERE a.name CONTAINS '字节' OR b.name CONTAINS '字节' RETURN type(r) AS rel, labels(a) AS a_label, labels(b) AS b_label LIMIT 20

这条查询不关心方向，能告诉你「字节跳动」这个实体到底参与了哪些关系、方向是什么，是排空结果的第一把钥匙。

5.2 向量 RAG 的答案与图谱证据冲突

现象：图谱里明确写着张一鸣创立了字节跳动，生成的答案却说创始人是别人。

原因：向量检索到的文档片段更长、更详细，LLM 的注意力被长文本带走了，图谱证据虽然也拼在提示词里，但排位靠后，权重被稀释。

解决：在系统提示词里显式声明证据优先级，并说明冲突时的处理规则：

提示：图谱证据优先于文档证据；文档证据只用于补充背景细节，不能推翻图谱中已存在的关系。

如果业务上确实需要文档更新图谱，那应该走图谱更新流程，而不是在生成时让模型自由选择。

5.3 Agent 反复调用同一个工具，进入死循环

现象：日志里 graph_query 连续被调用五次，参数一模一样。

原因：工具返回了空结果，LLM 以为执行没成功，于是再试一次。本质是提示词里缺了一句「空结果也是有效结果」。

解决：工具返回值统一加 status 字段，同时在 ReAct 循环里记录工具调用历史，发现连续两次相同参数调用同一工具时，直接注入一条观察结果「该查询已执行过且结果为空，请换思路」。

5.4 知识库里存了图片，检索却命不中

现象：图片路径存进了知识库，向量检索怎么搜都搜不到。

原因：文本 embedding 模型不处理像素，它只对「图片路径字符串」做向量化，路径本身不含任何语义信息。

解决：先用多模态模型生成图片的文字描述，把描述 embedding 进向量库，图片路径作为元数据保存。命中描述后，再把图片路径返回给应用层。

desc = vlm_describe(image_path) # 描述样例："办公楼外观，玻璃幕墙，楼顶有公司标志" vector_store.add(desc, meta={"image_path": image_path})

5.5 关系类型拆太细，图谱稀疏到没法查

现象：边界案例能查到，常规问题经常断链。

原因：schema 设计阶段把关系拆得过细，「曾任 CEO」「现任 CEO」「曾担任董事」各建一种关系，数据一分散，每种关系都只剩几条，图遍历基本走不通。

解决：先合并成粗粒度关系，细节放属性。比如统一建 WORKS_AT 关系，用 from_year、to_year 和 role 属性区分，等数据量上来再考虑迁移拆分。这是第 2 章「先粗后细」原则的代价体现。

6. 验证到进阶：评测集、增量更新与一条长期习惯

链路能跑通只是起点，能不能持续演进取决于两件事：有量化指标，有低成本的更新路径。最后这一章讲一套我用了很久的验证方法，外加一个值得长期坚持的工程习惯。

6.1 用 30 条问题把链路钉死：评测集与三个指标

不要凭感觉判断「效果还不错」。建一个至少三十条评测题的集子，覆盖三类问题：

跑评测时重点看三个指标：图谱命中率（Cypher 查到了非空结果的比例）、答案准确率（人工或 LLM 打分）、平均工具调用次数。调用次数超过 4 次说明 Agent 在绕路，需要查路由策略。

6.2 增量更新：新文档进来，只补关系不重建图谱

新文档到达后，向量库直接 add 新块；图谱不能直接跑抽取脚本，因为同一份文档重复抽取会造成关系残留。我的做法是给每个文档一个 source_id，入库前先清理该 source 创建的关系：

MATCH ()-[r]->() WHERE r.source_id = $source_id DELETE r;

注意这里只删关系，不删实体节点。实体可能被其他文档共享，删节点会误伤别的数据源。删完关系后重新跑三元组抽取，整个更新过程就是幂等的。

6.3 一条习惯：把每次失败沉淀成评测用例

我的经验是，每次线上或测试出现检索为空、答错、死循环，当天就把问题、预期答案、失败原因写成一个 JSON 用例丢进评测集。三个月之后，评测集就是最真实的需求清单，每次改动提示词或 schema，先跑一遍评测集，分数下降就回滚，分数上升再发布。

这个习惯帮我避免了至少十次「改回去又改回来」的返工。效果从玄学变成了可对比的数字，希望帮到你。

本文还有配套的精品资源，点击获取