Langchain-Chatchat结合Jaeger实现分布式追踪

Langchain-Chatchat 结合 Jaeger 实现分布式追踪

在企业级 AI 应用日益复杂的今天，一个看似简单的“提问-回答”背后，往往隐藏着数十个模块的协同工作。以本地知识库问答系统为例，从 PDF 解析到文本切片、向量化、检索再到大模型生成答案，整个流程涉及多个异构组件和潜在性能瓶颈。当用户反馈“为什么这个查询要等三秒？”或“为什么这次返回空结果？”，传统的日志排查方式常常显得力不从心——我们能看到每一步发生了什么，却难以看清它们之间的关联。

这正是分布式追踪（Distributed Tracing）的价值所在。通过为一次请求赋予全局唯一的 trace ID，并贯穿其生命周期中的每一个处理阶段，我们可以构建出一张完整的调用路径图谱。而将这一能力引入像Langchain-Chatchat这样的本地化 RAG 系统中，不仅能提升调试效率，更让系统的可观测性迈上新台阶。

为什么需要在 Langchain-Chatchat 中引入追踪？

Langchain-Chatchat 是当前开源社区中最具代表性的本地知识库问答实现之一。它基于 LangChain 框架，支持文档加载、分块、嵌入、向量检索与 LLM 回答生成的完整 RAG 流程，所有数据处理均在本地完成，保障了企业敏感信息不出内网。

但随着功能扩展，系统逐渐呈现出“类微服务”的特征：虽然运行在同一进程内，但各模块职责清晰、链式调用频繁。例如：

docs = loader.load() texts = text_splitter.split_documents(docs) vectorstore.add_documents(texts) retriever.get_relevant_documents(query) llm.invoke(prompt)

每一行都可能成为延迟源头。若某次响应变慢，开发者不得不逐层打印耗时日志，手动拼接上下文，甚至重启服务加埋点才能定位问题。这种“黑盒式”调试不仅低效，还容易遗漏跨模块的异常传播。

此时，Jaeger 的介入就显得尤为关键。作为 CNCF 孵化的主流分布式追踪工具，Jaeger 提供了标准兼容、低侵入、可视化强的全链路监控能力。将其与 OpenTelemetry SDK 集成后，Langchain-Chatchat 的每一次问答请求都能自动生成结构化的 trace 数据，清晰展示各阶段耗时、错误状态及元数据标签。

更重要的是，这种集成并不改变原有业务逻辑。你依然可以使用RetrievalQA或自定义 chain，只需在关键节点添加轻量级 span 包装，即可实现无感追踪。

如何为 RAG 流水线注入可观测性？

要实现端到端追踪，核心在于利用OpenTelemetry的 context propagation 机制，在整个处理链条中传递 trace 上下文。以下是典型集成方案的关键步骤。

初始化追踪器

首先配置 OpenTelemetry 的全局 TracerProvider，并连接至 Jaeger Collector：

from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.jaeger.thrift import JaegerExporter # 设置全局 tracer provider trace.set_tracer_provider(TracerProvider()) # 配置 Jaeger exporter jaeger_exporter = JaegerExporter( agent_host_name="localhost", agent_port=6831, service_name="langchain-chatchat" ) # 添加批量处理器 span_processor = BatchSpanProcessor(jaeger_exporter) trace.get_tracer_provider().add_span_processor(span_processor) # 获取 tracer tracer = trace.get_tracer(__name__)

这里使用 UDP 协议发送数据至本地 Jaeger Agent（默认端口 6831），适合生产环境低延迟上报。开发阶段也可切换为ConsoleSpanExporter直接输出到终端验证格式。

在关键函数中创建 Span

接下来，围绕核心处理函数封装 span。每个 span 记录操作名称、属性和事件，形成可追溯的操作单元。

import time def load_document(file_path): with tracer.start_as_current_span("load_document") as span: span.set_attribute("file.path", file_path) span.set_attribute("file.type", "pdf") # 模拟实际解析耗时 time.sleep(0.5) span.add_event("Document parsed", attributes={"page.count": 12}) return ["chunk_1", "chunk_2", "chunk_3"] def vectorize_chunks(chunks): with tracer.start_as_current_span("vectorize_chunks") as span: span.set_attribute("chunk.count", len(chunks)) model_name = "BAAI/bge-small-zh" span.set_attribute("embedding.model", model_name) time.sleep(1.2) # 模拟向量化耗时 span.add_event("Embedding completed", {"vector.dim": 768}) return [f"vec_{i}" for i in range(len(chunks))] def generate_answer(context_vectors): with tracer.start_as_current_span("generate_answer") as span: span.set_attribute("llm.model", "chatglm3-6b") span.set_attribute("prompt.length", 1024) time.sleep(0.8) result = "根据公司规定，年假为5-15天不等，具体视工龄而定。" span.set_attribute("response.length", len(result)) return result

注意：with tracer.start_as_current_span()会自动继承父 span 的 trace ID，确保所有操作属于同一条调用链。同时，通过set_attribute添加结构化字段（如模型名、chunk 数量），便于后续筛选分析；add_event则用于标记关键动作点，比如“解析完成”、“检索命中”。

构建根 Span 并启动全流程

最后，在用户请求入口处创建根 span，作为整条 trace 的起点：

if __name__ == "__main__": query = "公司年假政策是怎么规定的？" with tracer.start_as_current_span("qa_request") as root_span: root_span.set_attribute("user.query", query) root_span.set_attribute("request.id", "req-12345") # 可结合外部 requestId # 执行流水线 docs = load_document("hr_policy.pdf") vectors = vectorize_chunks(docs) answer = generate_answer(vectors) print("Answer:", answer)

运行后，该请求的所有 span 将被收集并发送至 Jaeger 后端，最终在 UI 中呈现如下视图：

TRACE: abc123xyz └── qa_request (2.5s) ├── load_document (0.5s) │ └── Event: Document parsed (page.count=12) ├── vectorize_chunks (1.2s) │ └── Event: Embedding completed (vector.dim=768) └── generate_answer (0.8s) └── Attributes: response.length=48

一目了然地看出：向量化是主要耗时环节，占总时间近一半。如果未来考虑优化性能，优先替换高效嵌入模型或启用缓存将是合理选择。

实际架构与部署建议

在一个典型的生产环境中，完整的追踪链路应包含以下组件：

graph TD A[用户请求] --> B[Langchain-Chatchat] B --> C[OpenTelemetry SDK] C --> D[Jaeger Agent] D --> E[Jaeger Collector] E --> F[(Storage: Elasticsearch)] F --> G[Jaeger UI]

• OpenTelemetry SDK：嵌入应用代码，负责生成和导出 trace。
• Jaeger Agent：以 DaemonSet 形式运行在主机上，接收 UDP 数据，减少网络开销。
• Collector：接收 agent 转发的数据，进行校验、采样、打标后写入存储。
• Elasticsearch：持久化 trace 数据，支持按服务、操作名、标签快速检索。
• Jaeger UI：提供图形化界面，查看调用拓扑、时间轴、延迟分布等。

采样策略：平衡细节与性能

全量采集 trace 会对系统造成显著负担，尤其在高并发场景下。因此需合理配置采样率：

from opentelemetry.sdk.trace.sampling import TraceIdRatioBasedSampler # 仅采样 5% 的请求 sampler = TraceIdRatioBasedSampler(0.05) trace.set_tracer_provider(TracerProvider(sampler=sampler))

开发环境可设为AlwaysOnSampler全开，线上则推荐 1%-10% 的比例。对于特定调试需求，也可通过 HTTP header 注入采样指令（如x-debug-trace: true）临时开启。

安全与合规注意事项

尽管追踪极大提升了运维能力，但也带来新的风险点：

• 避免记录敏感内容：不要将原始文档文本、用户身份、API Key 等写入 span attribute 或 event。
• 脱敏处理：对 query 内容做摘要或哈希处理后再记录，例如：
  python span.set_attribute("query.md5", hashlib.md5(query.encode()).hexdigest())
• 访问控制：Jaeger UI 应限制访问权限，仅允许授权人员查询 trace。
• 数据保留周期：设置索引 TTL（如 7 天），防止审计数据无限增长。

此外，建议在日志中输出当前 trace ID，实现 logs-traces 联查：

from opentelemetry import context from opentelemetry.trace import get_current_span current_span = get_current_span() trace_id = current_span.get_span_context().trace_id print(f"[trace_id={trace_id:x}] Starting document parsing...")

这样可在 ELK 中通过 trace ID 关联日志与追踪，进一步增强排错能力。

它解决了哪些真实痛点？

这套组合拳在实际项目中已展现出明确价值：

🔍 快速定位性能瓶颈

某次用户反馈“上传合同后查询极慢”。通过 Jaeger 发现，“PDF 解析”平均耗时达 4.2 秒，远高于其他环节。深入分析发现是因 PyPDF2 对扫描件 OCR 支持差导致重试堆积。随后切换至pdfplumber+Tesseract方案，耗时降至 0.9 秒。

🛠️ 捕获静默失败

有段时间部分文档检索结果为空，但日志无报错。启用 tracing 后发现，“文本分块”阶段因正则表达式匹配异常提前退出，且未抛出异常。通过 span 中的 error 标记快速锁定问题函数，并修复边界条件判断逻辑。

⚖️ 支持 A/B 测试对比

团队想评估“语义分块 vs 固定长度分块”的效果差异。通过在同一 trace 中并行执行两种策略，并分别打上strategy=semantic和strategy=fixed标签，直接在 Jaeger 中比较两者的耗时与下游召回率，辅助决策。

📜 满足审计要求

金融客户要求保留每次查询的操作轨迹。借助 trace 记录的完整路径、时间戳与参数，系统可输出符合 GDPR/SOC2 规范的审计报告，证明数据处理过程透明可控。

不止于“能用”，更要“可信、可观、可控”

Langchain-Chatchat 的本质是一个高度模块化的 RAG 引擎，它的优势在于灵活性与安全性——你可以自由替换嵌入模型、向量库、分词器，所有数据留在本地。但这同时也带来了复杂性管理的挑战。

而 Jaeger 的加入，正是为了应对这种复杂性。它不改变系统功能，却深刻改变了我们理解和维护系统的方式。从前我们依赖经验猜测哪里慢、哪里错；现在我们可以基于事实做出判断。

更重要的是，这种集成方式具有很强的通用性。无论是基于 FastAPI 的 Web 接口层，还是 Celery 异步任务队列，只要遵循 OpenTelemetry 标准，就能无缝接入同一套追踪体系。未来若迁移到 Kubernetes 环境，也能平滑过渡为服务级 tracing。

所以，这不仅仅是一次技术叠加，而是从“可用系统”迈向“可信赖系统”的重要一步。在一个越来越强调 AI 可解释性与工程落地能力的时代，让每一次推理都有迹可循，或许才是企业真正愿意长期投入的关键所在。