Langchain-Chatchat结合Airflow实现ETL流程

魔都财观

653人浏览 · 2025-12-19 09:08:16

魔都财观 · 2025-12-19 09:08:16 发布

Langchain-Chatchat 结合 Airflow 实现 ETL 流程

在企业智能化转型的浪潮中，知识管理正从“能查”迈向“会想”。越来越多组织希望构建基于私有文档的智能问答系统，让员工像问人一样快速获取内部政策、技术手册或合规指南。然而，一个真正可用的知识助手，不仅要有理解语言的能力，更需要一套稳定、自动、可追溯的数据更新机制。

Langchain-Chatchat 作为当前最受欢迎的本地化知识库开源框架之一，已经解决了“如何用大模型读懂私有文档”的问题。它支持将 PDF、Word 等文件转化为语义向量，并在离线环境中完成高质量问答，极大提升了数据安全性。但现实中的挑战远不止推理——文档每天都在变，制度更新、产品迭代、流程优化层出不穷。如果每次都要人工重新导入一遍，系统的实用价值就会大打折扣。

这时候，就需要引入工程化的思维：把知识库的维护变成一条自动化流水线。Apache Airflow 正是为此而生。作为业界主流的工作流调度平台，Airflow 能够精确控制任务依赖、实现失败重试、提供可视化监控，完美契合 ETL（Extract, Transform, Load）场景的需求。

当 Langchain-Chatchat 遇上 Airflow，我们得到的不再只是一个能回答问题的原型系统，而是一个真正可以长期运行、持续进化的企业级知识中枢。

核心架构设计：从手动脚本到生产级流水线

传统做法往往是写个 Python 脚本跑一次处理流程，然后靠 crontab 定时执行。这种方式初期简单直接，但随着文档数量增多、格式复杂化、错误处理需求上升，很快就会暴露出问题：任务失败无告警、状态不可追踪、无法判断是否重复处理、缺乏版本管理和协作能力。

我们需要的是一个具备以下特征的系统：

自动化触发：无需人工干预，定期检查并处理新文档。
流程可拆解：各环节职责分明，便于调试和扩展。
执行可观测：每一步都记录日志，失败时能快速定位原因。
结果可验证：确保每一次更新都是有效的、一致的。

这正是 Airflow 的强项。通过定义 DAG（有向无环图），我们将整个知识库构建过程拆分为多个原子任务，形成一条清晰的数据管道：

[检测新文档] 
   ↓
[解析文本内容]
   ↓
[清洗与分块]
   ↓
[生成向量嵌入]
   ↓
[写入向量数据库]
   ↓
[通知更新完成]

每个节点都可以独立配置重试策略、超时时间、资源限制，甚至跨服务器执行。更重要的是，所有任务的状态都能在 Web UI 中一目了然地查看，运维人员不再需要登录服务器翻日志。

Langchain-Chatchat 如何支撑本地知识服务

Langchain-Chatchat 并非简单的 RAG 工具链封装，而是针对中文企业场景深度优化的一体化解决方案。它的核心优势在于“全链路本地化”：从文档解析到答案生成，全程不依赖任何外部 API。

多模态文档解析能力

系统内置了对常见办公格式的支持：
- PyPDF2 / pdfplumber 解析 PDF，保留表格结构
- docx2txt 和 python-docx 处理 Word 文档
- unstructured 支持扫描件 OCR（需集成 Tesseract）
- Markdown、TXT 直接读取

这些解析器统一由 LangChain 的 DocumentLoader 接口抽象，使得后续流程无需关心原始格式。

中文友好的文本分块策略

长文本切片是影响检索质量的关键步骤。若按固定字符数切割，容易切断句子语义；若不分块，则超出模型上下文窗口。Chatchat 采用递归式分块（RecursiveCharacterTextSplitter），优先按照中文标点符号（如句号、问号、换行符）进行分割，尽可能保持段落完整性。

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "！", "？", " ", ""]
)

这种策略特别适合处理规章制度、操作手册等逻辑性强的文本，避免出现“上一句没说完，下一句已跳题”的尴尬情况。

向量化与高效检索

文本块经 BGE、m3e 或 Sentence-BERT 类中文嵌入模型编码后，存入 FAISS 或 Chroma 向量数据库。FAISS 因其轻量级、高查询速度，常用于单机部署场景；而 Chroma 更适合需要持久化和多客户端访问的环境。

关键在于，向量索引必须支持增量更新。否则每次新增一份文档就要重建整个库，成本过高。幸运的是，FAISS 提供了 add_embeddings 方法，允许动态追加新向量，配合 HNSW 索引还能保持近似搜索性能。

Airflow 构建健壮的 ETL 流水线

如果说 Langchain-Chatchat 是大脑，那么 Airflow 就是神经系统，负责感知变化、协调动作、反馈状态。

下面是一段典型的 DAG 定义，展示了如何将知识库更新流程工程化：

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
import os

def check_for_new_docs():
    doc_dir = "/data/docs/"
    processed_log = "/data/logs/processed.log"
    new_files = []

    with open(processed_log, 'r') as f:
        processed = set(f.read().splitlines())

    for filename in os.listdir(doc_dir):
        if filename.endswith(('.pdf', '.txt', '.docx')) and filename not in processed:
            new_files.append(filename)

    return new_files  # 返回给下游任务

def process_document(ti):
    new_files = ti.xcom_pull(task_ids='check_new_docs')
    for file in new_files:
        result = os.system(f"python /app/process_doc.py --file {file}")
        if result != 0:
            raise RuntimeError(f"Failed to process {file}")
        # 记录已处理
        with open("/data/logs/processed.log", "a") as f:
            f.write(file + "\n")

default_args = {
    'owner': 'ai-team',
    'depends_on_past': False,
    'start_date': datetime(2025, 4, 1),
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'langchain_chatchat_etl',
    default_args=default_args,
    description='自动更新本地知识库',
    schedule_interval='0 2 * * *',  # 每日凌晨两点执行
    catchup=False,
    tags=['knowledge-base', 'etl'],
)

t1 = PythonOperator(
    task_id='check_new_docs',
    python_callable=check_for_new_docs,
    dag=dag,
)

t2 = PythonOperator(
    task_id='process_documents',
    python_callable=process_document,
    dag=dag,
)

t1 >> t2

这段代码看似简单，实则蕴含了几个关键设计思想：