Legora 如何用 GPT-6 Astra 在几分钟内审查 41 份财务文件

#!/usr/bin/env python3
import json
import os
import sys
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

BASE_URL = os.environ.get("LLM_BASE_URL", "https://example.com/v1")
API_KEY = os.environ.get("LLM_API_KEY")
MODEL = os.environ.get("LLM_MODEL", "replace-with-your-review-model")
DOC_DIR = Path(sys.argv[1] if len(sys.argv) > 1 else "documents")

SYSTEM_PROMPT = """You are reviewing financial documents for factual inconsistencies.
Return JSON only with this shape:
{
  "findings": [
    {
      "category": "numeric_mismatch|period_mismatch|missing_disclosure|calculation_error|other",
      "severity": "low|medium|high",
      "claim": "short description",
      "evidence": "exact quotation from the document",
      "location": "page, section, or paragraph if available",
      "confidence": 0.0
    }
  ]
}
Do not invent evidence. If evidence is insufficient, return an empty findings array.
"""

def review(path: Path) -> dict:
    text = path.read_text(encoding="utf-8")
    payload = {
        "model": MODEL,
        "temperature": 0,
        "response_format": {"type": "json_object"},
        "messages": [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"FILE: {path.name}\n\n{text}"},
        ],
    }
    request = urllib.request.Request(
        f"{BASE_URL.rstrip('/')}/chat/completions",
        data=json.dumps(payload).encode("utf-8"),
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=180) as response:
        result = json.load(response)
    parsed = json.loads(result["choices"][0]["message"]["content"])
    return {"file": path.name, "findings": parsed.get("findings", [])}

if not API_KEY:
    raise SystemExit("Set LLM_API_KEY before running")

paths = sorted([*DOC_DIR.glob("*.txt"), *DOC_DIR.glob("*.md")])
if not paths:
    raise SystemExit(f"No .txt or .md files found in {DOC_DIR}")

reports = []
with ThreadPoolExecutor(max_workers=4) as pool:
    futures = {pool.submit(review, path): path for path in paths}
    for future in as_completed(futures):
        path = futures[future]
        try:
            reports.append(future.result())
        except Exception as exc:
            reports.append({"file": path.name, "error": str(exc), "findings": []})

reports.sort(key=lambda item: item["file"])
Path("review-results.json").write_text(
    json.dumps(reports, ensure_ascii=False, indent=2),
    encoding="utf-8",
)
print(f"Reviewed {len(reports)} documents; results written to review-results.json")

export LLM_BASE_URL="https://your-provider.example/v1"
export LLM_API_KEY="your-api-key"
export LLM_MODEL="your-review-model"
python3 review_documents.py documents