图序列化对RAG成本精度的影响

Agent Lab Journal

Guides Glossary

更新于 2026 年 7 月 31 日 · 45 分钟阅读 · 可重现的基准测试

一个图可能包含完全正确的证据，但仍然产生错误的答案，因为其文本表示占用了上下文窗口、分离了相关事实或使边方向变得模糊。本指南将这个失败模式转变为可重复的实验：以多种方式序列化一个图，测量发送给模型的实际 token，在涉及模型之前测试确定性遍历，然后在固定的上下文预算下对多跳答案进行评分。

格式是检索系统的一部分

格式是检索系统的一部分

GraphRAG 将检索与图结构相结合，使答案可以遵循关系，而不是仅依赖文本相似度。典型的请求检索一个子图、序列化它，然后将该文本放在模型提示中。因此，序列化器位于检索和推理之间的关键路径上。

图数据库可能存储紧凑的标识符和类型化的关系，但提示通常会将它们展开为重复的字段名：

{
  "source": {"id": "svc-api", "type": "service"},
  "relationship": {"type": "depends_on"},
  "target": {"id": "svc-auth", "type": "service"}
}

重复的键、大括号、引号、逗号、类型标签和重复的节点属性都会消耗 token。这种语法不是自动浪费：显式结构可以防止歧义。工程问题是每个 token 是否足以改善正确遍历的概率，从而证明其成本合理。

更小的表示也可能失败。"A B C"这行很便宜，但它没有说明 A 是否依赖于 B、拥有 B 还是替代了 B。因此，有用的目标不是最少 token。它是在固定的上下文窗口、延迟目标和预算下的最佳测量答案质量。

三个不同的精度层

不要将每个错误都归结为"模型搞错了"。独立测量这些层：

序列化保真度：解析器能否重构原始节点、边、方向、标签和必需的属性？

序列化保真度：解析器能否重构原始节点、边、方向、标签和必需的属性？

遍历保真度：重构的图能否正确回答确定性路径查询？

遍历保真度：重构的图能否正确回答确定性路径查询？

答案质量：语言模型能否使用序列化证据来生成预期的多跳答案？

答案质量：语言模型能否使用序列化证据来生成预期的多跳答案？

如果第一层失败，改变提示无法修复缺失的信息。如果第一层和第二层通过但第三层失败，顺序、冗余度、模型行为或指令是更合理的原因。

具体案例：依赖关系影响分析

具体案例：依赖关系影响分析

考虑一个内部服务图。节点代表服务和数据库。有向的 depends_on 边从消费者指向其依赖项。第二个关系 owned_by 将服务连接到其团队。

操作问题是：

如果 db-identity 失败，哪些面向客户的服务会受到传递影响，这些服务由哪些团队拥有？

回答这个问题至少需要两个操作：从失败的数据库反向遍历依赖边，然后从受影响的服务跟随所有权边。删除方向、省略关系类型或丢失所有权记录的表示无法支持答案，即使每条剩余的行在语法上都有效。

这也是截断变得危险的地方。大型 JSON 文档可能会将关闭边或所有权部分放在可用预算之外。提示仍然可读，但最后一跳所需的证据不存在。

下面的基准比较了同一有向、类型化图的五种无损表示。所有格式都保留节点标识符、节点类型、边类型和方向。

格式	示例边	预期的权衡
Pretty JSON	{"source":"api", "relation":"depends_on", "target":"auth"}	显式且易于检查；缩进和重复的键增加了大小。
Compact JSON	{"s":"api","r":"depends_on","t":"auth"}	保留机器可读结构，键更短且没有空格。
JSONL	{"k":"e","s":"api","r":"depends_on","t":"auth"}	可流式处理且独立可解析的记录；键仍然重复。
CSV 记录	E,api,depends_on,auth	紧凑且传统；当值包含分隔符时引用规则很重要。
类型化边列表	api -depends_on-> auth	可读且紧凑；需要声明的语法和转义策略。

不要在不标记的情况下在主排名中包含有意无损格式。例如，无类型对列表可能仅因为删除了问题所需的信息而在 token 上获胜。

对每种格式保持以下固定：

图和提供记录的顺序；

图和提供记录的顺序；

查询集和预期答案；

查询集和预期答案；

tokenizer 和模型版本；

tokenizer 和模型版本；

提示指令和输出模式；

提示指令和输出模式；

最大输入预算和截断策略；

最大输入预算和截断策略；

生成设置，包括温度和最大输出 token；

生成设置，包括温度和最大输出 token；

重复运行次数和评分代码。

重复运行次数和评分代码。

指标	说明
序列化字节	UTF-8 字节长度。适用于存储和传输，但不能替代模型特定的 token 测量。
Token 计数	由用于评估的确切 tokenizer 生成的计数。
往返保真度	解码表示是否得到规范的节点和边集。
遍历精度	解码答案等于黄金答案的确定性图查询的比例。
预算生存率	应用相同 token 预算后剩余的必需证据记录的比例。
多跳精确匹配	标准化后的模型答案等于预期标识符集的比例。

推荐假设

在运行测试前写出假设。例如：

Compact JSON 将比 Pretty JSON 使用更少的 token，同时保留相同的确定性结果。

Compact JSON 将比 Pretty JSON 使用更少的 token，同时保留相同的确定性结果。

类型化边列表将减少语法开销，但可能需要更强的格式指令。

类型化边列表将减少语法开销，但可能需要更强的格式指令。

在严格的预算下，记录感知截断将优于原始字符或 token 切片。

在严格的预算下，记录感知截断将优于原始字符或 token 切片。

按与查询的相关性对边进行排序将更重要，因为预算变得更小。

按与查询的相关性对边进行排序将更重要，因为预算变得更小。

这些是假设，不是结果。基准设计用于在您的图、tokenizer、模型和查询分布上确认或拒绝它们。

参考实现

参考实现

以下单文件程序创建确定性合成图，以五种格式序列化它，解码每种格式，验证往返、运行可达性查询并测量 token。默认图足够小可以检查。增加 --services 和 --noise-edges 来进行上下文压力测试。

python -m venv .venv
. .venv/bin/activate
python -m pip install tiktoken

在选择在实验中使用的版本后固定安装的依赖项：

python -m pip freeze > requirements-lock.txt
python --version
python -m pip show tiktoken

另存为 benchmark_graph_formats.py

#!/usr/bin/env python3
import argparse
import csv
import io
import json
import random
from collections import defaultdict, deque
from dataclasses import dataclass, asdict

import tiktoken

@dataclass(frozen=True, order=True)
class Node:
    id: str
    type: str

@dataclass(frozen=True, order=True)
class Edge:
    source: str
    relation: str
    target: str

def make_graph(service_count, noise_edges, seed):
    rng = random.Random(seed)

nodes = [
        Node("db-identity", "database"),
        Node("team-core", "team"),
        Node("team-growth", "team"),
        Node("svc-web", "service"),
        Node("svc-api", "service"),
        Node("svc-auth", "service"),
        Node("svc-profile", "service"),
    ]
    edges = [
        Edge("svc-web", "depends_on", "svc-api"),
        Edge("svc-api", "depends_on", "svc-auth"),
        Edge("svc-profile", "depends_on", "svc-auth"),
        Edge("svc-auth", "depends_on", "db-identity"),
        Edge("svc-web", "owned_by", "team-growth"),
        Edge("svc-api", "owned_by", "team-core"),
        Edge("svc-auth", "owned_by", "team-core"),
        Edge("svc-profile", "owned_by", "team-growth"),
    ]

extra_services = [
        Node(f"svc-noise-{i:04d}", "service")
        for i in range(max(0, service_count - 4))
    ]
    nodes.extend(extra_services)

all_services = [n.id for n in nodes if n.type == "service"]
    existing = set(edges)
    attempts = 0
    while len(existing) < len(edges) + noise_edges and attempts < noise_edges * 50 + 100:
        attempts += 1
        source = rng.choice(all_services)
        target = rng.choice(all_services)
        if source != target:
            existing.add(Edge(source, "depends_on", target))

edges = sorted(existing)
    return sorted(nodes), edges

def canonical(nodes, edges):
    return (
        tuple(sorted((n.id, n.type) for n in nodes)),
        tuple(sorted((e.source, e.relation, e.target) for e in edges)),
    )

def encode_pretty_json(nodes, edges):
    value = {
        "nodes": [asdict(n) for n in nodes],
        "edges": [asdict(e) for e in edges],
    }
    return json.dumps(value, indent=2, ensure_ascii=False)

def decode_pretty_json(text):
    value = json.loads(text)
    nodes = [Node(x["id"], x["type"]) for x in value["nodes"]]
    edges = [
        Edge(x["source"], x["relation"], x["target"])
        for x in value["edges"]
    ]
    return nodes, edges

def encode_compact_json(nodes, edges):
    value = {
        "n": [[n.id, n.type] for n in nodes],
        "e": [[e.source, e.relation, e.target] for e in edges],
    }
    return json.dumps(value, separators=(",", ":"), ensure_ascii=False)

def decode_compact_json(text):
    value = json.loads(text)
    nodes = [Node(*x) for x in value["n"]]
    edges = [Edge(*x) for x in value["e"]]
    return nodes, edges

def encode_jsonl(nodes, edges):
    records = []
    records.extend(
        json.dumps(
            {"k": "n", "id": n.id, "t": n.type},
            separators=(",", ":"),
            ensure_ascii=False,
        )
        for n in nodes
    )
    records.extend(
        json.dumps(
            {"k": "e", "s": e.source, "r": e.relation, "t": e.target},
            separators=(",", ":"),
            ensure_ascii=False,
        )
        for e in edges
    )
    return "\n".join(records)

def decode_jsonl(text):
    nodes, edges = [], []
    for line in text.splitlines():
        if not line.strip():
            continue
        value = json.loads(line)
        if value["k"] == "n":
            nodes.append(Node(value["id"], value["t"]))
        elif value["k"] == "e":
            edges.append(Edge(value["s"], value["r"], value["t"]))
        else:
            raise ValueError(f"Unknown record kind: {value['k']}")
    return nodes, edges

def encode_csv_records(nodes, edges):
    output = io.StringIO(newline="")
    writer = csv.writer(output, lineterminator="\n")
    writer.writerow(["kind", "a", "b", "c"])
    for n in nodes:
        writer.writerow([

基准测试为什么从确定性解析开始

模型不是解析器一致性测试。在花费推理预算之前，要先证明每个声称无损的格式都能重建同一张图。该程序比较的是规范元组而不是序列化字符串，所以无害的排序和空格差异不计为失败。

查询套件随后在解码后的图上运行。如果遍历评分低于1.0，说明序列化或解码器改变了可用的图语义。在这里停止并修复表示方式，然后再评估语言模型。

运行基准测试

python benchmark_graph_formats.py \
  --services 40 \
  --noise-edges 80 \
  --seed 7 \
  --encoding o200k_base \
  --output baseline.json

for size in 40 100 250 500
do
  python benchmark_graph_formats.py \
    --services "$size" \
    --noise-edges "$((size * 2))" \
    --seed 7 \
    --encoding o200k_base \
    --output "results-${size}.json"
done

使用你所选模型所需的编码。不要仅因为在这个例子中看到某个编码名称就复制它。将其与模型标识符和依赖锁定文件一起记录。

检查生成的表示

wc -c graph.*.txt
sed -n '1,30p' graph.pretty_json.txt
sed -n '1,30p' graph.typed_edges.txt
python -m json.tool baseline.json

记录感知的预算测试

原始令牌切片可能会把JSON对象或CSV记录切成两半。这测量的是损坏程度而不是紧凑程度。运行两种策略并分别报告：

原始预算：编码完整表示，保留前B个令牌

记录感知预算：追加完整记录直到下一条记录仍能容纳。对于整体JSON，构建最大的适配有效对象。

使用多个预算而不是一个方便的阈值：

budgets = [512, 1024, 2048, 4096, 8192]

对于每个预算，存储：

• 截断表示是否解析成功；
• 保留的节点和边数量；
• 每条金标准路径是否仍然存在；
• 确定性遍历准确度；
• 模型答案评分（若启用了模型评估）。

测试多种排序方式

序列化格式和记录顺序相互影响。至少比较以下几种：

• 规范标识符顺序；
• 使用记录种子的确定性打乱顺序；
• 查询相关边优先；
• 从查询实体出发的广度优先顺序；
• 节点后跟边，以及在支持的地方交错的节点-边记录。

一种格式不应该获得其他格式被否定的排序优势。要么在任何地方使用等价的顺序，要么将排序视为单独的实验因素。

多跳模型评估

在所有格式通过确定性检查后，测试模型是否能使用它们。多跳答案应该依赖于多个链接的事实，而不是直接揭示结果的标签。

You answer only from the graph data below.

Graph semantics:
- "A depends_on B" means A is a consumer of B.
- If B fails, A may be affected.
- Dependency impact propagates transitively in the reverse direction.
- "A owned_by T" identifies the team responsible for A.

Return JSON only:
{
  "affected_services": ["service-id"],
  "owners": {
    "service-id": ["team-id"]
  }
}

Question:
If db-identity fails, which services are transitively affected,
and which teams own those services?

GRAPH_FORMAT:
{{format_name}}

GRAPH_DATA:
{{serialized_graph}}

在除了format_name和serialized_graph外的格式之间逐字保持这些说明完全相同。如果某种格式需要语法解释，将语法包含在其令牌计数中。否则，紧凑的自定义格式通过将说明隐藏在测量输入之外会获得不公平的优势。

使用确定性图代码而不是另一个模型来生成预期答案。在比较前规范化金标准和候选输出：

• 拒绝缺失的必需字段；
• 去重并排序标识符数组；
• 精确比较标识符而不使用语义相似度；
• 分别为受影响的服务和所有权映射评分。

对于集合值答案，报告精度、召回率和F1。设P为预测的标识符集，G为金标准集：

precision = |P ∩ G| / |P|
recall    = |P ∩ G| / |G|
F1        = 2 × precision × recall / (precision + recall)

在运行实验前定义空集的行为。也报告严格精确匹配，因为F1评分可能会隐藏一个长正确列表中的一个不受支持的服务。

即使低温度生成也可能出现变异。对每种格式和查询使用相同次数的尝试。存储每个原始请求、原始响应、解析结果、规范化答案、延迟和供应商返回的使用值。当 API 已经提供 token 使用量时，不要之后再重新计算。

{
  "run_id": "locally-assigned-id",
  "graph_hash": "sha256-of-canonical-graph",
  "format": "typed_edges",
  "ordering": "bfs_from_query",
  "budget_tokens": 4096,
  "model": "exact-model-identifier",
  "tokenizer": "exact-tokenizer-identifier",
  "prompt_tokens_reported": 0,
  "completion_tokens_reported": 0,
  "parse_success": false,
  "exact_match": false,
  "affected_services_f1": 0.0,
  "owners_exact_match": false,
  "latency_ms": 0,
  "raw_response_path": "responses/run-id.txt"
}

上面的零值是定义字段的占位符，不是基准测试结果。

首先存储 token 使用量。之后根据显式记录的价格表应用定价：

input_cost =
  prompt_tokens / 1_000_000 * input_price_per_million

output_cost =
  completion_tokens / 1_000_000 * output_price_per_million

total_cost = input_cost + output_cost

定价会变化，缓存输入规则也可能不同。保持原始使用量与价格假设分离，使历史运行可以在不重写证据的情况下进行比较。

验证和解释

预检验证

每种格式都有 round_trip_equal：true。

每种格式都有 round_trip_equal：true。

每种未截断的格式的遍历准确度都是 1.0。

每种未截断的格式的遍历准确度都是 1.0。

相同的规范图哈希出现在每次运行中。

相同的规范图哈希出现在每次运行中。

分词器与评估的模型相对应。

分词器与评估的模型相对应。

语法指令包含在测量的提示 token 中。

语法指令包含在测量的提示 token 中。

所有随机种子、包版本、预算和模型设置都被记录。

所有随机种子、包版本、预算和模型设置都被记录。

金标准答案来自确定性遍历。

金标准答案来自确定性遍历。

使用帕累托比较

单一加权分数可能隐藏重要的权衡。首先确定帕累托前沿：如果另一种格式在所有测试预算下使用不超过当前格式的 token，并且在每个预算上实现相等或更好的准确度（至少有一项是严格改进），则该格式被支配。

然后在非支配格式中使用操作约束进行选择。例如，团队可能愿意为标准解析、更好的可观测性和更少的转义 bug 接受适度的 token 增加。

确定性 token 计数不需要置信区间，但模型答案分数需要。报告查询数和重复试验次数。对于精确匹配等比例，使用合适的二项式区间或引导查询级结果。避免将一个成功答案的差异呈现为一般格式优势。

分离这些结论

格式效率：完整图所需的 token。

格式效率：完整图所需的 token。

预算健壮性：图受约束时的答案质量。

预算健壮性：图受约束时的答案质量。

推理可用性：当所有必要证据都适合时的质量。

推理可用性：当所有必要证据都适合时的质量。

工程可靠性：解析失败、转义错误、schema 漂移和调试成本。

工程可靠性：解析失败、转义错误、schema 漂移和调试成本。

最小结果表

	格式	Token	往返相等	遍历准确度	预算	解析率	多跳精确匹配
Pretty JSON	本地测量	本地测量	本地测量	记录值	本地测量	本地测量
Compact JSON	本地测量	本地测量	本地测量	记录值	本地测量	本地测量
JSONL	本地测量	本地测量	本地测量	记录值	本地测量	本地测量
CSV	本地测量	本地测量	本地测量	记录值	本地测量	本地测量
Typed edge list	本地测量	本地测量	本地测量	记录值	本地测量	本地测量

值得显式测试的失败案例

重复标识符

如果两条节点记录共享一个标识符但在类型或属性上不同，应决定解码是拒绝图还是应用记录的合并规则。静默的后写入优先行为可能导致格式顺序改变答案。

自定义边列表通常工作良好，直到标识符包含管道符、制表符、换行符、反斜杠或类似箭头的序列。添加对抗性测试用例并验证往返测试。没有转义规范的紧凑语法不是生产格式。

关系方向歧义

depends_on B 和 B is_dependency_of A 从不同的角度描述相同的关系。混合两种约定会反转影响分析。将语义放在提示中，并明确编码方向。

关系类型坍缩

如果 depends_on、owned_by 和 deployed_in 被序列化为未标记的对，确定性可达性可能返回语法连接但语义无效的路径。

边可能引用被检索或截断遗漏的节点。应决定模型是否需要节点记录、标识符是否足够，以及解码器是拒绝还是保留该边。

断裂的单体 JSON

在 token 边界处切割 JSON 文档通常会产生无效的语法。支持记录的构造或 JSONL 可以保留