pandas 数据处理：AI 项目的数据预处理实战

AI 项目中有一个规律几乎从不失效：70% 的时间花在数据上，30% 的时间花在模型上。无论是准备训练数据、预处理 RAG 文档、还是分析 LLM 输出日志，最终都要落到一张表格上：读取、清洗、过滤、统计、导出。pandas 就是做这件事的标准工具。

本文从 Java 开发者的视角出发，用熟悉的类比解释 pandas 的核心概念，附一个完整的 LLM 日志分析实战案例。

1.1 AI 项目中数据处理的典型场景

从数据读取到聚合输出的 pandas 典型处理流程

在进入代码之前，先明确 pandas 在 AI 工程中的具体用途：

场景一：训练数据清洗
从业务系统导出的用户对话记录通常杂乱无章，包含 HTML 标签、重复数据、空值、格式不统一的时间戳。喂给模型之前必须清洗干净，否则"垃圾进、垃圾出"。

场景二：RAG 文档预处理
RAG（Retrieval-Augmented Generation，检索增强生成，让 AI 在回答前先从知识库中检索相关内容）的知识库文档在向量化之前，需要提取正文、去除页眉页脚、按固定长度切块、过滤低质量段落。这些操作本质上是对一张"文档表"的批量处理。

场景三：评估结果分析
模型评估完成后，会产出一个包含问题、标准答案、模型输出、各项评分的 CSV 文件。通过 pandas 可以快速统计通过率、按类别分组分析、找出最差的 case 集中人工 review。

1.2 DataFrame 基础：把它想象成一张 Excel 表格

DataFrame（数据帧）是 pandas 的核心数据结构。对 Java 开发者来说，最直观的类比是：

DataFrame ≈ Excel 表格，每一列有名字和统一的数据类型，每一行有行号（索引）。

更严格的类比是：List<Map<String, Object>>，但 pandas 在内存布局和运算效率上远超这种结构。

python

import pandas as pd

# 从字典创建 DataFrame，类比 new ArrayList<>() 然后逐行 add
data = {
    "question": ["什么是RAG?", "如何部署模型?", "什么是Token?"],
    "answer": ["RAG是检索增强生成...", "可以用Docker...", "Token是..."],
    "score": [0.9, 0.7, 0.85],
    "model": ["gpt-4o", "gpt-4o", "gpt-3.5"]
}
df = pd.DataFrame(data)
print(df)
#          question           answer  score   model
# 0      什么是RAG?  RAG是检索增强生成...   0.90  gpt-4o
# 1   如何部署模型?     可以用Docker...   0.70  gpt-4o
# 2    什么是Token?         Token是...   0.85  gpt-3.5

1.2.1 读写常见格式

以下为代码示例，非程序员可跳过代码，重点看文字说明。

格式	读取	写入	适用场景
CSV	`pd.read_csv("file.csv")`	`df.to_csv("out.csv", index=False)`	通用，人类可读
JSON	`pd.read_json("file.json")`	`df.to_json("out.json")`	API 日志，嵌套数据
Excel	`pd.read_excel("file.xlsx")`	`df.to_excel("out.xlsx")`	与业务方交换数据
Parquet	`pd.read_parquet("file.parquet")`	`df.to_parquet("out.parquet")`	大数据集，列式存储

Parquet 是大型 AI 数据集的首选格式（列式存储格式，按列而非按行存储数据）：相比 CSV，文件体积通常缩小 5-10 倍，读取速度提升 3-5 倍，因为它按列存储，只读取需要的列。

1.2.2 基本操作：索引、切片、过滤

python

# 以下为代码示例，非程序员可跳过代码，重点看文字说明

# 选取单列（类比 list.stream().map(row -> row.get("score"))）
scores = df["score"]

# 选取多列
subset = df[["question", "score"]]

# 按条件过滤（类比 stream().filter(row -> row.getScore() > 0.8)）
high_score = df[df["score"] > 0.8]

# 选取前5行（类比 stream().limit(5)）
top5 = df.head(5)

# 按列排序（类比 stream().sorted(Comparator.comparing(Row::getScore).reversed())）
sorted_df = df.sort_values("score", ascending=False)

# 基础统计（一行代码，Java要手写很多）
print(df["score"].describe())
# count    3.000000
# mean     0.816667
# min      0.700000
# max      0.900000

1.3 数据处理流水线

在开始讲清洗方法之前，先看整体流程：

1.4 数据清洗五板斧

1.4.1 第一斧：去重

重复数据在 AI 项目中危害极大。重复的训练样本会让模型过拟合（模型记住了训练数据的细节，反而在新数据上表现变差），重复的 RAG 文档会让检索结果单调。

python

# 以下为代码示例，非程序员可跳过代码，重点看文字说明

# 查看重复数量
print(f"重复行数: {df.duplicated().sum()}")

# 基于所有列去重
df_clean = df.drop_duplicates()

# 基于特定列去重（question相同就算重复，保留第一条）
df_clean = df.drop_duplicates(subset=["question"], keep="first")

print(f"去重前: {len(df)} 行，去重后: {len(df_clean)} 行")

1.4.2 第二斧：缺失值处理

缺失值（NaN，Not a Number 的缩写，pandas 用它表示"此处无值/数据缺失"）是数据集中的空洞，必须处理，否则许多运算会直接报错或产生错误结果。

python

# 以下为代码示例，非程序员可跳过代码，重点看文字说明

# 查看每列的缺失值数量
print(df.isnull().sum())

# 方案一：删除含缺失值的行（数据量充足时）
df_dropped = df.dropna(subset=["answer"])  # 只要 answer 列为空就删

# 方案二：填充缺失值
df_filled = df.fillna({
    "score": 0.0,           # 分数缺失填0
    "model": "unknown",     # 模型名缺失填unknown
    "answer": ""            # 答案缺失填空字符串
})

# 方案三：用前一行的值填充（时序数据常用）
df_ffill = df.ffill()

场景	推荐策略	原因
评分列缺失，数据量多	`dropna` 删除该行	缺分数的记录无法用于分析
可选字段缺失	`fillna` 填默认值	保留记录，不影响主要分析
时序日志中间有空洞	`ffill` 前向填充	用上一个有效值近似
文本列缺失	`fillna("")` 填空串	防止字符串操作报错

1.4.3 第三斧：类型转换

pandas 读取 CSV 时，会尽力猜测每列的数据类型，但经常猜错。比如把数字列当成字符串，或把时间列当成普通字符串。

python

# 以下为代码示例，非程序员可跳过代码，重点看文字说明

# 查看当前类型（类比 Java 的 getClass()）
print(df.dtypes)
# question     object   <- object 就是字符串
# score        object   <- 本来应该是 float，被读成了字符串！
# created_at   object   <- 时间被读成了字符串

# 修正类型
df["score"] = df["score"].astype(float)
df["created_at"] = pd.to_datetime(df["created_at"])

# 转换后就可以做时间运算了
df["hour"] = df["created_at"].dt.hour  # 提取小时
df["date"] = df["created_at"].dt.date  # 提取日期

1.4.4 第四斧：字符串清洗（.str accessor）

文本数据是 AI 项目中最常见的数据类型，也是最脏的数据类型。.str 是 pandas 专门用于字符串列的操作接口（accessor，访问器），可以用类似 Python 字符串方法的语法批量处理整列数据。

python

# 以下为代码示例，非程序员可跳过代码，重点看文字说明

import re

# 假设 answer 列包含从 HTML 爬取的内容，带有 HTML 标签
df["answer_clean"] = df["answer"].str.replace(
    r"<[^>]+>", "", regex=True   # 正则去除 HTML 标签
)

# 去除首尾空白字符（类比 Java 的 String.trim()）
df["answer_clean"] = df["answer_clean"].str.strip()

# 截断过长文本（RAG 文档分块时常用）
MAX_LEN = 512
df["answer_trunc"] = df["answer_clean"].str[:MAX_LEN]

# 统一大小写（处理模型名称时常用）
df["model"] = df["model"].str.lower()   # "GPT-4O" -> "gpt-4o"

# 检查是否包含关键词（过滤敏感内容）
df["has_error"] = df["answer"].str.contains("error|exception|failed",
                                             case=False, regex=True)

# 提取模型版本号（用正则）
df["model_version"] = df["model"].str.extract(r"(\d+\.?\d*)")

.str 的核心价值：一行代码处理整列，比 Java 写一个 for 循环逐行处理简洁得多，且性能更好。

1.4.5 第五斧：异常值处理

异常值（Outlier，统计学术语，指远偏离正常范围的数据点）是远偏离正常范围的数据点，比如响应时间通常是 2-5 秒，但某条记录是 3600 秒，很可能是超时后的错误记录。

python

# 以下为代码示例，非程序员可跳过代码，重点看文字说明

# 方法一：基于业务规则过滤
# score 合法范围是 0-1，过滤掉范围外的值
df = df[(df["score"] >= 0) & (df["score"] <= 1)]

# 方法二：IQR 方法（统计学方法，不需要理解原理，会用即可）
# IQR（Interquartile Range，四分位距）：将数据排序后，中间50%的数据范围，超出1.5倍IQR视为异常值
Q1 = df["latency_ms"].quantile(0.25)  # 下四分位数
Q3 = df["latency_ms"].quantile(0.75)  # 上四分位数
IQR = Q3 - Q1
lower = Q1 - 1.5 * IQR
upper = Q3 + 1.5 * IQR
df = df[(df["latency_ms"] >= lower) & (df["latency_ms"] <= upper)]

1.5 大文件分块读取

当 LLM 调用日志文件达到 GB 级别时，直接 pd.read_csv() 会耗尽内存。chunksize 参数（分块大小）让 pandas 分批读取文件，每次只加载一部分到内存。

类比：就像 Java 中用 BufferedReader 逐行读文件，而不是一次性 Files.readAllBytes()。

python

# 以下为代码示例，非程序员可跳过代码，重点看文字说明

import pandas as pd

# 不用 chunksize（内存爆炸风险）
# df = pd.read_csv("huge_log.csv")  # 1GB 文件直接载入内存

# 使用 chunksize 分块处理
CHUNK_SIZE = 10_000  # 每批处理 1 万行
results = []

for chunk in pd.read_csv("huge_log.csv", chunksize=CHUNK_SIZE):
    # chunk 是一个普通的 DataFrame，包含当前批次的数据
    # 对每个 chunk 做清洗和聚合
    chunk_clean = chunk.dropna(subset=["answer"])
    chunk_clean = chunk_clean[chunk_clean["score"] > 0]

    # 聚合统计（按模型分组统计成功率）
    chunk_stat = chunk_clean.groupby("model")["success"].mean()
    results.append(chunk_stat)

# 合并所有批次的统计结果
final_result = pd.concat(results).groupby(level=0).mean()
print(final_result)

内存使用对比：

方式	1GB CSV 内存占用	风险
直接读取	约 3-5 GB（pandas 内部膨胀）	内存溢出崩溃
chunksize=10000	约 50-100 MB（稳定）	无

1.6 pandas + pydantic：批量数据校验

在第6章中已经介绍过 pydantic 用于单条数据的校验。在 AI 项目中，经常需要对 DataFrame 中的每一行进行结构化校验——比如验证 LLM 输出是否符合预期的 JSON 格式。

方案一：逐行 validate（简单，慢）

python

# 以下为代码示例，非程序员可跳过代码，重点看文字说明

from pydantic import BaseModel, ValidationError
from typing import Optional
import pandas as pd

class LLMOutput(BaseModel):
    question: str
    answer: str
    score: float
    model: str
    tool_calls: Optional[int] = 0

errors = []
valid_rows = []

for idx, row in df.iterrows():  # iterrows() 逐行遍历，类比 Java for-each
    try:
        validated = LLMOutput(**row.to_dict())
        valid_rows.append(validated.model_dump())
    except ValidationError as e:
        errors.append({"row": idx, "error": str(e)})

df_valid = pd.DataFrame(valid_rows)
print(f"校验通过: {len(df_valid)} 行，失败: {len(errors)} 行")

方案二：批量向量化校验（快，推荐）

python

# 以下为代码示例，非程序员可跳过代码，重点看文字说明

# 用 pandas 自身的条件判断做批量校验，比逐行快 10-100 倍
def validate_batch(df: pd.DataFrame) -> pd.DataFrame:
    mask = (
        df["question"].notna() &            # question 不为空
        df["answer"].str.len() > 10 &       # answer 长度大于 10
        df["score"].between(0, 1) &         # score 在 [0,1] 范围内
        df["model"].str.startswith("gpt")   # model 名称以 gpt 开头
    )
    invalid = df[~mask]
    if len(invalid) > 0:
        print(f"发现 {len(invalid)} 行不合法数据")
    return df[mask]

df_valid = validate_batch(df)

1.7 实战：分析 LLM 输出日志，统计工具调用成功率

场景：有一个 LLM Agent 的调用日志 CSV，需要分析每个工具的调用成功率，找出问题最多的工具。

日志格式示例：

code

request_id,timestamp,model,tool_name,tool_success,latency_ms,error_msg
req_001,2024-01-15 10:23:01,gpt-4o,search_web,1,1234,
req_002,2024-01-15 10:23:05,gpt-4o,calc_math,0,89,ZeroDivisionError
req_003,2024-01-15 10:23:08,gpt-3.5,search_web,1,987,

python

# 以下为代码示例，非程序员可跳过代码，重点看文字说明

import pandas as pd

# ---- 第一步：读取数据 ----
df = pd.read_csv("llm_tool_log.csv")
print(f"原始数据: {len(df)} 行")

# ---- 第二步：数据清洗 ----
# 类型修正
df["timestamp"] = pd.to_datetime(df["timestamp"])
df["tool_success"] = df["tool_success"].astype(int)
df["latency_ms"] = df["latency_ms"].astype(float)

# 去重（同一 request_id 不应有重复记录）
df = df.drop_duplicates(subset=["request_id"])

# 缺失值处理（error_msg 为空是正常情况，填空字符串）
df["error_msg"] = df["error_msg"].fillna("")

# ---- 第三步：字符串清洗 ----
df["tool_name"] = df["tool_name"].str.strip().str.lower()
df["model"] = df["model"].str.strip().str.lower()

# ---- 第四步：过滤异常值 ----
# 延迟超过 60 秒的记录视为超时异常
df = df[df["latency_ms"] <= 60_000]

print(f"清洗后: {len(df)} 行")

# ---- 第五步：统计分析 ----
# 按工具名分组，统计成功率和平均延迟
tool_stats = df.groupby("tool_name").agg(
    total_calls=("tool_success", "count"),
    success_rate=("tool_success", "mean"),       # mean() 对 0/1 列等于成功率
    avg_latency_ms=("latency_ms", "mean"),
    p99_latency_ms=("latency_ms", lambda x: x.quantile(0.99))
).round(3).sort_values("success_rate")

print("\n=== 工具调用成功率排行（从低到高）===")
print(tool_stats.to_string())

# ---- 第六步：找出常见错误 ----
failed = df[df["tool_success"] == 0]
error_counts = failed["error_msg"].str.extract(
    r"(\w+Error|\w+Exception)"  # 提取错误类型
)[0].value_counts()

print("\n=== 错误类型 TOP 10 ===")
print(error_counts.head(10))

# ---- 第七步：按时间分析趋势 ----
df["hour"] = df["timestamp"].dt.floor("h")  # 按小时取整
hourly = df.groupby("hour")["tool_success"].mean()
print("\n=== 按小时的工具成功率 ===")
print(hourly.tail(24))  # 最近 24 小时

# ---- 第八步：导出结果 ----
tool_stats.to_csv("tool_stats_report.csv")
failed.to_csv("failed_calls.csv", index=False)
print("\n报告已保存")

运行后，tool_stats_report.csv 会直接告诉你哪个工具最不稳定，哪个工具延迟最高，重点排查方向一目了然。

1.8 与 Java 的对比：Stream API vs pandas

操作	Java Stream API	pandas
过滤	`stream().filter(r -> r.getScore() > 0.8)`	`df[df["score"] > 0.8]`
映射	`stream().map(r -> r.getName().toLowerCase())`	`df["name"].str.lower()`
分组聚合	`stream().collect(groupingBy(..., counting()))`	`df.groupby("model").size()`
排序	`stream().sorted(Comparator.comparing(...))`	`df.sort_values("score")`
统计	需手写 Collectors	`df["score"].describe()`
内存模型	对象列表，有装箱开销	连续内存数组，无装箱
大文件处理	需手写分批逻辑	`chunksize` 参数原生支持
类型安全	编译期检查	运行时，dtypes 需手动核查

pandas 在数据处理的简洁性上远超 Java Stream，但代价是运行时类型安全需要程序员自己保证。引入 pydantic 校验是弥补这一缺陷的标准做法。

1.9 RAG 系统中的文档预处理实战

RAG（Retrieval-Augmented Generation，检索增强生成）系统的核心是知识库的质量。知识库文档在向量化之前，需要经过严格的预处理流水线。pandas 是这个流水线的标准工具：

python

import pandas as pd
import re
from pathlib import Path

def preprocess_rag_documents(
    raw_docs_path: str,
    output_path: str,
    chunk_size: int = 512,
    chunk_overlap: int = 50,
) -> None:
    """
    RAG 文档预处理完整流水线。
    输入：原始文档（PDF解析后的文本、爬取的网页等）
    输出：可直接向量化的文档块
    """

    # 第一步：读取原始文档
    df = pd.read_csv(raw_docs_path)
    print(f"原始文档: {len(df)} 篇")

    # 第二步：基础清洗
    # 去除空白文档
    df = df.dropna(subset=["content"])
    df = df[df["content"].str.len() > 50]

    # 去除重复文档（完全相同的内容）
    df = df.drop_duplicates(subset=["content"])

    # 清洗文本（去除PDF页眉页脚的常见模式）
    df["content"] = (
        df["content"]
        .str.replace(r"第\d+页", "", regex=True)       # 去除页码
        .str.replace(r"版权所有.*?\n", "", regex=True) # 去除版权信息
        .str.replace(r"\s{3,}", " ", regex=True)       # 压缩多余空白
        .str.strip()
    )

    # 第三步：质量过滤
    # 过滤掉质量太低的文档（如只有数字/符号的段落）
    def is_quality_content(text: str) -> bool:
        if len(text) < 50:
            return False
        # 中文字符比例应该超过 30%
        chinese_chars = len(re.findall(r'[\u4e00-\u9fff]', text))
        total_chars = len(text.replace(" ", ""))
        return total_chars > 0 and chinese_chars / total_chars > 0.3

    quality_mask = df["content"].apply(is_quality_content)
    df = df[quality_mask]
    print(f"质量过滤后: {len(df)} 篇")

    # 第四步：文档分块（Chunking）
    # 将长文档切割成固定大小的块，适合向量化
    chunks = []
    for _, row in df.iterrows():
        text = row["content"]
        doc_id = row.get("id", "unknown")
        source = row.get("source", "unknown")

        # 简单的固定长度分块（实际项目可以用 langchain 的 TextSplitter）
        for i in range(0, len(text), chunk_size - chunk_overlap):
            chunk_text = text[i:i + chunk_size]
            if len(chunk_text) < 50:  # 丢弃太短的块
                continue
            chunks.append({
                "chunk_id": f"{doc_id}_{i//chunk_size}",
                "doc_id": doc_id,
                "source": source,
                "content": chunk_text,
                "chunk_index": i // chunk_size,
            })

    # 第五步：转成 DataFrame 并保存
    chunks_df = pd.DataFrame(chunks)
    print(f"分块后: {len(chunks_df)} 个文档块")

    # 保存为 Parquet（高效列式存储，比 CSV 快 5-10 倍读取）
    chunks_df.to_parquet(output_path, index=False)
    print(f"已保存到: {output_path}")

    # 打印质量统计
    print("\n=== 分块质量统计 ===")
    print(f"平均块长度: {chunks_df['content'].str.len().mean():.0f} 字符")
    print(f"最短块: {chunks_df['content'].str.len().min()} 字符")
    print(f"最长块: {chunks_df['content'].str.len().max()} 字符")
    print(f"来源分布:\n{chunks_df['source'].value_counts()}")


# 使用示例
# preprocess_rag_documents(
#     raw_docs_path="raw_docs.csv",
#     output_path="processed_chunks.parquet",
#     chunk_size=512,
# )

与 LangChain TextSplitter 的关系：

LangChain 提供了更智能的 RecursiveCharacterTextSplitter，能按语义边界（段落、句子）切割，而不是粗暴的固定长度切割。实际 RAG 项目中，建议用 LangChain 的分割器，用 pandas 做清洗和统计。

python

from langchain.text_splitter import RecursiveCharacterTextSplitter
import pandas as pd

# 用 LangChain 分割器 + pandas 做统计
splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "！", "？", " "]
)

df = pd.read_csv("raw_docs.csv")
all_chunks = []

for _, row in df.iterrows():
    chunks = splitter.split_text(row["content"])
    for i, chunk in enumerate(chunks):
        all_chunks.append({
            "doc_id": row["id"],
            "chunk_index": i,
            "content": chunk,
        })

chunks_df = pd.DataFrame(all_chunks)
print(f"总共 {len(chunks_df)} 个文档块")
print(f"平均长度: {chunks_df['content'].str.len().mean():.0f} 字符")

1.10 小结

使用场景	pandas 的作用	核心方法
RAG 文档预处理	清洗、过滤、统计	`.str` + `dropna` + `apply`
LLM 日志分析	工具成功率、延迟分析	`groupby().agg()`
训练数据清洗	去重、类型修正、异常值	`drop_duplicates` + `astype`
大规模数据处理	分块避免内存溢出	`chunksize`
数据质量校验	批量验证	向量化条件 + pydantic