Langchain-Chatchat审计日志功能实现方案

Langchain-Chatchat审计日志功能实现方案

在企业级AI应用逐步落地的今天，一个看似“智能”的问答系统是否真正可信，往往不取决于它回答得多快、多准，而在于——当出问题时，能不能说清楚：谁，在什么时候，问了什么，系统是怎么回应的，依据又来自哪份文件？

这正是合规与安全的核心诉求。尤其是在金融、医疗、法律等高敏感领域，任何脱离监管的AI行为都可能带来巨大风险。Langchain-Chatchat 作为当前主流的开源本地知识库项目，凭借其对私有文档的支持和完全离线运行的能力，已成为许多组织构建内部智能助手的首选。但“本地化”只是起点，真正的企业可用性，还得靠审计日志（Audit Logging）来兜底。

没有日志的系统就像黑箱操作——即便技术再先进，也难以通过内审或外部合规检查。本文将从实战角度出发，深入剖析如何为 Langchain-Chatchat 构建一套完整、可靠且低侵入的审计追踪体系。

---

要让每一次问答都能被追溯，我们需要解决三个关键问题：

1. 怎么捕获全过程事件？
2. 如何确保每条记录都有据可查？
3. 怎样不影响系统性能又能满足合规要求？

答案就藏在 Langchain 的回调机制、向量检索的元数据设计以及结构化日志工程实践中。

先来看最核心的一环：事件捕获。

传统的做法是在 API 接口里手动加日志打印，比如用户一提问就立刻写一条记录。但这只能拿到输入，拿不到后续的检索结果和模型输出；如果中途失败，信息就不完整。更麻烦的是，这种硬编码方式耦合度高，后期想改字段或者接入监控平台时会非常痛苦。

Langchain 提供了一个优雅的解决方案：回调系统（Callbacks）。它本质上是观察者模式的实现，允许我们在 Chain、Retriever、LLM 等组件的关键生命周期节点插入自定义逻辑，而无需改动主流程代码。

例如，我们可以注册一个BaseCallbackHandler，监听以下几个关键事件：

• on_retriever_start/on_retriever_end：获取检索开始前的问题和结束后返回的文档列表；
• on_llm_start：提取最终传给大模型的完整 Prompt；
• on_llm_end：拿到模型生成的原始回答；
• on_chain_error：记录异常堆栈，便于事后排查。

这些钩子就像是分布在整个问答流水线上的探针，把原本分散的信息串联成一条完整的操作轨迹。更重要的是，这种方式完全非侵入——你不需要动一行原有的调用逻辑，只需要在初始化 QA 链时注入回调处理器即可。

from langchain.callbacks.base import BaseCallbackHandler class AuditCallbackHandler(BaseCallbackHandler): def __init__(self, user_id: str, session_id: str = None): self.user_id = user_id self.session_id = session_id self.question = "" self.retrieved_docs = [] def on_llm_start(self, serialized, prompts, **kwargs): full_prompt = "\n".join(prompts) lines = [line.strip() for line in full_prompt.split("\n") if line.strip()] if lines: self.question = lines[-1] # 通常最后一行是用户问题 def on_retriever_end(self, documents, **kwargs): self.retrieved_docs = documents def on_llm_end(self, response, **kwargs): answer = response.generations[0][0].text.strip() log_audit_event( user_id=self.user_id, question=self.question, response=answer, source_docs=self.retrieved_docs, session_id=self.session_id )

这个处理器会在 LLM 完成响应后自动触发日志记录，整合了问题、答案和知识来源三大要素。配合 FastAPI 中间件提前解析 JWT Token 获取user_id和客户端 IP，我们就能实现身份绑定，做到“责任可界定”。

但仅仅记录文本还不够。用户可能会质疑：“你说这个结论是从文档来的，凭什么证明？”这就引出了第二个关键技术点：溯源能力。

Langchain-Chatchat 使用向量数据库（如 FAISS、Chroma）进行语义检索。它的强大之处在于能跨文档找到相关内容片段（chunk），但也带来了新的挑战：每个 chunk 是从哪一页、哪个文件切出来的？如果 metadata 没有保存好，溯源就成了空谈。

正确的做法是在文档加载阶段就注入足够的上下文信息。以 PDF 为例：

from langchain.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter loader = PyPDFLoader("policy.pdf") pages = loader.load() for i, page in enumerate(pages): page.metadata.update({ "source": "policy.pdf", "page": i + 1, "doc_type": "pdf", "category": "internal_policy" }) splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) chunks = splitter.split_documents(pages)

这样，每一个被索引的文本块都会携带源文件名、页码、分类等信息。当它在未来某次查询中被命中时，这些 metadata 也会原样返回，并可通过回调机制写入审计日志：

{ "sources": [ { "filename": "policy.pdf", "page": 42, "content_snippet": "根据公司第8.3条规定，员工请假需提前三个工作日申请..." } ] }

从此，每一条回答背后都有迹可循。不仅是合规需要，对业务本身也是一种保障——运维人员可以快速判断某条错误回答是否源于知识库质量问题，而不是模型幻觉。

有了数据采集机制，接下来就是日志本身的工程实现。这里有几个必须考虑的设计原则：

• 结构化优先：不要用纯文本日志，而是采用 JSON 格式输出单行日志，方便后续被 ELK、Splunk 或 Grafana Loki 等工具采集分析。
• 异步写入：日志 I/O 不应阻塞主流程，否则会影响用户体验。推荐使用守护线程或消息队列（如 Redis/RabbitMQ）做缓冲。
• 防篡改设计：启用追加写模式（append-only），禁止修改历史记录；设置严格文件权限（如 600），仅限管理员访问。
• 轮转与归档：使用TimedRotatingFileHandler按天分割日志，避免单个文件过大；结合压缩策略降低存储成本。

下面是一个轻量但生产可用的日志模块示例：

import logging import json import threading from datetime import datetime from logging.handlers import TimedRotatingFileHandler audit_logger = logging.getLogger("audit") handler = TimedRotatingFileHandler("logs/audit.log", when="D", interval=1, backupCount=90) handler.setFormatter(logging.Formatter('%(message)s')) audit_logger.addHandler(handler) audit_logger.setLevel(logging.INFO) audit_logger.propagate = False def log_audit_event(user_id, question, response, source_docs, session_id=None, extra=None): entry = { "timestamp": datetime.now().isoformat(), "event_type": "qa_interaction", "user_id": user_id, "session_id": session_id, "question": question, "response": response, "source_count": len(source_docs), "sources": [ { "filename": doc.metadata.get("source", "unknown"), "page": doc.metadata.get("page"), "content_snippet": doc.page_content[:200] + "..." } for doc in source_docs ], "extra": {**extra} if extra else {} } def _write(): try: audit_logger.info(json.dumps(entry, ensure_ascii=False)) except Exception as e: print(f"[ERROR] 写入审计日志失败: {e}") thread = threading.Thread(target=_write, daemon=True) thread.start()

这套机制已经在多个实际部署场景中验证过稳定性。即使在高并发环境下，也能保证日志不丢失、不阻塞主服务。

当然，任何功能都需要权衡取舍。引入审计日志不可避免地带来一些额外开销：

• 存储增长：每条交互平均约 1~5KB，按每日万次请求估算，一年约需 3.6GB 存储空间；
• 内存占用：回调处理器需缓存中间状态，建议控制会话粒度，避免长期驻留；
• 敏感信息处理：问题或回答中可能包含手机号、身份证号等 PII 数据，应在记录前做脱敏处理（如正则替换）。

为此，建议配置以下防护策略：

• 启用字段过滤规则，自动掩码匹配到的敏感词；
• 设置日志保留周期（如90天），到期自动删除；
• 禁止通过 Web 接口直接下载日志文件，防止横向渗透；
• 关键系统可对接 SIEM 平台，实现实时告警（如检测高频查询、关键词触发等异常行为）。

整个系统的协作流程如下：

1. 用户发起提问，前端携带认证 Token；
2. FastAPI 中间件解析 Token，提取user_id和ip；
3. 初始化AuditCallbackHandler并注入 QA Chain；
4. LangChain 执行过程中，回调机制逐步收集事件数据；
5. 在on_llm_end触发时汇总并异步写入审计日志；
6. 日志按天轮转，支持定期归档与审计导出。

这样的架构既保持了原有系统的简洁性，又增强了可观测性和合规能力。无论是应对 ISO 27001、等保2.0 还是 GDPR 要求，都能提供有力支撑。

回过头看，Langchain-Chatchat 的价值不仅在于“能答”，更在于“敢管”。在一个 AI 能力日益强大的时代，越智能的系统，越需要清晰的责任边界。通过回调机制实现无侵入监控，借助 metadata 实现精准溯源，再辅以结构化日志工程，我们得以构建出一个既能提效、又能问责的可信智能体。

这套方案不仅适用于 Langchain-Chatchat，也为其他基于 LangChain 的本地化 LLM 应用提供了可复用的审计框架模板。未来还可进一步扩展：比如结合用户行为分析识别潜在滥用，或将日志流式推送至 Kafka 做实时风控。

技术终将服务于治理。而一个好的企业级 AI 系统，从来不只是算法有多炫，而是它能否经得起一次严格的内部审计。