AI Agent数据接入：让Agent读取你的各种数据源

数据接入：从混乱到有序的四层架构

第一层：数据源连接

from langchain_community.document_loaders import (
    TextLoader, PyPDFLoader, CSVLoader, WebBaseLoader
)

def load_documents(source_type: str, source_path: str):
    """统一的文档加载入口"""
    loaders = {
        'txt': TextLoader(source_path, encoding='utf-8'),
        'pdf': PyPDFLoader(source_path),
        'csv': CSVLoader(source_path),
        'web': WebBaseLoader(source_path),
    }
    loader = loaders.get(source_type)
    if not loader:
        raise ValueError(f"不支持的数据源类型: {source_type}")
    return loader.load()

第二层：文本分块策略对比

推荐策略：用递归字符分割作为默认方案，对准确率要求高的场景切换到语义分块。

第三层：向量化存储

from langchain_chroma import Chroma
from langchain_huggingface import HuggingFaceEmbeddings

# 中文向量模型（免费、离线可用）
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-large-zh-v1.5",
    model_kwargs={'device': 'cpu'},
    encode_kwargs={'normalize_embeddings': True}
)

# 创建向量库
vector_store = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 检索测试
results = vector_store.similarity_search_with_score("退货流程是什么？", k=3)
for doc, score in results:
    print(f"相关度: {score:.4f} | 内容: {doc.page_content[:100]}")

第四层：AI Agent检索

def rag_query(query: str, k: int = 3) -> str:
    """RAG检索+生成"""
    docs = vector_store.similarity_search(query, k=k)
    context = "\n---\n".join([d.page_content for d in docs])

    prompt = f"""基于以下资料回答问题。如果资料中没有相关信息，请明确说"找不到相关信息，请补充资料"。

资料：
{context}

问题：{query}
回答："""
    return llm.invoke(prompt)

实施步骤

第一步：盘点现有数据源，分类（数据库/文档/API/网页）。

第二步：对所有文档建立统一目录索引，打上标签（部门/类型/更新时间）。

第三步：部署向量数据库（推荐Chroma，零配置入门），嵌入bge-large-zh-v1.5。

第四步：写Python脚本定时同步（新文档入库、过期文档标记），设置每日增量更新cron。

第五步：上线RAG功能后持续监控——记录"未找到"率，如果>10%说明知识库覆盖不足，需补充资料。