Graphiti 与 OpenTelemetry 分布式追踪：为 AI Agent 知识图谱接入可观测性

Graphiti 与 OpenTelemetry 分布式追踪：为 AI Agent 知识图谱接入可观测性

【免费下载链接】graphitiBuild Real-Time Knowledge Graphs for AI Agents项目地址: https://gitcode.com/GitHub_Trending/grap/graphiti

导读

Graphiti 是一个面向 AI Agent 的实时知识图谱构建框架，而 OpenTelemetry 是业界标准的可观测性规范。本文基于仓库中的 OTEL_TRACING.md 文档，系统讲解如何为 Graphiti 接入 OpenTelemetry 分布式追踪：包括依赖安装、TracerProvider 与 ConsoleSpanExporter 的配置、将 tracer 传入Graphiti构造器、以及 Kuzu 内存数据库下的使用方式。读完本文，你将掌握 Graphiti 追踪的开关逻辑、span 命名规范、底层实现原理与可复现的完整示例，并了解如何把追踪数据进一步导出到 Jaeger 等后端以支撑生产级可观测性建设。

一、为什么要给 Graphiti 接入分布式追踪

Graphiti 的核心价值在于把非结构化文本、JSON 等异构数据持续转化为知识图谱：每次add_episode都会触发实体/关系抽取、节点解析、去重、边构建、属性提取等多个耗时阶段；每次search则要完成查询向量嵌入、多作用域执行、边/节点/社区多路召回与重排。这些阶段横跨 LLM 调用、Embedding 服务与图数据库操作，链路长、外部依赖多，没有追踪手段几乎无法定位性能瓶颈。

Graphiti 在 graphiti_core/tracer.py 中定义了一套薄薄的追踪抽象层，并通过 graphiti_core/graphiti.py 在Graphiti初始化时将其注入到 LLM 客户端、搜索模块等内部组件中。追踪是完全可选的能力：不传 tracer 时，内部会退化为NoOpTracer/NoOpSpan，所有 span 操作均为空实现，零开销，不影响正常功能。

二、安装依赖

Graphiti 的追踪基于 OpenTelemetry 官方 SDK 实现，需要额外安装两个包：

uv add opentelemetry-sdk

• opentelemetry-api：提供trace、Tracer、Span、StatusCode等核心 API；
• opentelemetry-sdk：提供TracerProvider、SimpleSpanProcessor、ConsoleSpanExporter等 SDK 实现，供运行时真正产生和导出 span。

Graphiti 源码对缺失依赖做了防御性处理：graphiti_core/tracer.py 中通过try: from opentelemetry.trace import ... except ImportError探测OTEL_AVAILABLE标志；若未安装 OpenTelemetry，即使显式传入 tracer，OpenTelemetryTracer的构造也会抛出ImportError并提示安装命令。在 examples/opentelemetry/pyproject.toml 中，官方示例项目声明的最低版本为opentelemetry-api>=1.20.0与opentelemetry-sdk>=1.20.0，可作为版本参考。

三、基础用法：三行配置开启追踪

接入流程分为三步：创建TracerProvider并挂载导出器 → 注册为全局 tracer provider → 把 tracer 传给Graphiti构造器。

from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor from graphiti_core import Graphiti # 1. 设置 OpenTelemetry：Provider + SpanProcessor + Exporter provider = TracerProvider() provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter())) trace.set_tracer_provider(provider) # 2. 获取 tracer 并传给 Graphiti tracer = trace.get_tracer(__name__) graphiti = Graphiti( uri="bolt://localhost:7687", # 默认使用 Neo4j，URI 为 bolt 协议 user="neo4j", password="password", tracer=tracer, trace_span_prefix="myapp.graphiti", # 可选，默认 "graphiti" )

参数说明：

Graphiti.__init__中tracer与trace_span_prefix的完整签名和语义见 graphiti_core/graphiti.py。

关于 span 前缀，底层实现有一个细节值得注意：graphiti_core/tracer.py 中OpenTelemetryTracer会执行self._span_prefix = span_prefix.rstrip('.')，自动去除前缀末尾多余的.；实际生成的 span 全名为f'{self._span_prefix}.{name}'（见 graphiti_core/tracer.py）。因此即使传入"myapp.graphiti."这类带尾点字符串，最终 span 名也不会出现.llm.generate之类的双点噪音。

四、与 Kuzu（内存图数据库）配合使用

Graphiti 的图驱动层是可插拔的，除了默认的 Neo4j，还支持 FalkorDB、Amazon Neptune 与 Kuzu。其中 Kuzu 是嵌入式内存图数据库，适合本地开发、CI 与快速原型，无需启动外部服务。

from graphiti_core.driver.kuzu_driver import KuzuDriver kuzu_driver = KuzuDriver() graphiti = Graphiti(graph_driver=kuzu_driver, tracer=tracer)

传入graph_driver后，Graphiti构造器将直接采用该驱动而跳过 Neo4j 连接（见 graphiti_core/graphiti.py）。此时无需再传uri/user/password；trace_span_prefix同样适用，例如可设为"graphiti.example"以区分示例应用与库内部调用。

五、完整可运行示例：stdout 追踪

仓库在 examples/opentelemetry/ 目录下提供了完整可运行的示例（源码见 otel_stdout_example.py），它把追踪数据输出到标准输出，适合零成本体验。其依赖声明在 examples/opentelemetry/pyproject.toml：graphiti-core（editable 本地路径引用）、kuzu>=0.11.2与 OpenTelemetry 两件套。

5.1 运行方式

uv sync export OPENAI_API_KEY=your_api_key_here uv run otel_stdout_example.py

运行前提：需要可用的 OpenAI API Key（示例默认走 OpenAI 的 LLM、Embedding 与 Reranker 客户端），并联网调用。示例中 Graphiti 使用KuzuDriver，因此无需额外启动图数据库。

5.2 追踪配置代码

from opentelemetry import trace from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor # 设置 OpenTelemetry：为服务命名并输出到 stdout resource = Resource(attributes={'service.name': 'graphiti-example'}) provider = TracerProvider(resource=resource) provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter())) trace.set_tracer_provider(provider) tracer = trace.get_tracer(__name__) graphiti = Graphiti( graph_driver=kuzu_driver, tracer=tracer, trace_span_prefix='graphiti.example', )

相比基础版，示例额外引入了Resource，为所有 span 附加service.name=graphiti-example资源属性——在导出到 Jaeger 等后端时，这是服务维度聚合与过滤的关键字段。

5.3 示例的执行流程

main()依次完成以下动作，每一步都会在控制台打印人类可读日志，同时以 span 形式输出结构化追踪信息：

1. 构建图索引与约束：await graphiti.build_indices_and_constraints()；
2. 依次写入三段 episode——两段文本（EpisodeType.text：Kamala Harris 的履历、任期）与一段结构化 JSON（EpisodeType.json：Gavin Newsom 的职务信息），均通过graphiti.add_episode(...)提交（见 otel_stdout_example.py）；
3. 执行两次语义搜索：'Who was the California Attorney General?'与'What positions has Gavin Newsom held?'，并打印前 3 条结果的fact与valid_at（时间有效性字段，体现 Graphiti 的时序知识图谱特性）；
4. 最后在finally中调用await graphiti.close()释放资源。

5.4 你能在 stdout 上看到什么

启用后，add_episode的内部流程会被包裹在一个graphiti.example.add_episodespan 中（见 graphiti_core/graphiti.py），其中包含实体抽取、节点解析、边构建等子阶段；每次 LLM 调用则产生graphiti.example.llm.generatespan（如 graphiti_core/llm_client/openai_base_client.py）。每次search会产生一个层级化的 span 树，输出形式大致为：

graphiti.example.search.execute_scopes graphiti.example.search.embed_query_vector graphiti.example.search.edge_search graphiti.example.search.edge_search.execute_methods graphiti.example.search.edge_search.rerank ...

六、底层原理：追踪抽象层与零开销设计

Graphiti 的追踪能力建立在 graphiti_core/tracer.py 中的两层抽象之上：

• Tracer/TracerSpan（抽象基类）：定义start_span(name)与add_attributes/set_status/record_exception接口；
• NoOpTracer/NoOpSpan：所有方法空实现，start_span通过@contextmanager直接 yield 一个空 span——这是“零开销”的来源；
• OpenTelemetryTracer/OpenTelemetrySpan：把 OpenTelemetry 的Span/StatusCode包装进上述接口；
• create_tracer(otel_tracer, span_prefix)：工厂函数，otel_tracer is None或 OpenTelemetry 未安装时返回NoOpTracer，否则返回OpenTelemetryTracer（见 graphiti_core/tracer.py）。

几个值得注意的实现细节：

• 防御性包装：OpenTelemetrySpan的add_attributes会自动过滤None值、把非原始类型（如列表、对象）转换为字符串，避免因属性类型非法而抛错（graphiti_core/tracer.py）；OpenTelemetryTracer.start_span在内部出错时也会退化为NoOpSpan，确保追踪故障绝不拖垮业务（graphiti_core/tracer.py）。
• 状态与异常：set_status('error'/'ok')映射到 OpenTelemetry 的StatusCode.ERROR/OK（graphiti_core/tracer.py），异常通过record_exception记录。
• 统一注入：Graphiti初始化时调用create_tracer并把结果通过self.llm_client.set_tracer(self.tracer)注入 LLM 客户端（graphiti_core/graphiti.py）；搜索模块则从clients.tracer读取，缺失时回退到NoOpTracer（见 graphiti_core/search/search.py 与_resolve_tracer）。

搜索链路中的 span 命名

搜索是追踪信息最丰富的场景。Graphiti 在 graphiti_core/search/search.py 中用_trace_phase(search_tracer, name, attributes)工具函数包裹每个阶段（search.py），该函数还会在阶段成功时set_status('ok')、失败时记录异常并set_status('error')。已观测到的核心 span 名包括：

span 上还附带了可观测属性，例如search.embed_query_vector会记录query.length与query_vector.dimension，search.execute_scopes会记录scope.edges、result.edges等（见 tests/utils/search/test_search_tracing.py）。这些属性组合前缀后即为真实输出的完整 span 名（如myapp.graphiti.search.edge_search.rerank）。

七、追踪行为测试：可观测性如何被验证

仓库中的 tests/utils/search/test_search_tracing.py 专门验证了追踪行为，可作为理解行为的“活文档”：

• test_search_emits_trace_spans_for_edge_similarity：用一个自定义RecordingTracer捕获所有 span，断言一次 cosine similarity 边检索会产生search.embed_query_vector、search.execute_scopes、search.edge_search、search.edge_search.execute_methods、search.edge_search.rerank等 span，并校验 span 属性（如query.length、query_vector.dimension、scope.edges）是否正确写入（test_search_tracing.py）；
• test_search_uses_noop_tracer_when_client_has_no_tracer与test_edge_search_uses_noop_tracer_when_none_is_passed：验证未提供 tracer 时搜索流程回退到NoOpTracer，功能不受影响（test_search_tracing.py）。

这两类测试分别覆盖了“开启追踪时的 span 正确性”与“关闭追踪时的零开销降级”两条关键路径。

八、进阶：导出到 Jaeger 等后端

ConsoleSpanExporter只适合本地调试，生产环境通常需要把 span 发送到集中式追踪后端。OpenTelemetry 生态提供了多种导出器，最常用的是OTLPSpanExporter（配合 Jaeger、Tempo、Zipkin 等支持 OTLP 的后端）。替换方式只需把ConsoleSpanExporter换成 OTLP 导出器，并选用BatchSpanProcessor以批量异步发送、降低对业务请求的阻塞：

from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.trace.export import BatchSpanProcessor provider = TracerProvider() provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(endpoint="http://jaeger:4317"))) trace.set_tracer_provider(provider)

注意：以上 OTLP 示例代码需要在项目中额外添加opentelemetry-exporter-otlp-proto-grpc等依赖，且当前仓库示例未内置该代码，请结合自身部署环境（Jaeger 的 OTLP 端口、网络可达性）调整 endpoint 地址。后续流程与基础用法完全一致——tracer = trace.get_tracer(__name__)，再传入Graphiti(...)即可。

九、总结与最佳实践

Graphiti 的 OpenTelemetry 集成可以总结为四句话：

1. 可选项，零成本：不传tracer即使用NoOpTracer，功能与性能完全不受影响，生产环境可按需灰度开启；
2. 三行接入：TracerProvider→add_span_processor→trace.set_tracer_provider，再把tracer与trace_span_prefix传给Graphiti；
3. 前缀隔离：借助trace_span_prefix为不同服务/环境（如myapp.graphiti、graphiti.example）隔离 span 命名空间，配合service.name资源属性在后端按服务聚合；
4. 先 stdout 后后端：开发阶段用ConsoleSpanExporter快速验证，生产环境切换BatchSpanProcessor+ OTLP 导出器，实现全链路可观测。

对 AI Agent 知识图谱这类“LLM 调用 + 向量检索 + 图遍历”多重外部依赖叠加的复杂系统而言，分布式追踪是定位延迟、诊断失败、理解端到端行为最直接的手段。从 OTEL_TRACING.md 出发，配合 examples/opentelemetry/otel_stdout_example.py 运行一遍，再对照 graphiti_core/tracer.py 与 tests/utils/search/test_search_tracing.py 阅读源码，你就能完整掌握 Graphiti 可观测性的接入与原理。

【免费下载链接】graphitiBuild Real-Time Knowledge Graphs for AI Agents项目地址: https://gitcode.com/GitHub_Trending/grap/graphiti