LangChain：把“3月5号打车去机场，滴滴出行花了86块”这样的口语

把“3月5号打车去机场，滴滴出行花了86块”这样的口语，抽取成 date / amount / currency / category / merchant 这样的结构化报销记录。 实现方式：LangChain 的 with_structured_output + Pydantic + 本地 Ollama。 这篇的重点不是“能抽出来”，而是怎么知道抽得对：

• 用标注好的评测集算字段准确率；
• 校验失败时记录下来，而不是让程序崩溃；
• 加一道“依据检查”：金额和日期必须能在原文里找到。

先看清成品（box 上真实运行，qwen2.5 1.5B，只用 CPU，6 条约 11 秒）

• 字段准确率：date、amount、currency、category 都是 100%，merchant 83%（6 条里错 1 条）。
• “忘了花多少钱”：模型给出 amount: 0，被 Pydantic 的 gt=0 拦下，记为解析失败。
• “报销一下上周的出租车费”：模型编造了 100 元和 2026-01-01，依据检查报出“金额 100.0 不在原文中、日期 2026-01-01 不在原文中”。
• pytest 6 项通过，测试不需要 Ollama。

合规提醒：本系列只用本地 Ollama 模型，数据不出本机。如果换成云端模型，请先确认数据合规和服务商条款，API Key 放在环境变量里，不要共享账号。模型的许可证以它在 Ollama 或 Hugging Face 页面上写的为准。示例数据都是我自己编的。本文不列价格。

做完之后你会得到什么？

需要准备哪些环境？

安装：pip install -r requirements.txt

文件结构

60-langchain-A-src/
├── extract.py
├── cases.json
├── requirements.txt
└── test_extract.py

第 1 步：抽取链

"""把口语化的报销记录抽取成结构化数据：LangChain + 本地 Ollama。
链：ChatPromptTemplate | ChatOllama.with_structured_output(Expense, include_raw=True)
- Ollama 的 json_schema 模式会把 schema 交给模型做约束解码，但**字段值对不对**仍要靠 Pydantic 校验和评测集
- 解析失败时不抛错，记录 parsing_error，交给人工处理"""
import datetime as dt, json, os, sys
from typing import Literal
from pydantic import BaseModel, Field, field_validator
from langchain_core.prompts import ChatPromptTemplate

MODEL = os.environ.get("OLLAMA_MODEL", "qwen2.5-local:1.5b")

class Expense(BaseModel):
    date: dt.date = Field(description="消费日期，YYYY-MM-DD；文本只写了月日时用参考年份")
    amount: float = Field(gt=0, description="金额数字，不带货币符号")
    currency: Literal["CNY", "USD", "JPY", "EUR"] = Field(description="币种代码；“元/块/人民币”是 CNY")
    category: Literal["交通", "餐饮", "住宿", "办公", "其他"]
    merchant: str = Field(description="商家或平台名称；没提到就写 未知")

    @field_validator("merchant")
    @classmethod
    def strip(cls, v):
        return v.strip() or "未知"

PROMPT = ChatPromptTemplate.from_messages([
    ("system", "你是报销录入助手。从用户的一句话里抽取一条报销记录。参考年份：{year}。只依据原文，不要编造。"),
    ("human", "{text}"),
])

def build_chain(llm=None):
    if llm is None:
        from langchain_ollama import ChatOllama
        llm = ChatOllama(model=MODEL, temperature=0)
    return PROMPT | llm.with_structured_output(Expense, include_raw=True)

def extract(chain, text, year=2026):
    r = chain.invoke({"text": text, "year": year})
    if r.get("parsing_error") or r.get("parsed") is None:
        return None, str(r.get("parsing_error") or "模型没有给出可解析的结果")
    return r["parsed"], None

import re
MONTHS = ["january","february","march","april","may","june","july","august","september","october","november","december"]

def ungrounded(text, e):
    """模型可能编造原文没有的数字：金额和日期必须能在原文里找到依据，否则交给人工复核。"""
    issues, low = [], text.lower()
    nums = {float(x) for x in re.findall(r"\d+(?:\.\d+)?", text)}
    if e.amount not in nums:
        issues.append(f"金额 {e.amount} 不在原文中")
    m, d = e.date.month, e.date.day
    pats = [rf"(?<!\d)0?{m}\s*[月/.-]\s*0?{d}(?!\d)", rf"{MONTHS[m-1]}\s+0?{d}(?!\d)"]
    if not any(re.search(p, low) for p in pats):
        issues.append(f"日期 {e.date} 不在原文中")
    return issues

def evaluate(chain, cases):
    fields = ["date", "amount", "currency", "category", "merchant"]
    hit = {f: 0 for f in fields}; rows = []
    for c in cases:
        got, err = extract(chain, c["text"])
        g = json.loads(got.model_dump_json()) if got else {}
        ok = {f: g.get(f) == c["expect"][f] for f in fields}
        for f in fields:
            hit[f] += ok[f]
        rows.append((c["text"], g or err, [f for f in fields if not ok[f]]))
    return {f: hit[f] / len(cases) for f in fields}, rows

if __name__ == "__main__":
    cases = json.load(open(sys.argv[1] if len(sys.argv) > 1 else "cases.json", encoding="utf-8"))
    acc, rows = evaluate(build_chain(), cases)

    chain = build_chain()
    for t in ["昨天和同事聚餐，忘了花多少钱", "报销一下上周的出租车费"]:
        got, err = extract(chain, t)
        print("复核", t, "→", (err.splitlines()[0] if err else None) or ungrounded(t, got) or got)
    for text, got, wrong in rows:
        print(("✓" if not wrong else "✗ " + ",".join(wrong)), text, "→", got)
    print("字段准确率：", {k: f"{v:.0%}" for k, v in acc.items()})

第 2 步：评测集

[
 {"text": "3月5号打车去机场，滴滴出行花了86块", "expect": {"date": "2026-03-05", "amount": 86.0, "currency": "CNY", "category": "交通", "merchant": "滴滴出行"}},
 {"text": "2026-04-12 在全季酒店住了一晚 429 元", "expect": {"date": "2026-04-12", "amount": 429.0, "currency": "CNY", "category": "住宿", "merchant": "全季酒店"}},
 {"text": "4月20日请客户在海底捞吃饭，人民币 612.5", "expect": {"date": "2026-04-20", "amount": 612.5, "currency": "CNY", "category": "餐饮", "merchant": "海底捞"}},
 {"text": "5/8 京东买了两包A4纸和墨盒，一共 159 元", "expect": {"date": "2026-05-08", "amount": 159.0, "currency": "CNY", "category": "办公", "merchant": "京东"}},
 {"text": "June 3rd, Uber ride in San Francisco, $23.40", "expect": {"date": "2026-06-03", "amount": 23.4, "currency": "USD", "category": "交通", "merchant": "Uber"}},
 {"text": "7月1日东京出差，便利店买饮料 380 日元", "expect": {"date": "2026-07-01", "amount": 380.0, "currency": "JPY", "category": "餐饮", "merchant": "未知"}}
]

第 3 步：测试

import json
from langchain_core.runnables import RunnableLambda
from langchain_core.messages import AIMessage
from pydantic import ValidationError
import extract as E

class FakeLLM:
    """测试替身：with_structured_output 返回和 LangChain include_raw=True 相同形状的 dict。"""
    def __init__(self, payload):
        self.payload = payload
    def with_structured_output(self, schema, include_raw=False):
        def run(_):
            raw = AIMessage(content=json.dumps(self.payload, ensure_ascii=False))
            try:
                return {"raw": raw, "parsed": schema.model_validate(self.payload), "parsing_error": None}
            except ValidationError as e:
                return {"raw": raw, "parsed": None, "parsing_error": e}
        return RunnableLambda(run)

CASES = json.load(open("cases.json", encoding="utf-8"))

def test_extract_ok():
    got, err = E.extract(E.build_chain(FakeLLM(CASES[0]["expect"])), CASES[0]["text"])
    assert err is None and got.amount == 86.0 and str(got.date) == "2026-03-05"

def test_validation_error_is_reported_not_raised():
    bad = {**CASES[0]["expect"], "amount": 0}
    got, err = E.extract(E.build_chain(FakeLLM(bad)), "x")
    assert got is None and "greater than 0" in err

def test_bad_category_rejected():
    got, err = E.extract(E.build_chain(FakeLLM({**CASES[0]["expect"], "category": "娱乐"})), "x")
    assert got is None

def test_grounding_accepts_all_labels():
    for c in CASES:
        assert E.ungrounded(c["text"], E.Expense(**c["expect"])) == [], c["text"]

def test_grounding_flags_invented_values():   # 真实 qwen2.5 1.5B 对这句话给出的结果
    e = E.Expense(date="2026-01-01", amount=100, currency="CNY", category="交通", merchant="出租车")
    assert E.ungrounded("报销一下上周的出租车费", e) == ["金额 100.0 不在原文中", "日期 2026-01-01 不在原文中"]

def test_evaluate_counts():
    acc, rows = E.evaluate(E.build_chain(FakeLLM(CASES[1]["expect"])), CASES[:2])
    assert acc["amount"] == 0.5 and rows[1][2] == []

第 4 步：运行

$ ollama pull qwen2.5:1.5b && export OLLAMA_MODEL=qwen2.5:1.5b    # box 上是 qwen2.5-local:1.5b
$ python extract.py
复核 昨天和同事聚餐，忘了花多少钱 → Failed to parse Expense from completion {"amount": 0, "category": "餐饮", "currency": "CNY", "date": "2026-01-01", "merchant": "聚餐"}. Got: 1 validation error for Expense
复核 报销一下上周的出租车费 → ['金额 100.0 不在原文中', '日期 2026-01-01 不在原文中']
✓ 3月5号打车去机场，滴滴出行花了86块 → {'date': '2026-03-05', 'amount': 86.0, 'currency': 'CNY', 'category': '交通', 'merchant': '滴滴出行'}
✓ 2026-04-12 在全季酒店住了一晚 429 元 → {...}
✓ 4月20日请客户在海底捞吃饭，人民币 612.5 → {...}
✓ 5/8 京东买了两包A4纸和墨盒，一共 159 元 → {...}
✓ June 3rd, Uber ride in San Francisco, $23.40 → {'date': '2026-06-03', 'amount': 23.4, 'currency': 'USD', ...}
✗ merchant 7月1日东京出差，便利店买饮料 380 日元 → {..., 'merchant': '便利店'}
字段准确率： {'date': '100%', 'amount': '100%', 'currency': '100%', 'category': '100%', 'merchant': '83%'}
$ python -m pytest -q
6 passed

调试示例

1. 模型编造了原文没有的数字。 对“报销一下上周的出租车费”，模型返回了一个完全合法的 Expense(date=2026-01-01, amount=100.0, ...)，Pydantic 校验全部通过，但原文里根本没有金额和日期。单靠 schema 约束拦不住这种情况，所以我加了 ungrounded()：金额必须作为数字出现在原文中，日期的月和日也必须能在原文中找到（支持“3月5号”、“5/8”、“04-12”、“June 3rd”这几种写法）。测试 test_grounding_accepts_all_labels 确认 6 条正确样例不会被误报。

2. 标注口径不清。 唯一错误的那条是“便利店买饮料”：我标注的商家是“未知”，因为没有具体店名；模型给出的是“便利店”。这其实是标注规则没有说清楚，不完全是模型的问题。我把规则写进了 Field 的描述（“没提到就写 未知”），但 1.5B 模型仍然给出“便利店”。所以评测结果如实保留了这个 83%，没有为了好看去改标注。

常见问题

模板规则

• 先有评测集，再调提示词。
• 解析失败和依据不足的结果都进入人工复核队列。
• Literal 枚举限制取值范围。
• 报告里写明模型、版本和准确率。

项目资源

结构化输出解决的是格式问题，对不对还得靠评测集和依据检查来保证。