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

📘 教程 2026-07-19 约 3 分钟阅读

公司的数据散落在各种地方——数据库、Excel、API、PDF。怎么接入？

💡 你将学到

公司的数据散落在各种地方——数据库、Excel、API、PDF。怎么接入？

📜 目录

数据接入：从混乱到有序的四层架构
第一层：数据源连接
第二层：文本分块策略对比
第三层：向量化存储
第四层：AI 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()

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

策略	块大小	重叠	适用场景	召回率
固定长度	512 tokens	64	通用文档	78%
语义分块	动态	自适应	FAQ/问答	92%
递归字符分割	1000 chars	200	代码/结构化文本	85%
Markdown标题分割	按章节	0	长文档	88%

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

第三层：向量化存储

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)