用 LLM 批量提取结构化数据：评论/合同/日志到干净 JSON

jwang426

255人浏览 · 2026-04-06 08:00:29

jwang426 · 2026-04-06 08:00:29 发布

背景

文本里藏着结构化数据，但手动提取很费劲——评论要分情感和投诉类型，合同要抠出金额和日期，日志要识别错误等级。写正则脆、训模型重，用 LLM 做这件事准确率够用，换个 prompt 就能调整提取逻辑。

这篇分享中模型调用走 ai.tikhub.io，用到不同模型对比，一个base_url较为方便。

基础方案：强制 JSON 输出

直接在 prompt 里要求输出 JSON 不够稳——模型有时会在前后加解释文字导致解析报错。用 response_format 参数更可靠：

# extract_basic.py
import os, json
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["TIKHUB_API_KEY"],
    base_url="https://ai.tikhub.io/v1"
)

def extract_from_review(text: str) -> dict:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        response_format={"type": "json_object"},
        messages=[
            {
                "role": "system",
                "content": """从用户评论提取以下字段，只返回 JSON：
{
  "sentiment": "positive | negative | neutral",
  "has_complaint": true | false,
  "complaint_type": "价格 | 质量 | 服务 | 物流 | null",
  "rating_implied": 1-5 的数字或 null
}"""
            },
            {"role": "user", "content": text}
        ]
    )
    return json.loads(resp.choices[0].message.content)

# 测试
samples = [
    "质量不错，就是快递太慢了，等了一周",
    "完全是垃圾！买了两个都坏了，客服不给退款",
    "性价比很高，用了三个月没出问题，推荐"
]
for s in samples:
    print(json.dumps(extract_from_review(s), ensure_ascii=False))

输出：

{"sentiment": "neutral", "has_complaint": true, "complaint_type": "物流", "rating_implied": 3}
{"sentiment": "negative", "has_complaint": true, "complaint_type": "服务", "rating_implied": 1}
{"sentiment": "positive", "has_complaint": false, "complaint_type": null, "rating_implied": 5}

复杂场景：长合同信息提取

合同这类长文本有两个额外问题：上下文够不够、嵌套字段怎么约束。

# extract_contract.py
import os, json
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["TIKHUB_API_KEY"],
    base_url="https://ai.tikhub.io/v1"
)

SCHEMA = """
{
  "parties": {"party_a": "甲方名称", "party_b": "乙方名称"},
  "contract_type": "服务合同 | 采购合同 | 劳动合同 | 其他",
  "amount": {"value": 数字或null, "currency": "CNY | USD", "payment_terms": "描述"},
  "dates": {
    "signed_date": "YYYY-MM-DD 或 null",
    "effective_date": "YYYY-MM-DD 或 null",
    "expiry_date": "YYYY-MM-DD 或 null"
  },
  "key_obligations": ["主要义务列表，每条50字以内"],
  "penalty_clauses": ["违约条款摘要，没有则为空数组"]
}
"""

def extract_contract(text: str) -> dict:
    resp = client.chat.completions.create(
        model="gemini-2.5-pro",  # 100万 token 上下文，长合同不截断
        response_format={"type": "json_object"},
        messages=[
            {
                "role": "system",
                "content": f"从合同中提取信息，按以下结构返回，找不到的字段填 null：\n{SCHEMA}"
            },
            {"role": "user", "content": text}
        ]
    )
    return json.loads(resp.choices[0].message.content)

with open("contract.txt", encoding="utf-8") as f:
    print(json.dumps(extract_contract(f.read()), ensure_ascii=False, indent=2))

这里换成 gemini-2.5-pro 是因为上下文窗口 100 万 token，几十页的合同不会截断。Base URL 不变，改一个参数的事。

批量处理：并发控制 + 错误重试

实际场景往往是几百上千条，需要解决限速和解析失败两个问题：

# batch_extract.py
import os, json, asyncio
from openai import AsyncOpenAI
from typing import Optional

client = AsyncOpenAI(
    api_key=os.environ["TIKHUB_API_KEY"],
    base_url="https://ai.tikhub.io/v1"
)

async def extract_one(
    text: str, prompt: str, model: str,
    sem: asyncio.Semaphore, retries: int = 3
) -> Optional[dict]:
    async with sem:
        for i in range(retries):
            try:
                resp = await client.chat.completions.create(
                    model=model,
                    response_format={"type": "json_object"},
                    messages=[
                        {"role": "system", "content": prompt},
                        {"role": "user", "content": text}
                    ]
                )
                return json.loads(resp.choices[0].message.content)
            except json.JSONDecodeError:
                if i == retries - 1:
                    return None
                await asyncio.sleep(1)
            except Exception:
                if i == retries - 1:
                    return None
                await asyncio.sleep(2 ** i)  # 指数退避


async def batch_extract(
    texts: list[str], prompt: str,
    model: str = "gpt-4o-mini", max_concurrent: int = 5
) -> list[Optional[dict]]:
    sem = asyncio.Semaphore(max_concurrent)
    return await asyncio.gather(*[
        extract_one(t, prompt, model, sem) for t in texts
    ])


async def main():
    PROMPT = """提取字段返回 JSON：
{"sentiment": "positive|negative|neutral", "has_complaint": true|false}"""

    texts = ["快递很快，东西好", "质量太差，要退款", "还行吧"]
    results = await batch_extract(texts, PROMPT)

    ok = [(t, r) for t, r in zip(texts, results) if r]
    print(f"成功 {len(ok)}/{len(texts)} 条")
    for t, r in ok:
        print(f"{t} → {r}")

asyncio.run(main())

失败的条目记录下来统一补跑，比每条死等更高效。max_concurrent=5 偏保守，可以根据实际限速情况往上调。

多模型对比

不同任务对模型的要求差别挺大，简单跑一组对比：

# compare.py
import os, json, asyncio, time
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key=os.environ["TIKHUB_API_KEY"],
    base_url="https://ai.tikhub.io/v1"
)

PROMPT = """提取字段返回 JSON：
{"sentiment": "positive|negative|neutral", "has_complaint": true|false, "complaint_type": "价格|质量|服务|物流|null"}"""

TEXTS = [
    "产品不错但价格偏高",
    "客服态度极差，问题拖了两周没解决"
]

async def test(model: str, text: str):
    t = time.time()
    resp = await client.chat.completions.create(
        model=model,
        response_format={"type": "json_object"},
        messages=[{"role": "system", "content": PROMPT}, {"role": "user", "content": text}]
    )
    return json.loads(resp.choices[0].message.content), round(time.time() - t, 2)

async def main():
    models = ["gpt-4o-mini", "gpt-4o", "claude-sonnet-4-20250514"]
    for text in TEXTS:
        print(f"\n文本：{text}")
        for m in models:
            result, elapsed = await test(m, text)
            print(f"  {m:<35} {elapsed}s  {result}")

asyncio.run(main())