一、为什么很多 RAG 系统一进入真实业务就开始失真

很多团队第一次做 RAG，路径都非常相似：

1. 文档切块
2. 向量化入库
3. 查询时做 Top-K 召回
4. 把召回结果拼进 Prompt
5. 让大模型生成答案

这个方案可以快速做出 Demo，但很难稳定支撑企业场景。

因为真实业务里的问题，通常不是一句话能靠相似度直接命中的。

例如：

“上周支付网关超时事故影响了哪些高价值客户？这些客户在事故前后是否还命中过风控异常？相关链路的负责人、修复动作和变更记录分别是什么？”

这个问题同时包含：

• • 事故实体
• • 服务依赖关系
• • 用户分层
• • 风控关联事件
• • 负责人归属
• • 变更时间线

如果仍然只用平面式向量检索，系统大概率会出现三类问题：

• • 能召回“支付网关”文档，但召不回“事故影响用户”和“相关负责人”之间的关系
• • 能召回多个碎片，但 LLM 只能自己脑补因果链
• • 能生成一段流畅答案，但无法证明每个事实从哪里来

这正是传统 RAG 在复杂知识场景下的根本短板：

它擅长找相似文本，不擅长还原知识结构。

而 GraphRAG 与 Agentic RAG，分别解决的是这两个层面的问题：

• • GraphRAG 解决“知识之间如何连接”
• • Agentic RAG 解决“系统应该按什么策略去找”

前者让检索从“命中若干 chunk”升级为“返回可推理的知识子图”，后者让检索从“一次性调用”升级为“具备规划、验证、重试、降级能力的控制循环”。

如果企业要做的不是问几个 FAQ，而是面向运维诊断、金融风控、企业知识分析、研发助手、流程调查、审计溯源等复杂场景，那么这两者几乎一定会出现在同一套架构里。

---

二、先把边界讲清：GraphRAG 和 Agentic RAG 分别解决什么

2.1 传统 RAG 的能力边界

传统 RAG 的核心过程是：

用户问题  -> Query Embedding  -> 向量检索 Top-K  -> 拼接上下文  -> LLM 生成

它的优势很明确：

• • 实现成本低
• • 接入快
• • 对 FAQ、制度问答、说明文档检索比较有效

它的问题也同样明确：

• • 无法显式建模实体关系
• • 对跨文档多跳问题不稳定
• • 对复杂业务链路的还原能力弱
• • 上下文一长就容易噪声膨胀

2.2 GraphRAG 的本质

GraphRAG 的核心不是“再加一个图数据库”，而是：

把原本散落在文档里的隐性知识结构抽出来，变成显式可检索的实体、关系、社区和摘要层。

所以 GraphRAG 检索的不是一堆独立 chunk，而是：

• • 节点
• • 边
• • 社区
• • 子图摘要
• • 支撑这些结构的原文证据

2.3 Agentic RAG 的本质

Agentic RAG 的核心也不是“多调几次模型”，而是：

让系统把检索和生成看作一个可控制、可回退、可评估的任务执行过程。

它会显式回答下面这些问题：

• • 当前问题应该走哪种检索策略
• • 是否需要拆分子问题
• • 第一次召回是否足够
• • 是否要重写查询
• • 是否要调用图谱、SQL、搜索、外部 API 等多个工具
• • 在成本、延迟、效果之间如何动态取舍

2.4 两者不是替代关系，而是层次不同

维度	GraphRAG	Agentic RAG
核心关注点	知识结构	检索策略
解决问题	找到“关系”	决定“怎么找”
优势	多跳、全局综述、知识连接	规划、迭代、工具编排、动态决策
成本	离线索引高	在线推理高
典型失败点	图谱构建质量差	迭代失控、延迟过高

最适合生产的方式通常不是二选一，而是：

用 Agent 负责调度，用 GraphRAG 负责提供高质量结构化上下文。

---

三、GraphRAG 的技术本质：从“相似文本”到“结构化知识检索”

3.1 GraphRAG 为什么有效

很多企业知识天然就带图结构属性：

• • 系统与系统之间有依赖
• • 用户与订单之间有关联
• • 事故与告警之间有因果
• • 组织与职责之间有归属
• • 规则与例外之间有约束

如果这些关系不被抽取出来，那么 LLM 只能从若干无序 chunk 中“猜”关系。

GraphRAG 则把知识组织成如下结构：

实体（Entity）  服务、团队、接口、用户组、规则、事故、配置项关系（Relation）  depends_on、owned_by、impacts、belongs_to、resolved_by社区（Community）  支付域、订单域、风控域、稳定性治理域摘要（Summary）  节点摘要、社区摘要、子图摘要、时间线摘要

一旦在线检索拿到的是一个“关系子图”，模型就能基于显式结构进行回答，而不是单纯依据语义相近性拼接。

3.2 GraphRAG 的标准离线建图流程

生产级 GraphRAG 不只是“抽实体入 Neo4j”，通常至少包括下面七步：

文档采集

解析与标准化

语义切块

实体 / 关系 / Claim 抽取

实体消歧与归一

社区发现与摘要生成

图索引 + 向量索引 + 元数据索引

这七步里，真正决定 GraphRAG 质量的，不是图库本身，而是中间四个步骤：

• • 抽取得准不准
• • 消歧做得好不好
• • 社区切得稳不稳
• • 摘要能不能代表局部结构

3.3 实体抽取不是“NER”那么简单

企业 GraphRAG 中，实体抽取往往至少要做三层建模。

第一层是类型建模：

• • Service
• • Application
• • API
• • Database
• • Incident
• • ChangeTicket
• • Team
• • Owner
• • CustomerTier
• • Policy

第二层是别名建模：

• • payment-gateway
• • pay-gw
• • 支付网关
• • PGW

第三层是上下文属性建模：

• • 所属租户
• • 环境
• • 时间范围
• • 来源系统
• • 可信度
• • 权限标签

如果没有这些属性，后续很难做：

• • 权限过滤
• • 时间范围问答
• • 多租户隔离
• • 灰度索引切换

3.4 实体消歧是 GraphRAG 成败关键

真实文档中的同一实体，经常会以多种形式出现：

• • 中英文混用
• • 缩写与全称并存
• • 不同团队有不同叫法
• • 相同名称在不同租户下含义不同

所以生产级消歧通常不是一个动作，而是分层判定：

1. 规则归一
2. embedding 相似度粗筛
3. LLM 语义比对复核
4. 结合租户、系统、时间、标签做最终归并

一个常见错误是只按名称去重，这很容易把不同租户中的同名服务错误合并，直接造成串库。

3.5 社区发现的价值，不只是“聚类好看”

GraphRAG 和普通知识图谱的一个重要差异是：

它非常强调社区层和社区摘要层。

原因很简单。

用户的问题并不总是指向单个实体，也可能是：

• • “支付稳定性最近的主要风险有哪些”
• • “这个季度用户投诉集中在哪些链路”
• • “跨部门协作里最常见的瓶颈是什么”

这类问题如果仍然从具体实体开始检索，路径会非常长，噪声也很高。
而社区摘要层提供了一个更适合全局问题的语义入口。

也就是说，GraphRAG 实际上至少有两层检索面：

• • 实体面
• • 社区面

它们分别服务于局部问题和全局问题。

3.6 GraphRAG 的三种常见检索模式

1. Local Search

从种子实体出发，做 1~2 跳图展开，适合：

• • “支付网关依赖哪些服务”
• • “订单补偿任务由谁负责”
• • “事故 4821 影响了哪些模块”

2. Global Search

从社区摘要层开始检索，适合：

• • “支付域的主要稳定性问题有哪些”
• • “风控误杀集中在哪些节点”
• • “跨境支付链路的薄弱点是什么”

3. Hybrid / DRIFT Search

先锁定实体，再补充所属社区和邻域摘要，适合：

• • “支付网关为什么总是被投诉”
• • “某个事故在更大系统背景里意味着什么”

---

四、Agentic RAG 的技术本质：从“一次检索”到“受控推理循环”

4.1 Agentic RAG 不是无限调用模型，而是有限状态机

生产系统里最忌讳的一件事，就是让 Agent “自由发挥”。

真正能上线的 Agentic RAG，一定是一个强约束状态机：

接入请求  -> 意图识别 / 路由  -> 任务拆分  -> 选择检索工具  -> 召回证据  -> 评估是否充分  -> 不充分则改写 / 补检 / 降级  -> 生成答案  -> 事实校验 / 引用校验  -> 返回结果

这套流程和普通“多轮调用”最大的区别在于：

• • 每一步都有清晰职责
• • 每一步都可观测
• • 每一步都可以配置超时、预算、阈值、熔断规则

4.2 一个生产级 Agentic RAG 至少要有这六个角色

角色	作用	适合模型
Router	分类场景、判断走向	小模型
Planner	拆子问题、设定检索计划	中模型
Retriever	执行向量、图谱、SQL、API 检索	非模型 / 工具
Evaluator	评估证据相关性与完整性	小模型
Generator	基于证据生成答案	大模型
Verifier	做引用一致性与事实门控	小模型或规则引擎

这六个角色不一定是六个服务，但职责上最好是拆开的。

4.3 为什么 Agentic RAG 能比普通 RAG 更强

因为它可以显式处理三类传统 RAG 很弱的问题。

1. 子问题拆分

例如：

“哪些高价值客户受到了支付事故影响，他们在其他模块是否也遇到过相似问题？”

这句话至少可以拆成：

1. 找支付事故影响名单
2. 识别高价值客户
3. 查询这些客户在其他模块的异常记录
4. 汇总相关负责人和时间线

普通 RAG 常常把这些都塞进一次检索里，召回结果会混成一团。

2. 工具路由

不是所有信息都应该走向量检索。

• • 客户等级可能在 CRM API
• • 事故工单可能在 Jira / 工单系统
• • 服务依赖在 GraphRAG
• • 精确编号在 Elasticsearch / SQL

Agentic RAG 的强项，是让系统按问题类型走最合适的工具。

3. 证据不足时的动态补检

第一次召回不够时，系统可以：

• • 改写 query
• • 放宽过滤条件
• • 改用图谱检索
• • 追加 SQL 或 API 检索
• • 直接降级为“不足以回答”

这比让模型拿着不充分上下文硬答，风险低得多。

4.4 生产中必须有的硬约束

没有约束的 Agentic RAG，在生产里几乎必然失控。

建议至少设置如下上限：

项目	建议值	说明
最大迭代次数	2~3	再高通常收益很小
单次总超时	8s~30s	取决于业务 SLA
单次召回上限	5~8 份证据	防止上下文膨胀
总 token 预算	16K~64K	防止单请求失控
工具调用上限	3~6 次	防止横向爆炸
去重窗口	最近 2~3 轮	防止重复检索

生产系统不是为了“让模型想得更久”，而是为了“让系统在预算内做最值钱的那几步”。

---

五、融合架构：GraphRAG × Agentic RAG 的生产级分层设计

5.1 推荐采用“控制面 + 数据面”双层架构

最稳定的做法，不是把所有逻辑堆到一个 Agent 服务里，而是拆成两层：

控制面

负责：

• • 路由
• • 规划
• • 状态机编排
• • 预算控制
• • 超时控制
• • 结果校验
• • Trace 聚合

数据面

负责：

• • 向量检索
• • 图谱检索
• • BM25 检索
• • SQL 查询
• • API 查询
• • 排序与过滤

这种拆法的好处是很明确的：

• • 控制逻辑清晰
• • 数据服务可独立扩容
• • 检索工具可插拔
• • 更容易做多租户和权限治理

5.2 一套可落地的整体架构

API Gateway

Orchestrator / Agent Runtime

Router / Planner / Evaluator / Verifier

Retrieval Gateway

GraphRAG Service

Vector Search Service

Keyword / SQL / API Connectors

Neo4j / Graph Store

Elastic / Milvus / pgvector

Redis Cache

OpenTelemetry / Metrics / Logs

Kafka / CDC Pipeline

Indexing Service

5.3 在线查询链路应该怎么走

一个复杂问题进入系统后，推荐链路如下：

1. 网关完成鉴权、限流、租户识别
2. Orchestrator 创建请求级状态
3. Router 判断是 FAQ、精确查询、关系分析还是全局综述
4. Planner 决定是否拆成子问题
5. Retrieval Gateway 根据计划调用 GraphRAG、向量、BM25、SQL 或外部 API
6. Evaluator 对证据相关性和完整性打分
7. 若证据不足，则触发有限次补检
8. Generator 基于筛选后的证据生成答案
9. Verifier 检查引用、事实映射、敏感信息
10. 统一返回答案、引用、置信度、链路标识

5.4 离线建图链路应该怎么走

离线链路推荐完全异步化：

1. 文档变更通过 CDC / MQ / 定时扫描 产生事件
2. 文档解析服务做格式抽取与标准化
3. 分块服务做语义切块与元数据补全
4. 抽取服务做实体 / 关系 / claim 提取
5. 消歧服务完成归一与去重
6. 图索引服务更新图谱与社区摘要
7. 向量服务更新实体向量、chunk 向量、摘要向量
8. 版本服务完成索引切换与缓存预热

在线链路不应该承担这些重任务。

5.5 多租户设计必须在第一天就考虑

GraphRAG 和 Agentic RAG 一旦进入企业环境，多租户几乎是必选项。

至少要隔离以下信息：

• • tenant_id
• • knowledge_base_id
• • index_version
• • permission_tags
• • data_domain

推荐策略是：

• • 图节点和边带租户属性
• • 向量索引支持元数据过滤
• • Agent 状态中显式传递租户上下文
• • 检索前过滤，而不是生成后再做清洗

后补权限控制，风险非常高。

---

六、工程化升级重点：高并发、可扩展、可降级、可回滚

6.1 高并发场景下，系统真正的瓶颈在哪

RAG 系统在高峰期的瓶颈通常不是一个点，而是四类资源同时紧张：

• • Embedding 与 LLM 推理资源
• • 图查询与向量查询资源
• • Redis / ES / Neo4j 连接池
• • 上下文拼装与序列化 CPU

如果只盯着模型延迟，很容易漏掉：

• • 连接池耗尽
• • 图遍历 fan-out 爆炸
• • 上下文过长导致生成端排队
• • Agent 迭代导致单请求放大为多次下游调用

6.2 生产级并发治理的五个原则

1. 限并发，不限线程

真正需要控制的是“昂贵操作”的并发数，而不是简单线程数。

例如：

• • 大模型生成
• • Embedding 批处理
• • 图谱多跳查询
• • 外部 API 调用

2. 重操作异步化

离线入库、社区重建、摘要刷新、全量重索引必须异步化。

3. 读写隔离

在线读索引与离线写索引尽量隔离，至少要支持版本切换。

4. 结果缓存前置

热门 query、子图摘要、实体邻域、query rewrite、rerank 结果都值得缓存。

5. 统一降级路径

当 GraphRAG 慢、外部 API 挂、模型超时、检索为空时，系统必须知道退到哪里。

6.3 一套实用的降级顺序

建议的降级顺序如下：

1. 1. GraphRAG + Agentic
2. 2. Vector + Agentic
3. 3. Hybrid RAG 单轮
4. 4. FAQ / 缓存命中
5. 5. 明确返回证据不足

降级不是失败，而是把错误从“胡说八道”变成“可预期的能力边界”。

6.4 索引版本化是生产系统的生命线

一个常见事故场景是：

• • 新文档正在入库
• • 一部分 chunk 进入了向量库
• • 图谱还没更新完
• • Agent 已经开始查询新旧混合结果

这时答案会非常不稳定。

推荐做法是显式引入 index_version：

• • draft 构建
• • shadow 验证
• • active 对外服务
• • rollback 快速回切

在线请求只查询 active 版本。

6.5 可观测性必须覆盖“检索过程”，而不是只看接口 RT

最少需要记录这些指标：

• • 请求总耗时
• • Router 分类结果
• • Planner 生成的子问题数
• • 每个工具调用耗时
• • GraphRAG 命中的种子实体
• • 图展开节点数 / 边数
• • 向量召回数 / BM25 召回数
• • 重排前后候选集变化
• • 最终上下文 token 数
• • 生成耗时
• • 引用缺失率
• • 降级率

如果这些指标不可见，RAG 系统出现错误时几乎无法定位根因。

---

七、生产级代码实战：一套更完整的 GraphRAG × Agentic RAG 实现骨架

下面的代码不是“概念伪代码”，而是按照真实工程模块拆分的骨架。
语言采用 Python 3.11 + FastAPI + LangGraph + Neo4j + Redis + Kafka，因为它更适合表达 Agent 编排与图谱检索。
如果团队主栈是 Java，也完全可以把相同分层迁移到 Spring Boot + Reactor + Redis + ES + Neo4j。

---

7.1 领域模型：先把状态定义清楚

# app/domain/models.pyfrom __future__ import annotationsfrom enum import Enumfrom pydantic import BaseModel, Fieldfrom typing import Anyclass RetrievalMode(str, Enum):    LOCAL_GRAPH = "local_graph"    GLOBAL_GRAPH = "global_graph"    HYBRID = "hybrid"    VECTOR_ONLY = "vector_only"class QueryContext(BaseModel):    tenant_id: str    user_id: str    kb_id: str    index_version: str    permission_tags: list[str] = Field(default_factory=list)    request_id: strclass Evidence(BaseModel):    source_id: str    source_type: str    content: str    score: float    metadata: dict[str, Any] = Field(default_factory=dict)class GraphSeed(BaseModel):    name: str    entity_type: str    score: floatclass SubgraphEdge(BaseModel):    source: str    target: str    relation: str    description: str | None = Noneclass RetrievalBundle(BaseModel):    mode: RetrievalMode    query: str    seeds: list[GraphSeed] = Field(default_factory=list)    edges: list[SubgraphEdge] = Field(default_factory=list)    evidences: list[Evidence] = Field(default_factory=list)    diagnostics: dict[str, Any] = Field(default_factory=dict)class AgentState(BaseModel):    original_query: str    current_query: str    context: QueryContext    plan: list[str] = Field(default_factory=list)    iteration: int = 0    max_iterations: int = 3    bundles: list[RetrievalBundle] = Field(default_factory=list)    evidence_pool: list[Evidence] = Field(default_factory=list)    seen_fingerprints: set[str] = Field(default_factory=set)    confidence: float = 0.0    final_answer: str | None = None    citations: list[dict[str, Any]] = Field(default_factory=list)    degraded: bool = False

这个模型里最重要的是三件事：

• • 把租户、版本、权限放进 QueryContext
• • 把“检索结果”和“原始证据”分开
• • 把 Agent 状态设计成可序列化对象，便于审计和回放

---

7.2 离线建图：事件驱动的增量索引管线

# app/indexing/pipeline.pyfrom __future__ import annotationsimport hashlibimport jsonfrom dataclasses import dataclassfrom typing import Iterablefrom kafka import KafkaProducer@dataclassclass DocumentEvent:    tenant_id: str    kb_id: str    doc_id: str    version: str    title: str    content: str    metadata: dictclass Chunker:    def split(self, event: DocumentEvent, size: int = 700, overlap: int = 120) -> list[dict]:        text = event.content.strip()        chunks: list[dict] = []        step = max(size - overlap, 1)        for offset in range(0, len(text), step):            chunk = text[offset: offset + size]            if not chunk:                continue            chunks.append({                "chunk_id": f"{event.doc_id}:{offset}",                "doc_id": event.doc_id,                "tenant_id": event.tenant_id,                "kb_id": event.kb_id,                "index_version": event.version,                "title": event.title,                "offset": offset,                "content": chunk,                "metadata": event.metadata,            })        return chunksclass ExtractionGateway:    async def extract(self, chunk: dict) -> dict:        raise NotImplementedErrorclass GraphIndexPublisher:    def __init__(self, brokers: list[str]) -> None:        self.producer = KafkaProducer(            bootstrap_servers=brokers,            value_serializer=lambda x: json.dumps(x, ensure_ascii=False).encode("utf-8"),        )    def publish(self, topic: str, payload: dict) -> None:        self.producer.send(topic, payload)        self.producer.flush()class IncrementalIndexPipeline:    def __init__(self, extractor: ExtractionGateway, publisher: GraphIndexPublisher) -> None:        self.chunker = Chunker()        self.extractor = extractor        self.publisher = publisher    async def process(self, event: DocumentEvent) -> None:        chunks = self.chunker.split(event)        for chunk in chunks:            extraction = await self.extractor.extract(chunk)            payload = {                "tenant_id": event.tenant_id,                "kb_id": event.kb_id,                "doc_id": event.doc_id,                "index_version": event.version,                "chunk": chunk,                "extraction": extraction,                "fingerprint": hashlib.sha256(                    f"{event.tenant_id}|{event.doc_id}|{chunk['chunk_id']}|{chunk['content']}".encode("utf-8")                ).hexdigest(),            }            self.publisher.publish("graphrag.extract.ready", payload)

这段代码体现的是“解耦”原则：

• • 文档采集和图写入分离
• • 抽取与建索引分离
• • 所有步骤天然支持失败重试和重放

对高并发系统来说，这种事件驱动比同步直写稳定得多。

---

7.3 实体归一与图谱写入：把“抽到了什么”变成“结构化可查”

# app/indexing/graph_writer.pyfrom __future__ import annotationsfrom neo4j import GraphDatabaseclass GraphWriter:    def __init__(self, uri: str, user: str, password: str) -> None:        self.driver = GraphDatabase.driver(uri, auth=(user, password))    def upsert_subgraph(        self,        tenant_id: str,        kb_id: str,        index_version: str,        doc_id: str,        entities: list[dict],        relations: list[dict],    ) -> None:        with self.driver.session() as session:            session.execute_write(                self._write_tx,                tenant_id,                kb_id,                index_version,                doc_id,                entities,                relations,            )    @staticmethod    def _write_tx(tx, tenant_id, kb_id, index_version, doc_id, entities, relations):        tx.run(            """            MERGE (d:Document {              tenant_id: $tenant_id,              kb_id: $kb_id,              index_version: $index_version,              doc_id: $doc_id            })            SET d.updated_at = datetime()            """,            tenant_id=tenant_id,            kb_id=kb_id,            index_version=index_version,            doc_id=doc_id,        )        for entity in entities:            tx.run(                """                MERGE (e:Entity {                  tenant_id: $tenant_id,                  kb_id: $kb_id,                  index_version: $index_version,                  canonical_name: $canonical_name                })                SET e.entity_type = $entity_type,                    e.aliases = $aliases,                    e.updated_at = datetime()                WITH e                MATCH (d:Document {                  tenant_id: $tenant_id,                  kb_id: $kb_id,                  index_version: $index_version,                  doc_id: $doc_id                })                MERGE (e)-[:EXTRACTED_FROM]->(d)                """,                tenant_id=tenant_id,                kb_id=kb_id,                index_version=index_version,                doc_id=doc_id,                canonical_name=entity["canonical_name"],                entity_type=entity["entity_type"],                aliases=entity.get("aliases", []),            )        for rel in relations:            tx.run(                """                MATCH (s:Entity {                  tenant_id: $tenant_id,                  kb_id: $kb_id,                  index_version: $index_version,                  canonical_name: $source                })                MATCH (t:Entity {                  tenant_id: $tenant_id,                  kb_id: $kb_id,                  index_version: $index_version,                  canonical_name: $target                })                MERGE (s)-[r:RELATION {                  tenant_id: $tenant_id,                  kb_id: $kb_id,                  index_version: $index_version,                  relation_type: $relation_type                }]->(t)                SET r.description = $description,                    r.updated_at = datetime()                """,                tenant_id=tenant_id,                kb_id=kb_id,                index_version=index_version,                source=rel["source"],                target=rel["target"],                relation_type=rel["relation_type"],                description=rel.get("description", ""),            )

这里有两个容易被忽视但很关键的点：

• • MERGE 的主键里带上了 tenant_id + kb_id + index_version
• • 文档和实体关系也被保留，后续才能回溯到原文证据

---

7.4 GraphRAG 检索服务：把“图”和“文本证据”一起返回

# app/retrieval/graphrag.pyfrom __future__ import annotationsfrom neo4j import GraphDatabasefrom app.domain.models import Evidence, GraphSeed, RetrievalBundle, RetrievalMode, SubgraphEdgeclass GraphRAGRetriever:    def __init__(self, uri: str, user: str, password: str) -> None:        self.driver = GraphDatabase.driver(uri, auth=(user, password))    def local_search(        self,        tenant_id: str,        kb_id: str,        index_version: str,        query_embedding: list[float],        top_k: int = 5,        hops: int = 2,    ) -> RetrievalBundle:        with self.driver.session() as session:            seed_rows = session.run(                """                CALL db.index.vector.queryNodes('entity_embedding_index', $top_k, $embedding)                YIELD node, score                WHERE node.tenant_id = $tenant_id                  AND node.kb_id = $kb_id                  AND node.index_version = $index_version                RETURN node.canonical_name AS name,                       node.entity_type AS entity_type,                       score                ORDER BY score DESC                """,                top_k=top_k,                embedding=query_embedding,                tenant_id=tenant_id,                kb_id=kb_id,                index_version=index_version,            ).data()            seeds = [GraphSeed(**row) for row in seed_rows]            names = [item.name for item in seeds]            if not names:                return RetrievalBundle(mode=RetrievalMode.LOCAL_GRAPH, query="")            edge_rows = session.run(                f"""                MATCH (s:Entity)                WHERE s.tenant_id = $tenant_id                  AND s.kb_id = $kb_id                  AND s.index_version = $index_version                  AND s.canonical_name IN $seed_names                OPTIONAL MATCH path = (s)-[r:RELATION*1..{hops}]-(n:Entity)                UNWIND r AS rel                RETURN s.canonical_name AS source,                       n.canonical_name AS target,                       rel.relation_type AS relation,                       rel.description AS description                LIMIT 80                """,                tenant_id=tenant_id,                kb_id=kb_id,                index_version=index_version,                seed_names=names,            ).data()            doc_rows = session.run(                """                MATCH (e:Entity)-[:EXTRACTED_FROM]->(d:Document)                WHERE e.tenant_id = $tenant_id                  AND e.kb_id = $kb_id                  AND e.index_version = $index_version                  AND e.canonical_name IN $seed_names                RETURN d.doc_id AS source_id,                       'document' AS source_type,                       coalesce(d.summary, d.doc_id) AS content,                       0.8 AS score,                       {                         tenant_id: d.tenant_id,                         kb_id: d.kb_id,                         index_version: d.index_version                       } AS metadata                LIMIT 20                """,                tenant_id=tenant_id,                kb_id=kb_id,                index_version=index_version,                seed_names=names,            ).data()        return RetrievalBundle(            mode=RetrievalMode.LOCAL_GRAPH,            query="",            seeds=seeds,            edges=[SubgraphEdge(**row) for row in edge_rows],            evidences=[Evidence(**row) for row in doc_rows],            diagnostics={"seed_count": len(seeds), "edge_count": len(edge_rows)},        )

GraphRAG 服务不要只返回边和节点，还要返回原始证据入口。
否则后面的生成器只能“看懂关系”，却没法做逐条引用。

---

7.5 检索网关：把多种检索能力统一成一个入口

# app/retrieval/gateway.pyfrom __future__ import annotationsimport asynciofrom app.domain.models import QueryContext, RetrievalBundle, RetrievalModefrom app.retrieval.graphrag import GraphRAGRetrieverclass RetrievalGateway:    def __init__(self, graph_retriever: GraphRAGRetriever, vector_retriever, embedder) -> None:        self.graph_retriever = graph_retriever        self.vector_retriever = vector_retriever        self.embedder = embedder    async def retrieve(self, query: str, mode: RetrievalMode, ctx: QueryContext) -> RetrievalBundle:        embedding = await self.embedder.embed(query)        if mode == RetrievalMode.LOCAL_GRAPH:            return await asyncio.to_thread(                self.graph_retriever.local_search,                ctx.tenant_id,                ctx.kb_id,                ctx.index_version,                embedding,            )        if mode == RetrievalMode.VECTOR_ONLY:            return await self.vector_retriever.search(query, ctx)        if mode == RetrievalMode.HYBRID:            graph_bundle, vector_bundle = await asyncio.gather(                asyncio.to_thread(                    self.graph_retriever.local_search,                    ctx.tenant_id,                    ctx.kb_id,                    ctx.index_version,                    embedding,                ),                self.vector_retriever.search(query, ctx),            )            merged = graph_bundle.model_copy(deep=True)            merged.mode = RetrievalMode.HYBRID            merged.evidences.extend(vector_bundle.evidences)            merged.diagnostics["vector_hits"] = len(vector_bundle.evidences)            return merged        return await self.vector_retriever.search(query, ctx)

检索网关的意义不只是“统一封装”，而是：

• • 给 Agent 一个稳定工具接口
• • 屏蔽底层图库、向量库、搜索引擎差异
• • 支持后续灰度接入新检索器

---

7.6 Agent 控制循环：把检索变成受控执行过程

# app/agent/workflow.pyfrom __future__ import annotationsimport hashlibfrom langgraph.graph import END, StateGraphfrom app.domain.models import AgentState, RetrievalModeasync def router_node(state: AgentState) -> AgentState:    query = state.current_query    if "影响" in query or "负责人" in query or "依赖" in query:        state.plan = [query]        state.confidence = 0.2    else:        state.plan = [query]    return stateasync def planner_node(state: AgentState) -> AgentState:    if "同时" in state.current_query or "以及" in state.current_query:        parts = [p.strip() for p in state.current_query.replace("以及", "，").split("，") if p.strip()]        state.plan = parts[:3] if parts else [state.current_query]    return stateasync def retriever_node(state: AgentState, gateway) -> AgentState:    bundles = []    for sub_query in state.plan:        mode = RetrievalMode.HYBRID if any(k in sub_query for k in ["影响", "负责人", "依赖"]) else RetrievalMode.VECTOR_ONLY        bundle = await gateway.retrieve(sub_query, mode, state.context)        bundle.query = sub_query        bundles.append(bundle)    state.bundles = bundles    for bundle in bundles:        for evidence in bundle.evidences:            fingerprint = hashlib.sha256(                f"{evidence.source_id}|{evidence.content}".encode("utf-8")            ).hexdigest()            if fingerprint in state.seen_fingerprints:                continue            state.seen_fingerprints.add(fingerprint)            state.evidence_pool.append(evidence)    state.iteration += 1    return stateasync def evaluator_node(state: AgentState) -> AgentState:    if not state.evidence_pool:        state.confidence = 0.0        return state    top_scores = sorted((item.score for item in state.evidence_pool), reverse=True)[:6]    state.confidence = sum(top_scores) / len(top_scores)    return stateasync def rewrite_node(state: AgentState) -> AgentState:    state.current_query = f"{state.original_query} 请关注事故、依赖、负责人和时间线"    state.plan = [state.current_query]    return stateasync def generator_node(state: AgentState, answer_service) -> AgentState:    state.final_answer, state.citations = await answer_service.generate(state)    return statedef route_after_evaluate(state: AgentState) -> str:    if state.confidence >= 0.70:        return "generate"    if state.iteration >= state.max_iterations:        state.degraded = True        return "generate"    return "rewrite"def build_graph(gateway, answer_service):    graph = StateGraph(AgentState)    graph.add_node("router", router_node)    graph.add_node("planner", planner_node)    graph.add_node("retriever", lambda state: retriever_node(state, gateway))    graph.add_node("evaluator", evaluator_node)    graph.add_node("rewrite", rewrite_node)    graph.add_node("generator", lambda state: generator_node(state, answer_service))    graph.set_entry_point("router")    graph.add_edge("router", "planner")    graph.add_edge("planner", "retriever")    graph.add_edge("retriever", "evaluator")    graph.add_conditional_edges(        "evaluator",        route_after_evaluate,        {            "rewrite": "rewrite",            "generate": "generator",        },    )    graph.add_edge("rewrite", "retriever")    graph.add_edge("generator", END)    return graph.compile()

这里的关键设计不是“用了 LangGraph”，而是：

• • 迭代次数有限
• • 证据池去重
• • 低置信度时补检
• • 到了上限就降级生成

生产系统里最怕的是无限补检和无上限上下文堆积。

---

7.7 生产级答案生成：必须带引用门控

# app/answering/service.pyfrom __future__ import annotationsfrom app.domain.models import AgentStateclass AnswerService:    def __init__(self, llm_client) -> None:        self.llm = llm_client    async def generate(self, state: AgentState) -> tuple[str, list[dict]]:        evidences = sorted(state.evidence_pool, key=lambda e: e.score, reverse=True)[:8]        context_lines = []        citations = []        for index, item in enumerate(evidences, start=1):            tag = f"Doc-{index}"            context_lines.append(f"[{tag}] {item.content}")            citations.append({                "tag": tag,                "source_id": item.source_id,                "source_type": item.source_type,                "score": item.score,                "metadata": item.metadata,            })        prompt = f"""你是企业知识问答系统中的回答生成器。用户问题：{state.original_query}检索证据：{chr(10).join(context_lines)}回答要求：1. 仅依据检索证据回答，不允许使用外部常识补全事实。2. 每个事实性结论都必须带引用，例如 [Doc-1]。3. 若证据不足，明确说明“检索证据不足以支持完整结论”。4. 若 state.degraded 为真，优先给出已确认部分，并说明信息边界。"""        answer = await self.llm.generate(prompt)        return answer, citations

RAG 系统一旦不做引用门控，越复杂的问题越容易“说得像真的”。
生产上，宁愿回答保守，也不要回答虚构。

---

7.8 FastAPI 接入层：统一接收、追踪、限时、返回

# app/api/server.pyfrom __future__ import annotationsimport asyncioimport timeimport uuidfrom fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelfrom app.domain.models import AgentState, QueryContextapp = FastAPI(title="GraphRAG Agentic RAG Service")class QueryRequest(BaseModel):    query: str    tenant_id: str    user_id: str    kb_id: str    permission_tags: list[str] = []class QueryResponse(BaseModel):    request_id: str    answer: str    confidence: float    degraded: bool    citations: list[dict]    latency_ms: float@app.post("/v1/query", response_model=QueryResponse)async def query(req: QueryRequest):    request_id = str(uuid.uuid4())    started = time.perf_counter()    state = AgentState(        original_query=req.query,        current_query=req.query,        context=QueryContext(            tenant_id=req.tenant_id,            user_id=req.user_id,            kb_id=req.kb_id,            index_version="active",            permission_tags=req.permission_tags,            request_id=request_id,        ),    )    try:        result = await asyncio.wait_for(app.state.agent.ainvoke(state), timeout=20)    except TimeoutError as exc:        raise HTTPException(status_code=504, detail="query timeout") from exc    except Exception as exc:        raise HTTPException(status_code=500, detail="query failed") from exc    latency_ms = (time.perf_counter() - started) * 1000    return QueryResponse(        request_id=request_id,        answer=result.final_answer or "检索证据不足以生成答案",        confidence=result.confidence,        degraded=result.degraded,        citations=result.citations,        latency_ms=latency_ms,    )

这里建议在网关外层再补：

• • 租户级限流
• • 用户级配额
• • 幂等请求键
• • Trace 注入
• • 敏感词预检查

---

7.9 缓存与限流：高并发场景不能缺

# app/runtime/guards.pyfrom __future__ import annotationsimport asyncioimport hashlibimport jsonclass AsyncSemaphoreGuard:    def __init__(self, max_concurrency: int) -> None:        self.sem = asyncio.Semaphore(max_concurrency)    async def run(self, coro):        async with self.sem:            return await coroclass QueryCache:    def __init__(self, redis_client, ttl_seconds: int = 300) -> None:        self.redis = redis_client        self.ttl_seconds = ttl_seconds    @staticmethod    def build_key(tenant_id: str, kb_id: str, query: str) -> str:        raw = f"{tenant_id}|{kb_id}|{query}"        return "rag:answer:" + hashlib.sha256(raw.encode("utf-8")).hexdigest()    async def get(self, tenant_id: str, kb_id: str, query: str):        key = self.build_key(tenant_id, kb_id, query)        data = await self.redis.get(key)        return json.loads(data) if data else None    async def set(self, tenant_id: str, kb_id: str, query: str, payload: dict):        key = self.build_key(tenant_id, kb_id, query)        await self.redis.set(key, json.dumps(payload, ensure_ascii=False), ex=self.ttl_seconds)

缓存不只是为了快，还能明显降低：

• • 模型调用成本
• • 图数据库压力
• • Agent 迭代放大的放大效应

---

八、真实案例：SaaS 支付事故分析助手如何用 GraphRAG × Agentic RAG 落地

8.1 业务背景

假设我们要做一个企业内部事故分析助手，服务对象包括：

• • SRE
• • 值班研发
• • 技术经理
• • 客服升级处理团队

知识源包括：

• • 事故复盘文档
• • 服务依赖 CMDB
• • 监控告警记录
• • 工单系统
• • 变更记录
• • 负责人映射表
• • 客户等级系统

8.2 一个典型复杂问题

“上周支付网关超时事故影响了哪些高价值客户？这些客户是否也在风控链路命中过异常？对应模块负责人和修复动作是什么？”

8.3 如果只用普通 RAG，会出现什么问题

可能会召回：

• • 支付网关事故复盘
• • 风控异常说明
• • 客户等级规则文档
• • 值班表

这些内容各自相关，但彼此之间没有被连接起来。
最终答案很可能是“似乎都对，但链路拼不起来”。

8.4 GraphRAG 在这个问题里解决了什么

GraphRAG 可以把这些对象显式连起来：

• • Incident -> impacts -> PaymentGateway
• • PaymentGateway -> affects -> CustomerGroup
• • Customer -> involved_in -> RiskAlert
• • Service -> owned_by -> Team
• • Incident -> resolved_by -> ChangeTicket

于是系统可以检索出一张真实的关系子图，而不是若干文本块。

8.5 Agentic RAG 在这个问题里解决了什么

Agent 会把问题拆成几个动作：

1. 在 GraphRAG 中查事故影响链
2. 在客户等级服务中补充高价值客户标签
3. 在风控事件库中查这些客户的关联异常
4. 在工单 / 变更系统中查修复动作与负责人
5. 汇总为最终答案

这就是“结构化上下文”和“动态任务执行”的组合。

8.6 最终回答应具备什么特征

一个生产可用的答案，至少要满足：

• • 说明影响对象
• • 说明关联路径
• • 说明负责人
• • 说明证据来源
• • 明确不知道的部分

也就是说，输出不是“看上去聪明”，而是“足够像审计材料”。

---

九、文章里的代码怎么继续升级成真正生产系统

如果团队准备从原型走向生产，建议继续补上下面这些模块。

9.1 检索评测体系

至少建设三类数据集：

• • 问答集
• • 多跳关系集
• • 无答案集

核心指标建议包括：

• • Recall@K
• • MRR
• • Citation Precision
• • Answer Groundedness
• • No-Answer Accuracy

9.2 运维治理能力

至少补齐：

• • 熔断器
• • 限流器
• • 指数退避重试
• • 模型 fallback
• • 灰度开关
• • 配置中心动态调参

9.3 索引运维能力

至少补齐：

• • 全量重建任务
• • 增量刷新任务
• • 社区摘要重算
• • 双索引切换
• • 回滚脚本
• • 数据校验任务

9.4 安全与合规

至少补齐：

• • PII 脱敏
• • 权限前置过滤
• • 查询审计日志
• • 敏感操作审批
• • Prompt 注入防护
• • 工具调用白名单

---

十、什么时候该上 GraphRAG，什么时候不该

GraphRAG 很强，但并不是所有系统都值得一开始就上。

10.1 适合上 GraphRAG 的场景

• • 知识天然有关系网络
• • 问题经常涉及多跳推理
• • 需要做全局摘要或跨域关联
• • 用户非常在意可解释性和溯源
• • 组织能承担离线索引复杂度

10.2 暂时不适合上 GraphRAG 的场景

• • 只是简单 FAQ
• • 文档量很小且结构单一
• • 团队还没有基本的 RAG 评测与治理能力
• • 连权限、元数据、版本控制都没建好

很多团队失败的原因不是 GraphRAG 不好，而是：

在基础索引治理还没成熟时，就过早上复杂结构。

正确顺序通常是：

1. 先把基础 RAG 做稳
2. 再做混合检索
3. 再做 Agent 化控制
4. 最后在关键复杂场景接入 GraphRAG

---

十一、面向架构师的最终建议：把它当成“智能检索系统”，不要只当成“模型功能”

如果把 GraphRAG × Agentic RAG 当成一个“模型功能”，你会把注意力都放在 Prompt 和模型选型上。
但如果把它看成一套“智能检索系统”，架构视角会完全不同。

你真正要设计的是：

• • 离线知识生产线
• • 在线检索决策面
• • 多工具数据访问层
• • 高并发与资源治理
• • 版本与回滚机制
• • 权限与合规体系
• • 观测与评测闭环

从这个角度看，GraphRAG 解决的是知识结构问题，Agentic RAG 解决的是执行控制问题，而生产架构解决的是稳定性问题。

三者缺一不可。

---

十二、结语：下一代企业 RAG，拼的不是“能不能答”，而是“能不能稳定、可信、可扩展地答”

普通 RAG 证明的是：

大模型可以接入企业知识。

GraphRAG 进一步证明的是：

企业知识不只是文本，更是结构。

Agentic RAG 进一步证明的是：

复杂问答不是一次检索，而是一个受控执行过程。

真正能进入核心业务的系统，最终比拼的从来不是某个单点技巧，而是整套工程能力：

• • 检索是否稳定
• • 关系是否可信
• • 过程是否可控
• • 成本是否可算
• • 故障是否可回滚
• • 回答是否可追责

如果你的系统已经从“文档问答”走向“复杂业务推理”，那么 GraphRAG × Agentic RAG 不再是锦上添花，而是架构升级的必经之路。

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】