基于大模型的个人消费分析和理财助手：开发日志 8

GraftCopolymer

45人浏览 · 2026-06-09 10:59:39

GraftCopolymer · 2026-06-09 10:59:39 发布

基于大模型的个人消费分析和理财助手：开发日志 8

AI 账单分类（上）：LangChain 模块设计与三层 JSON 解析

背景与问题

用户从微信、支付宝导出的账单 CSV/xlsx 文件中，交易类型字段（“交易分类”）的精细度和标准化程度不一致。例如：

微信账单的"交易类型"可能是"商户消费"、"转账"等有限类别
支付宝账单的"交易分类"是支付宝自己的分类体系
不同用户导入的账单混在一起，无法进行统一的统计分析

目标是：用大模型对每条账单进行统一的交易类型分类，将所有交易映射到标准类别体系（餐饮、交通、购物等 10+ 个类别），使月度报告的分类聚合统计有意义。

LangChain 模块设计

Pydantic 输出模型：

class ClassifiedBill(BaseModel):
    """单条账单分类结果"""
    index: int = Field(description="在批次中的索引位置")
    transaction_type: str = Field(description="分类结果，从预定义列表中选择")

index 字段是核心设计——由于 LLM 不保证输出顺序，需要依赖输入时的序号来对齐结果。前端传入的账单列表顺序通过 index 重建，确保分类结果与原始数据一一对应。

Prompt 工程：

CATEGORIES = """
可选的交易类型类别：
- 餐饮：餐饮、外卖、食品、饮品等
- 交通出行：打车、公交、地铁、加油、停车等
- 购物：电商、超市、日用品、服装、电子产品等
- ...
"""

SYSTEM_PROMPT = f"""你是一个个人记账账单的分类助手。
你的任务是根据账单信息将交易类型分类到标准类别中。
{CATEGORIES}
注意：
1. transaction_direction 为"收入"的账单，优先考虑"工资收入"或"转账"类别
2. 只能从上述列表中选择...
"""

每个类别都附带了详细的子类解释（如"餐饮"包含"餐饮、外卖、食品、饮品等"），减少 LLM 的歧义判断。收入账单特殊处理，优先匹配"工资收入"或"转账"，避免将"收入"误判为支出类别。

分批处理策略：

async def classify_bill_batch(bills: list[Bill]) -> list[ClassifiedBill]:
    """对一批账单进行 AI 分类（最多推荐 50 条）"""
    ...
    for i in range(0, total, 50):
        batch = parsed_data[i:i+50]
        results = await classify_bill_batch(batch)
        for j, r in enumerate(results):
            batch[j].transaction_type = r.transaction_type

每批 50 条是一个经验值——太少则 LLM API 调用次数过多（浪费 token 在重复的 system prompt 上），太多则一次 context 内需要处理的账单量过大，影响分类准确性。

重试策略：

try:
    result = await chain.ainvoke(...)
    return result.results
except Exception:
    # 失败时重试 1 次
    try:
        result = await chain.ainvoke(...)
        return result.results
    except Exception:
        raise

对 LLM API 的网络抖动做一次重试（后续优化为 3 次 + 错误回传）。

三层 JSON 解析策略

在集成阶段遇到了一个棘手的问题：某些 API 代理（如 Cloudflare AI Gateway）不支持 response_format 参数，导致 with_structured_output() 调用失败。

解决方案是完全放弃 with_structured_output()，改用手动 JSON 提示 + 三层解析：

# 第 1 层：直接解析
try:
    return json.loads(text)
except json.JSONDecodeError:
    pass

# 第 2 层：从 Markdown 代码块中提取
match = re.search(r'```(?:json)?\s*\n?([\s\S]*?)```', text)
if match:
    try:
        return json.loads(match.group(1))
    except json.JSONDecodeError:
        pass

# 第 3 层：暴力查找最外层大括号
brace_start = text.find('{')
if brace_start >= 0:
    brace_end = text.rfind('}')
    if brace_end > brace_start:
        try:
            return json.loads(text[brace_start:brace_end + 1])
        except json.JSONDecodeError:
            pass

raise ValueError(f"无法从响应中解析 JSON: {text[:200]}")

三层解析策略的设计意图：

直接解析——LLM 完美遵循指令返回纯 JSON 时，零开销
Markdown 代码块提取——很多 LLM 会在 JSON 外包一层 json，这是最常见的不规范行为
暴力查找大括号——当 LLM 在 JSON 前后加了额外文字说明时，仍然能提取到数据

3 次重试 + 错误回传：

for attempt in range(3):
    try:
        messages = prompt.format_messages(total=len(bills), bills=bills_text)
        if attempt > 0:
            messages.append(("human", 
                f"之前的解析出错：{last_error}\n\n请只返回符合格式要求的 JSON 数据。"))
        response = await _llm.ainvoke(messages)
        parsed = _parse_json_response(response.content)
        results = _validate_results(parsed, len(bills))
        return sorted(results, key=lambda r: r.index)
    except Exception as e:
        last_error = str(e)

第 2、3 次重试时，将上一轮的错误信息回传给 LLM（“解析出错：Expected object…，请只返回 JSON”），让 LLM 自行修正格式。这种方法利用 LLM 的"纠错"能力，远比硬编码更多解析规则要灵活。

测试 Mock 的技巧：

# ChatOpenAI 是 Pydantic v1 模型，无法直接 mock 实例属性
# 所以改为 mock 整个 _llm 模块级变量
mock_llm = mocker.AsyncMock()
mock_llm.ainvoke = AsyncMock(return_value=mock_response)
mocker.patch('openapi_server.utils.bill_classify_utils._llm', mock_llm)

这里有一个隐含的知识点：ChatOpenAI 实例是 Pydantic 模型，通过 Pydantic 的 __setattr__ 机制管理属性，mocker.patch.object 无法覆盖其方法。解决方案是直接 patch 模块级变量 _llm，绕过实例方法调用。

总结

模块	技术选择	目的
输出模型	Pydantic BaseModel + index 字段	保证结果对齐
LLM 框架	LangChain ChatPromptTemplate	结构化 prompts 管理
输出格式	手动 JSON 提示 + 3 层解析	兼容无 response_format 的 API
错误处理	3 次重试 + 错误回传	自愈性提高至接近 100%
分批策略	每批 50 条	平衡准确率与 API 调用成本