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

基于大模型的个人消费分析和理财助手开发日志 8AI 账单分类上LangChain 模块设计与三层 JSON 解析背景与问题用户从微信、支付宝导出的账单 CSV/xlsx 文件中交易类型字段“交易分类”的精细度和标准化程度不一致。例如微信账单的交易类型可能是商户消费、转账等有限类别支付宝账单的交易分类是支付宝自己的分类体系不同用户导入的账单混在一起无法进行统一的统计分析目标是用大模型对每条账单进行统一的交易类型分类将所有交易映射到标准类别体系餐饮、交通、购物等 10 个类别使月度报告的分类聚合统计有意义。LangChain 模块设计Pydantic 输出模型classClassifiedBill(BaseModel):单条账单分类结果index:intField(description在批次中的索引位置)transaction_type:strField(description分类结果从预定义列表中选择)index字段是核心设计——由于 LLM 不保证输出顺序需要依赖输入时的序号来对齐结果。前端传入的账单列表顺序通过 index 重建确保分类结果与原始数据一一对应。Prompt 工程CATEGORIES 可选的交易类型类别 - 餐饮餐饮、外卖、食品、饮品等 - 交通出行打车、公交、地铁、加油、停车等 - 购物电商、超市、日用品、服装、电子产品等 - ... SYSTEM_PROMPTf你是一个个人记账账单的分类助手。 你的任务是根据账单信息将交易类型分类到标准类别中。{CATEGORIES}注意 1. transaction_direction 为收入的账单优先考虑工资收入或转账类别 2. 只能从上述列表中选择... 每个类别都附带了详细的子类解释如餐饮包含餐饮、外卖、食品、饮品等减少 LLM 的歧义判断。收入账单特殊处理优先匹配工资收入或转账避免将收入误判为支出类别。分批处理策略asyncdefclassify_bill_batch(bills:list[Bill])-list[ClassifiedBill]:对一批账单进行 AI 分类最多推荐 50 条...foriinrange(0,total,50):batchparsed_data[i:i50]resultsawaitclassify_bill_batch(batch)forj,rinenumerate(results):batch[j].transaction_typer.transaction_type每批 50 条是一个经验值——太少则 LLM API 调用次数过多浪费 token 在重复的 system prompt 上太多则一次 context 内需要处理的账单量过大影响分类准确性。重试策略try:resultawaitchain.ainvoke(...)returnresult.resultsexceptException:# 失败时重试 1 次try:resultawaitchain.ainvoke(...)returnresult.resultsexceptException:raise对 LLM API 的网络抖动做一次重试后续优化为 3 次 错误回传。三层 JSON 解析策略在集成阶段遇到了一个棘手的问题某些 API 代理如 Cloudflare AI Gateway不支持response_format参数导致with_structured_output()调用失败。解决方案是完全放弃with_structured_output()改用手动 JSON 提示 三层解析# 第 1 层直接解析try:returnjson.loads(text)exceptjson.JSONDecodeError:pass# 第 2 层从 Markdown 代码块中提取matchre.search(r(?:json)?\s*\n?([\s\S]*?),text)ifmatch:try:returnjson.loads(match.group(1))exceptjson.JSONDecodeError:pass# 第 3 层暴力查找最外层大括号brace_starttext.find({)ifbrace_start0:brace_endtext.rfind(})ifbrace_endbrace_start:try:returnjson.loads(text[brace_start:brace_end1])exceptjson.JSONDecodeError:passraiseValueError(f无法从响应中解析 JSON:{text[:200]})三层解析策略的设计意图直接解析——LLM 完美遵循指令返回纯 JSON 时零开销Markdown 代码块提取——很多 LLM 会在 JSON 外包一层json这是最常见的不规范行为暴力查找大括号——当 LLM 在 JSON 前后加了额外文字说明时仍然能提取到数据3 次重试 错误回传forattemptinrange(3):try:messagesprompt.format_messages(totallen(bills),billsbills_text)ifattempt0:messages.append((human,f之前的解析出错{last_error}\n\n请只返回符合格式要求的 JSON 数据。))responseawait_llm.ainvoke(messages)parsed_parse_json_response(response.content)results_validate_results(parsed,len(bills))returnsorted(results,keylambdar:r.index)exceptExceptionase:last_errorstr(e)第 2、3 次重试时将上一轮的错误信息回传给 LLM“解析出错Expected object…请只返回 JSON”让 LLM 自行修正格式。这种方法利用 LLM 的纠错能力远比硬编码更多解析规则要灵活。测试 Mock 的技巧# ChatOpenAI 是 Pydantic v1 模型无法直接 mock 实例属性# 所以改为 mock 整个 _llm 模块级变量mock_llmmocker.AsyncMock()mock_llm.ainvokeAsyncMock(return_valuemock_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 调用成本这个模块的设计哲学是不假设 LLM 的行为是完美的。LLM 可能输出格式不对、可能漏掉条目、可能在压力下返回错误——但系统通过多层容错机制解析容错 重试纠错 批次隔离确保了在实际生产环境中的可靠性。