match-map

match-map 用 LLM 帮你把多种来源、不同结构的数据，持续归一化为一个确定的目标结构。

你只需定义目标 Pydantic schema。正常请求走快速、确定性的 match → map 规则；遇到新格式或坏规则时，LLM Agent 会读取失败数据和现有规则，调试并生成新的 match/map 版本。修复通过校验后，后续同类数据不再调用 LLM，直接复用已激活的 Python 规则。

它适合统一供应商 payload、历史 JSON、嵌套对象、日志文本和文档字段等异构数据。包根只提供一个业务入口 MatchMap，目标始终是同一个调用方定义的 schema。

安装

仅使用确定性转换流程：

pip install match-map

启用 Pydantic AI 自动修复：

pip install 'match-map[agent]'

从源码运行仓库示例：

pip install -e '.[examples]'

快速开始

from pydantic import BaseModel
from match_map import (
    LocalStorageConfig,
    MapRuleConfig,
    MatchMap,
    MatchMapConfig,
    RuleSource,
    RulesConfig,
    SandboxConfig,
)


class Person(BaseModel):
    name: str
    age: int


config = MatchMapConfig(
    storage=LocalStorageConfig(path=".match_map/person-rules"),
    sandbox=SandboxConfig(timeout=1.0, max_output_bytes=1_000_000),
    rules=RulesConfig(
        match=RuleSource(source="""
def match(ctx, data, available_maps):
    if isinstance(data, dict) and "full_name" in data:
        return "flat"
    raise ValueError("unsupported input")
"""),
        maps=(
            MapRuleConfig(name="flat", rule=RuleSource(source="""
def map(ctx, data):
    return {"name": data["full_name"], "age": data["years"]}
""")),
        ),
    ),
)

app = MatchMap(Person, config)

person = app.run({"full_name": "Ada", "years": 36})

run() 成功时返回目标 Pydantic 模型实例。无法路由时抛出 match_map.errors.NoMatchError；map 执行或目标模型校验失败时抛出 match_map.errors.MapConversionError。

规则的标准函数原型是 match(ctx, data, available_maps) 和 map(ctx, data)。ctx 是普通 dict，并且只包含 ctx["config"]，其值是调用方提供的 MatchMapConfig.func_config dict。 SDK 不会自动注入 LLM 或其他隐式数据；需要时由调用方显式放入 func_config["llm"]。 func_config 仅接受 JSON 可序列化数据，例如 MatchMapConfig(func_config={"tenant": "acme", "strict": True})。仓库内示例均已迁移到新 原型；当前版本仍暂时兼容已有的无 ctx 规则。

工作原理

match-map 包含两个基本步骤：match 和 map。map 用于完成任意数据转换，大多数时候是 JSON 字段的抽取和转换，偶尔也会伴随字符串拼接或其他操作；match 则针对不同的数据源，选择对应的 map 函数。

在此基础上，任何数据转换都可以被抽象成一个 match-map 过程。有了 LLM，尤其是 ReAct Agent 的能力后，我们可以让 Agent 自行观察数据、分析转换失败的原因，并决定如何构建或修改 match 和 map 函数，从而允许我们对任意数据源进行建模。

flowchart LR
    subgraph MM["Match-Map 转换流"]
        A["任意数据源"] --> B["match 识别数据并选择 map"]
        B --> C["map 抽取、转换或组合字段"]
        C --> D{"目标 Pydantic schema 校验"}
        D -->|"成功"| E["统一目标数据"]
        D -->|"失败"| F["失败数据"]
        B -->|"无法匹配"| F
    end

    subgraph RA["Agent ReAct 修复循环"]
        G["Observe：读取失败数据和现有规则"] --> H["Reason：分析数据结构与失败原因"]
        H --> I["Act：创建或更新 match/map"]
        I --> J["Observe：Sandbox 执行、schema 与回归验证"]
        J -->|"仍然失败"| H
        J -->|"验证通过"| K["激活新规则版本"]
    end

    F --> G
    K --> B

这两个流程彼此解耦：正常数据只经过确定性的 match-map 转换；失败数据才进入 Agent ReAct 循环。Agent 生成的是可复用的 Python 规则，最终结果仍由真实执行、目标 schema 和回归测试共同验证。

配置

下面是仓库 examples/config.example.yaml 使用的完整基础配置：

agent:
  enabled: true
  verbose: false
  extra_tool: []
  max_rounds: 5
  max_batch_rounds: 5
  failure_path: failures
  max_case_attempts: 3
  poll_interval: 1.0
  processing_timeout: 3600
  instructions: null
  prompt_template: null
  llm:
    provider: openai
    api: responses
    model: gpt-5.6-terra
    reasoning_effort: medium
    timeout: 60

sandbox:
  python_path: null
  timeout: 2.0
  max_output_bytes: 1000000

YAML 需要先转换为强类型配置；MatchMap 不接受普通 dict：

import yaml
from match_map import MatchMap, MatchMapConfig

raw = yaml.safe_load(open("examples/config.yaml", encoding="utf-8"))
config = MatchMapConfig.model_validate(raw)
app = MatchMap(Person, config)

不同场景如何配置

只使用已有规则，不启用 LLM

agent:
  enabled: false

适合规则已经稳定、生产环境不允许自动修改，或只想使用确定性 match → map 转换的场景。

使用 OpenAI Responses API

agent:
  enabled: true
  llm:
    provider: openai
    api: responses
    model: gpt-5.6-terra
    reasoning_effort: medium
    timeout: 60

省略 api_key 时使用 OPENAI_API_KEY 环境变量，避免把密钥写进配置文件。

使用第三方 OpenAI-compatible Chat API

agent:
  enabled: true
  llm:
    provider: openai
    api: chat
    base_url: https://example.com/compatible-mode/v1
    api_key: your-api-key
    model: your-model-name
    reasoning_effort: null
    timeout: 60

大多数兼容服务使用 api: chat；只有服务明确支持 /responses 时才选择 responses。不支持推理强度参数的模型应设置 reasoning_effort: null。

指定 match/map 子进程使用的 Python

sandbox:
  python_path: /absolute/path/to/python
  timeout: 2.0
  max_output_bytes: 1000000

路径必须指向存在且可执行的 Python；null 表示使用当前进程的 sys.executable。动态规则运行在短生命周期受限子进程中，并经过 AST、导入白名单、超时和输出大小检查。这是降低风险的隔离，不是可安全执行敌意代码的强 sandbox。

从指定目录加载初始规则并保存版本

from match_map import LocalStorageConfig, MapRuleConfig, MatchMapConfig, RuleSource, RulesConfig

config = MatchMapConfig(
    storage=LocalStorageConfig(path=".match_map/person-rules"),
    rules=RulesConfig(
        match=RuleSource(path="initial/match.py"),
        maps=(
            MapRuleConfig(name="vendor_a", rule=RuleSource(path="initial/maps/vendor_a.py")),
            MapRuleConfig(name="vendor_b", rule=RuleSource(path="initial/maps/vendor_b.py")),
        ),
    ),
)

storage.path 同时保存 ACTIVE 和 v1/、v2/ 等版本。相对规则路径以 storage.path 为基准；也可以通过 RuleSource(source="...") 直接提供源码，或用 rules.version 切换到已有版本。空规则文件会转换成明确报错、可供 Agent 修复的占位函数。

覆盖 Agent 提示词或增加应用工具

from match_map import AgentConfig, MatchMapConfig

agent = AgentConfig(
    enabled=True,
    instructions="你的完整 Agent instructions",
    prompt_template="修复以下状态：\n{state}",
    extra_tool=(my_custom_tool,),
    regression_items=(known_good_record,),
)
config = MatchMapConfig(agent=agent)

prompt_template 必须包含 {state}。extra_tool 只能通过 Python 注入 callable，YAML 中应保持 []。verbose: true 会把入队、修复状态和工具调用写到 stderr，但不会打印原始 case 内容。

Agent 修复与失败队列

数据转换和 Agent 修复是两个互不等待的循环。run() 不调用 LLM：转换失败时先把每条 case 原子写入本地 JSON 文件，然后仍然抛出原转换异常。应用可以在另一个 asyncio task 或独立 worker 进程中消费队列：

import asyncio

# 处理调用时已经存在的 pending 文件，然后返回统计值
summary = asyncio.run(app.repair_pending(max_cases=100))

# 服务进程中持续消费；设置 stop_event 后优雅停止
async def worker():
    stop_event = asyncio.Event()
    await app.repair_loop(stop_event=stop_event, max_cases_per_poll=100)

每个 case 始终对应一个文件，目录状态为：

rules/failures/
├── pending/       # 等待 Agent
├── processing/    # 已被一个 worker 原子领取
├── resolved/      # 修复成功，包含转换结果和规则版本
└── failed/        # 达到 max_case_attempts，保留最后错误

repair_pending() 对当前快照逐文件处理，LLM 不会一次接收整个失败数据集。失败但仍可重试 的 case 会返回 pending；持续 worker 会在后续轮次重试。

Agent 循环由 pydantic_ai.Agent 实现。以下六个 function tool 是 match-map 默认注册的修复工具：

推荐调用顺序是 ls_data → read_data → read_config → update_config。如果新建 map，通常还需再次调用 update_config(name="match", ...)，使 match 能将对应输入路由到新 map。

这六个是默认工具，不包含 native、output、MCP、toolset 或 capability 工具。通过 Python 配置的 AgentConfig.extra_tool 可以额外注册应用工具，因此配置了 extra_tool 时，Agent 的实际工具集会更多。

初始失败状态只包含失败数量和全部可用配置名称，不注入文件名、错误、源码或原始数据。Agent 先调用 ls_data，再按需批量调用 read_data 和 read_config 自行诊断。

持久记忆文件固定为 <storage.path>/MEMORY.md。每次 Agent batch loop 开始时都会重新从 磁盘读取，并使用 MEMORY_PROMPT_TEMPLATE 拼接到 system instructions；空记忆会注入 (empty)。因此一个 loop 通过 update_memory 写入的经验会在下一个 loop 自动生效，且 记忆不会绕过规则执行、回归测试或目标 schema 校验。

max_rounds 映射为 Pydantic AI UsageLimits 的请求与工具调用上限；max_batch_rounds 控制失败数据的批量回放次数。兼容 OpenAI 的第三方服务通常应选择 api: chat；支持 Responses API 的服务可选择 api: responses。

instructions 和 prompt_template 用于覆盖默认提示词。prompt_template 必须包含 {state}，运行时会将其替换为目标 schema、可用配置名称和失败数量组成的 JSON；规则源码、错误详情和原始数据由 Agent 通过默认工具按需读取。省略或设为 None 时使用导出的全局默认提示词。

提示词以包级常量导出：

from match_map import AGENT_INSTRUCTIONS, MEMORY_PROMPT_TEMPLATE, REPAIR_PROMPT_TEMPLATE

示例

仓库包含七个可运行场景，覆盖规则正确、map 错误、match 错误、规则全空、文本字段提取、字符串内 JSON 提取，以及在 map 中直接调用 Pydantic AI。完整说明见 examples/README.md。

python examples/case_01_correct.py
python examples/case_02_bad_map.py
python examples/case_03_bad_match.py
python examples/case_04_empty.py
python examples/case_05_text_extraction.py
python examples/case_06_string_to_json.py
python examples/case_07_pydantic_ai_map.py

# 清除七个示例产生的版本，一键恢复初始状态
python examples/reset_cases.py

测试

pip install -e '.[test]'
pytest

真实模型集成测试默认跳过；显式设置 MATCH_MAP_RUN_OPENAI=1 和相应 API Key 后才会运行付费请求。