LangExtract 自定义 Provider 插件开发指南：从路由注册到 Schema 约束的完整实践

LangExtract 的模型后端并非封闭设计：它通过一个基于正则路由与 entry point 的插件系统，允许开发者用自己的模型后端扩展库能力。本文以仓库中 官方 Provider 插件示例 为主体，系统讲解一个自定义 Provider 插件的目录结构、注册机制、结构化输出 Schema 的接入流程，以及从脚手架生成、安装测试到发布的完整步骤。读完本文，你可以独立为 LangExtract 编写一个可被 lx.extract() 直接调用的自定义后端，并正确处理结构化输出的 Schema 生命周期。

插件示例的定位与目录结构

examples/custom_provider_plugin/ 目录是 LangExtract 仓库内置的一个参考示例，用于演示如何创建自定义 Provider 插件。需要注意它在仓库 README 中明确声明：该示例不属于 LangExtract 包本身，执行 pip install langextract 时不会随之安装，仅作为参考实现随仓库发布。

示例目录组织如下（见 目录结构说明）：

custom_provider_plugin/
├── pyproject.toml                      # 包配置与元数据
├── README.md                            # 说明文档
├── langextract_provider_example/        # Python 包目录
│   ├── __init__.py                     # 包初始化（导出 Provider 类）
│   ├── provider.py                     # 自定义 Provider 实现
│   └── schema.py                       # 自定义 Schema 实现（可选）
└── test_example_provider.py            # 测试脚本

其中 init.py 只做一件事——将 Provider 类导出为包的公共接口：

from langextract_provider_example.provider import CustomGeminiProvider

__all__ = ["CustomGeminiProvider"]
__version__ = "0.1.0"

这一点很关键：pyproject.toml 中的 entry point 指向 langextract_provider_example:CustomGeminiProvider，插件加载器会导入包并在包级别查找该名字，因此 __init__.py 中的 re-export 是插件能被发现的必要条件。

Provider 核心实现：注册装饰器与 infer 契约

插件的骨架在 provider.py 中，最小可用的注册形态为：

from langextract.core import base_model
from langextract.providers import router

@router.register(
    r'^gemini',  # 该 Provider 处理的 model ID 模式
)
class CustomGeminiProvider(base_model.BaseLanguageModel):
    def __init__(self, model_id: str, **kwargs):
        # 初始化你的后端客户端

    def infer(self, batch_prompts, **kwargs):
        # 调用后端 API 并返回结果

示例的完整实现在此基础上展示了几个值得注意的细节：

1. dataclass 风格的字段声明。示例类使用 @dataclasses.dataclass(init=False) 显式声明了 model_id、api_key、temperature、response_schema、enable_structured_output 等字段（见 provider.py#L53-L58），其中 _client 用 repr=False, compare=False 排除在比较与打印之外——这是一种保持实例可比较、可调试的常见手法。
2. 依赖的惰性导入与友好报错。构造函数中先 import google.genai，失败时抛出 lx.exceptions.InferenceConfigError 并提示 pip install google-genai（provider.py#L77-L104）。API key 缺失时同样抛出配置类异常而非在推理阶段才失败。
3. infer 的迭代器契约。infer(batch_prompts, **kwargs) 接收一个 prompt 序列，对每个 prompt 逐个调用后端并以 yield [types.ScoredOutput(score=1.0, output=output)] 的形式产出结果（provider.py#L144-L184）；API 异常统一包装为 lx.exceptions.InferenceRuntimeError 并保留 original 异常链。max_output_tokens、top_p、top_k 等参数按需从 kwargs 透传到请求配置中。

从源码结构看，示例 Provider 只是把 Gemini 调用"套了一层壳"，目的是演示 Schema 如何贯穿整个调用链；实际开发时，infer 内部的 API 调用与客户端初始化都应替换为你自己的后端实现。

路由系统如何工作

@router.register(*patterns, priority=0) 的底层实现在 router.py 中。理解它可以帮你解释后文的"模式冲突"与"显式指定 Provider"两个话题：

• 注册即惰性。register 装饰器不会导入任何东西，它把编译后的正则、类加载器和 provider_id（模块路径:类名）存入全局 _entries 列表，并按 (provider_id, patterns, priority) 三元组去重（router.py#L108-L135）。
• 按优先级解析。resolve(model_id) 先将所有条目按 priority 降序排列，取第一个"任一 pattern 能 search 命中"的 Provider 返回；未命中时抛出 InferenceConfigError，错误信息中列出所有可用 pattern 并提示可通过 ModelConfig 显式指定 Provider（router.py#L138-L166）。resolve 与 resolve_provider 都带有 lru_cache(maxsize=128)，测试中清理注册表后需注意缓存失效。
• 按名字解析。resolve_provider(provider_name) 支持用 entry point 名字或类名（不区分大小写）定位 Provider 类，这正是 provider="CustomGeminiProvider" 能生效的路径（router.py#L169-L214）。
• 仓库另提供 registry = router 的向后兼容别名（providers/__init__.py），旧代码中的 lx.providers.registry.register(...) 写法仍然有效。

包配置：entry point 是插件被发现的钥匙

pyproject.toml 的关键片段：

[project]
name = "langextract-provider-example"  # 改成你的包名
version = "0.1.0"
requires-python = ">=3.10"
dependencies = [
    # 独立发布时放开注释，让插件自动安装 langextract：
    # "langextract",
    "google-genai>=0.2.0",  # 替换为你的后端 SDK
]

# 将 Provider 注册到 LangExtract 插件系统
[project.entry-points."langextract.providers"]
custom_gemini = "langextract_provider_example:CustomGeminiProvider"

[tool.setuptools.packages.find]
where = ["."]
include = ["langextract_provider_example*"]

这行 entry point 是 LangExtract 自动发现插件的依据。其加载链路在 providers/__init__.py 中实现：

• load_plugins_once() 是幂等的，且支持环境变量 LANGEXTRACT_DISABLE_PLUGINS（取值 1/true/yes）全局禁用插件加载（providers/__init__.py#L71-L88）；
• 它会先调用 load_builtins_once() 用 router.register_lazy 惰性注册内置 Provider（gemini、ollama、openai 等），再通过 importlib.metadata.entry_points() 扫描 langextract.providers 组，逐个 entry_point.load() 导入插件包（providers/__init__.py#L90-L120）；
• factory.py 在创建模型前会自动触发 providers.load_plugins_once()（factory.py#L156），因此正常业务代码无需手动加载；但测试脚本中需要显式调用，这是后文"常见陷阱"第一条的根源。

另外，plugins.py 中的 available_providers() 展示了内置 Provider 的优先级语义：核心内置 Provider（gemini、ollama）默认优先级最高，第三方插件只有在 allow_override=True 时才能覆盖同名内置项——默认情况下内置 Provider 总是胜出，这也是"模式冲突"要小心的原因之一。

自定义 Schema 支持：从示例到结构化输出的完整链路

结构化输出是 Provider 插件中相对进阶的部分。数据流为：

Examples → from_examples() → to_provider_config() → Provider kwargs → Inference

Schema 的抽象基类定义在 core/schema.py：BaseSchema 声明了三个必须实现的抽象成员——类方法 from_examples(examples_data, attribute_suffix)、实例方法 to_provider_config() 以及属性 requires_raw_output（core/schema.py#L39-L81），此外还提供 from_schema_dict()（用于用户显式传入 output_schema 的场景）与 sync_with_provider_kwargs()（响应调用方参数覆盖的钩子）等可选扩展点。

示例的 schema.py 展示了完整实现：

1. from_examples：从示例数据归纳约束。 方法遍历每个 ExampleData 的 extractions，收集所有出现过的 extraction_class 与属性键，生成一份 JSON Schema：extractions 为数组，每个元素要求 extraction_class（若示例中出现过具体类别则附加 enum 约束）与 extraction_text 必填（schema.py#L51-L122）。这意味着模型在推理时只会输出你标注示例中出现过的实体类别与属性，约束完全由示例"学会"。

2. to_provider_config：将 Schema 翻译成 Provider 参数。 示例返回 {"response_schema": schema_dict, "enable_structured_output": True, "output_format": "json"}（schema.py#L124-L144）。文档特别注明：这些 kwargs 会与用户传入的 kwargs 合并，且用户值优先（caller-wins merge 语义）。

3. requires_raw_output：决定输出是否需要围栏标记。 返回 True 表示 Provider 保证产出语法合法的裸 JSON（LangExtract 无需为其添加 markdown 围栏）；返回 False 则输出会包裹围栏（schema.py#L146-L154）。该属性是 BaseSchema 的必实现抽象属性。

Provider 侧的对接只有两处。其一，声明 Schema 类：

@classmethod
def get_schema_class(cls):
    return custom_schema.CustomProviderSchema  # 告知 LangExtract 你的 Schema

其二，实现 apply_schema 处理"应用/清除"双向状态（provider.py#L118-L142）：

def apply_schema(self, schema_instance):
    super().apply_schema(schema_instance)
    if schema_instance:
        config = schema_instance.to_provider_config()
        self.response_schema = config.get('response_schema')
        self.enable_structured_output = config.get('enable_structured_output', False)
    else:
        # 必须清理状态，否则上一次的结构化约束会"泄漏"到下一次推理
        self.response_schema = None
        self.enable_structured_output = False

None 分支对应 README"常见陷阱"第 5 条：LangExtract 会在 use_schema_constraints 生效时动态调用 apply_schema，Provider 必须在收到 None 时清空 Schema 状态。最后在 infer 中，当 response_schema 与 enable_structured_output 同时为真时，把 response_schema 与 response_mime_type="application/json" 一并放入请求配置（provider.py#L165-L170）。

安装与测试插件

示例插件的安装与验证流程（README Installation 一节）：

# 先进入示例目录
cd examples/custom_provider_plugin

# 以开发模式安装
pip install -e .

# 运行 Provider 测试（必须在该目录下执行）
python test_example_provider.py

test_example_provider.py 的测试逻辑值得参照：

# 手动导入 Provider 以触发注册。
# 注意：仅在不安装时靠手动导入才需要；
# 执行 `pip install -e .` 后，entry point 系统会自动完成这件事。
from langextract_provider_example import CustomGeminiProvider  # noqa: F401

import langextract as lx

def main():
  dotenv.load_dotenv(override=True)
  api_key = os.getenv("GEMINI_API_KEY") or os.getenv("LANGEXTRACT_API_KEY")
  if not api_key:
      print("Set GEMINI_API_KEY or LANGEXTRACT_API_KEY to test")
      return

  config = lx.factory.ModelConfig(
      model_id="gemini-3.5-flash",
      provider="CustomGeminiProvider",
      provider_kwargs={"api_key": api_key},
  )
  model = lx.factory.create_model(config)
  print(f"✓ Created {model.__class__.__name__}")

  results = list(model.infer(["Say hello"]))
  if results and results[0]:
      print(f"✓ Inference worked: {results[0][0].output[:50]}...")

它验证了两条链路：ModelConfig + create_model 的显式 Provider 解析，以及 infer 迭代器的真实产出（ScoredOutput.output 可访问）。

在 extract() 中使用自定义 Provider

由于示例注册的正则 ^gemini 与内置 Gemini Provider 相同，model_id="gemini-*" 的自动解析默认会命中内置 Provider（内置项优先级更高）。因此必须显式指定 Provider。README 给出两种等价用法：

import langextract as lx

# 方式 A：显式构建模型，再传入 extract()
config = lx.factory.ModelConfig(
    model_id="gemini-3.5-flash",
    provider="CustomGeminiProvider",
    provider_kwargs={"api_key": "your-api-key"},
)
model = lx.factory.create_model(config)

result = lx.extract(
    text_or_documents="Your text here",
    model=model,
    prompt_description="Extract key information",
    examples=[...],
)

# 方式 B：直接让 extract() 依据 ModelConfig 构建模型
result = lx.extract(
    text_or_documents="Your text here",
    config=lx.factory.ModelConfig(
        model_id="gemini-3.5-flash",
        provider="CustomGeminiProvider",
        provider_kwargs={"api_key": "your-api-key"},
    ),
    prompt_description="Extract key information",
    examples=[...],
)

两种方式最终都经由 router.resolve_provider 按名字找到 CustomGeminiProvider。如果你的模式不与内置 Provider 重叠（例如 ^myprovider），则无需显式指定，resolve() 会自动命中。

创建自己的 Provider：脚手架脚本

README 建议不要手工拷贝示例，而是使用仓库内置的插件生成脚本：

python scripts/create_provider_plugin.py MyProvider --with-schema

该脚本会生成一个包含全部样板代码的插件骨架，就绪待定制。从脚本源码看，它支持以下参数：

参数	说明
provider_name（位置参数）	Provider 名称，如 MyProvider、CustomLLM
--patterns	一个或多个 model ID 正则（默认：^{package_name}）
--package-name	包名（默认取 Provider 名的小写形式）
--with-schema	生成 Schema 支持（对应手动流程的第 4 步）
--no-install	跳过自动安装与测试
--force	允许覆盖已存在的插件目录

脚本的文档字符串说明它自动完成了"Provider 创建清单"的前 6 步：建立包结构、配置 entry point、生成 Provider 实现（默认 mock 推理，环境变量命名规则为 {PACKAGE}_API_KEY）、可选生成 Schema、创建测试脚本、生成 README，并附带 .gitignore 与 LICENSE 占位文件（create_provider_plugin.py#L16-L33、scripts/create_provider_plugin.py#L737-L766）。除非指定 --no-install，脚本还会自动执行 pip install -e . 并运行生成的 test_plugin.py，测试覆盖注册/模式匹配、推理、Schema 应用与 factory 集成等检查项（create_provider_plugin.py#L604-L635）。

手动创建流程（七步）

如果不使用脚手架，README 给出的手动流程可完整继承如下：

1. 拷贝并重命名

# 拷贝示例目录
cp -r examples/custom_provider_plugin/ ~/langextract-myprovider/

# 重命名包目录
cd ~/langextract-myprovider/
mv langextract_provider_example langextract_myprovider

2. 更新包配置（编辑 pyproject.toml）：

• 改 name = "langextract-myprovider"；
• 更新 description 与作者信息；
• 修改 entry point：myprovider = "langextract_myprovider:MyProvider"。

3. 修改 Provider 实现（编辑 provider.py）：

• 类名从 CustomGeminiProvider 改为 MyProvider；
• 更新 @router.register(...) 的模式以匹配你的 model ID；
• 将 Gemini API 调用替换为你的后端；
• 添加 Provider 特有参数。

4. 添加 Schema 支持（可选）（编辑 schema.py）：

• 类名改为 MyProviderSchema；
• 按你的提取格式定制 from_examples()；
• 按你的 API 要求更新 to_provider_config()；
• 根据 Provider 是否输出裸 JSON/YAML 还是围栏输出实现 requires_raw_output（BaseSchema 的抽象属性）。

5. 安装并验证注册：

pip install -e .

python -c "
from langextract.providers import load_plugins_once, router
load_plugins_once()
print('Provider registered:', any('myprovider' in str(e) for e in router.list_entries()))
"

这里的 router.list_entries() 会返回所有注册条目的 (patterns, priority) 列表（router.py#L238-L244），是确认插件加载成功最直接的调试手段；load_plugins_once() 的显式调用则是测试场景下的必备动作。

6. 编写测试：验证 Provider 能加载并完成基础推理；如实现了 Schema，验证其生成与应用；针对你自己的 API 测试错误处理路径。

7. 发布到 PyPI 并与社区共享：

python -m build
twine upload dist/*

社区侧的落地动作是向 COMMUNITY_PROVIDERS.md 提交 PR，把你的 Provider 加入社区 Provider 注册表，并通过项目仓库的 issue 渠道发布公告、获取反馈。发布前记得把 pyproject.toml 中注释掉的 "langextract" 依赖放开，让使用者 pip install 你的插件时自动获得 LangExtract 本体。

常见陷阱与源码级解释

README 列出 5 个高频陷阱，均可在源码中找到对应机制：

1. 忘记触发插件加载——插件是惰性加载的，load_plugins_once() 幂等且带全局开关，测试脚本与调试 REPL 中应显式调用一次（providers/__init__.py#L71-L88）。
2. 模式冲突——resolve() 按 priority 降序取首个命中项；与内置 Provider 使用相同 pattern 时，默认解析会倾向于内置项。想稳定命中自己的 Provider，要么使用不重叠的 pattern，要么在 ModelConfig 中显式指定 provider（router.py#L154-L166）。
3. 遗漏依赖——插件包自身的依赖（后端 SDK）必须完整列入 pyproject.toml 的 dependencies；示例中对 google-genai 缺失即抛出带安装提示的 InferenceConfigError，是好的错误设计范本。
4. Schema 不匹配——from_examples 归纳出的约束完全取决于你提供的示例；务必用真实示例数据跑一遍 Schema 生成，确认 enum、必填字段与属性键符合预期。
5. 未处理 None Schema——apply_schema(None) 是清除约束的正规通道，Provider 必须把 response_schema 等状态一并复位，否则上一次的结构化约束会泄漏到后续普通推理（provider.py#L118-L142）。

小结

LangExtract 的 Provider 插件机制由三个环节组成：pyproject.toml 的 langextract.providers entry point 负责发现，router.register 的正则路由负责解析，get_schema_class + apply_schema + BaseSchema 三件套负责结构化输出约束的下发与清理。以 examples/custom_provider_plugin/ 为参照、配合 scripts/create_provider_plugin.py 脚手架，你可以较快地完成一个可运行、可测试、可发布的自定义后端插件；深入 router.py、providers/__init__.py 与 core/schema.py 的源码，则能帮助你处理模式冲突、优先级与 Schema 生命周期等实际集成中容易踩坑的细节。