
txtai API 定制指南使用 Extensions 与 Dependencies 扩展自定义端点与请求中间件【免费下载链接】txtai All-in-one AI framework for semantic search, LLM orchestration and language model workflows项目地址: https://gitcode.com/GitHub_Trending/tx/txtaitxtai 的 API 服务开箱即用通过 YAML 配置即可快速启动 embeddings 检索、pipeline 流水线与 workflow 工作流服务。但在真实业务中往往还需要暴露自定义的业务端点、或在每次请求前附加额外的鉴权/认证逻辑。txtai 为此提供了两条官方定制路径Extensions扩展用于注册自定义 Python 端点Dependencies依赖用于注入随每个请求执行的中间件。本文将以 docs/api/customization.md 为主线结合仓库源码与测试用例完整讲解这两种机制的编写方法、加载原理与实战配置。定制机制总览两种方式一条主线根据官方文档txtai API 的两大定制手段定位如下Extensions扩展向 FastAPI 应用追加自定义端点适合把 YAML 工作流难以表达、或需要直接操作 Python 对象如自定义 pipeline的逻辑以原生端点暴露出去。Dependencies依赖为 API 注入请求级中间件在每次请求执行前运行典型场景是文档中提到的额外的授权步骤和/或认证方法。两者都通过环境变量接入应用生命周期均在 src/python/txtai/api/application.py 的lifespan启动钩子中完成装配。整个定制链路可概括为环境变量 EXTENSIONS / DEPENDENCIES │ ▼ APIFactory.get(path, base) ──► Resolver 解析类路径 基类校验 │ ▼ extension(app) / Depends(dep) ──► FastAPI 应用装配lifespan 阶段其中APIFactory.get最终调用 src/python/txtai/util/resolver.py 的Resolver它按.拆分路径、逐段__import__并getattr解析出目标类若传入base参数还会强制校验目标类必须是该基类的子类issubclass检查否则抛出ImportError。这意味着传入的类路径必须可被 Python 正常 import且应位于 API 服务进程的sys.path可达范围内。Extensions用 Python 定义自定义 API 端点基类契约所有扩展都必须继承 src/python/txtai/api/extension.py 中定义的Extension基类。该基类只声明一个钩子方法class Extension: def __call__(self, app): Hook to register custom routing logic and/or modify the FastAPI instance. Args: app: FastAPI application instance return__call__接收一个app参数即当前运行的FastAPI 应用实例。扩展的全部工作就是在该方法内完成两件事之一通过app.include_router(router)注册自定义路由推荐方式直接修改 FastAPI 实例例如添加全局中间件、异常处理器或自定义响应类型。不重写__call__的基类默认行为是空操作return测试 test/python/testapi/testextension.py 中的testEmpty用例验证了这一点Extension()(None)返回None。编写第一个扩展完整代码示例下面给出一个可直接套用的完整示例结构与仓库测试用例同构。假设我们要暴露一个GET /sample端点它调用一个自定义 pipeline 并把输入文本转为小写。第 1 步定义自定义 pipeline可选若端点逻辑不依赖 pipeline 可省略from txtai.pipeline import Pipeline class SamplePipeline(Pipeline): def __call__(self, text): return text.lower()第 2 步定义路由。推荐使用 FastAPI 的APIRouter将业务端点与主应用解耦from fastapi import APIRouter from txtai.api import application class SampleRouter: router APIRouter() staticmethod router.get(/sample) def sample(text: str): # application.get() 返回全局 API 实例 return application.get().pipeline(testapi.testextension.SamplePipeline, (text,))第 3 步定义扩展类在__call__中挂载路由from txtai.api import Extension class SampleExtension(Extension): def __call__(self, app): app.include_router(SampleRouter().router)第 4 步通过环境变量启用扩展export EXTENSIONStestapi.testextension.SampleExtension启动 API 服务后GET /sample?textTest%20String将返回test string——这正是 testextension.py 中testExtension用例断言的行为。扩展的加载原理扩展的装配发生在 application.py 的lifespan钩子中extensions os.environ.get(EXTENSIONS) if extensions: for extension in extensions.split(,): # Create instance and execute extension extension APIFactory.get(extension.strip(), Extension)() extension(application)关键细节EXTENSIONS支持逗号分隔的多个扩展类路径按顺序逐个实例化并执行APIFactory.get(..., Extension)中的第二个参数Extension就是Resolver的基类校验参数——传入的类必须继承自Extension否则启动即报错这保证了所有扩展遵守统一契约执行时机在内置路由注册完成之后因此扩展端点可以安全地与embeddings、pipeline、workflow等内置路由共存。从lifespan的整体顺序看application.py 第 76-121 行先读取CONFIG指向的 YAML → 创建 API 实例 → 按配置挂载内置 router → 再执行EXTENSIONS扩展 → 最后按需挂载 MCP 服务。扩展属于最后追加的一环可以覆盖几乎所有内置行为之后的定制需求。扩展中访问全局 API 实例在上面的路由代码中application.get()返回的是当前进程的全局 API 实例源码见 application.py 的get()函数直接返回模块级INSTANCE。通过该实例扩展端点可以调用application.get().pipeline(类路径, (参数,))执行任意已注册的自定义 pipeline访问application.get().embeddings直接做向量检索调用application.get().workflow触发工作流。这种模式让扩展既薄只做 HTTP 层适配又深底层完整复用 txtai 的 Application 能力。Dependencies按请求注入授权与认证中间件内置的默认 Token 授权txtai API 自带一套默认的token 授权Token Authorization机制。启用方式是在环境变量中设置TOKEN值为合法 token 的 SHA-256 哈希而不是明文 token 本身export TOKEN$(printf my-secret-token | sha256sum | cut -d -f1)其实现位于 src/python/txtai/api/authorization.pyclass Authorization: def __init__(self, token): self.token token def __call__(self, authorization: HTTPAuthorizationCredentials Depends(HTTPBearer())): if not hmac.compare_digest(self.token, self.digest(authorization.credentials)): raise HTTPException(status_code401, detailInvalid Authorization Token) def digest(self, token): return hashlib.sha256(token.encode(utf-8)).hexdigest()实现要点使用 FastAPI 的HTTPBearer依赖自动解析Authorization: Bearer token请求头对请求携带的 token 实时计算 SHA-256并与配置的哈希用hmac.compare_digest做常量时间比较避免时序侧信道攻击校验失败统一返回401 Invalid Authorization Token。该默认依赖在 application.py 的create()中被装配token os.environ.get(TOKEN) if token: dependencies.append(Depends(Authorization(token)))TOKEN未设置时API 不启用鉴权。测试 test/python/testapi/testauthorization.py 覆盖了三种情形无请求头返回 401、错误 token 返回 401、正确 tokenBearer token正常返回检索结果。编写自定义依赖默认 token 鉴权适合多数场景但业务上往往需要多因素认证、第三方身份服务校验、或请求上下文注入。此时通过DEPENDENCIES环境变量注入自定义依赖即可。自定义依赖的本质是一个可调用对象被 FastAPI 的Depends()包装后随每个请求执行。示例对应文档 54 号示例的主题方向class CustomAuth: def __init__(self, token): self.token token def __call__(self, authorization: HTTPAuthorizationCredentials Depends(HTTPBearer())): # 在此追加自定义认证逻辑例如调用外部身份服务、检查用户角色等 if authorization.credentials ! self.token: raise HTTPException(status_code403, detailForbidden)启用方式export DEPENDENCIESmyapp.auth.CustomAuthcreate()中的装配逻辑如下application.py 第 46-51 行deps os.environ.get(DEPENDENCIES) if deps: for dep in deps.split(,): dep APIFactory.get(dep.strip())() dependencies.append(Depends(dep))与EXTENSIONS一样DEPENDENCIES也支持逗号分隔的多个依赖不同的是这里APIFactory.get未传基类因此依赖类不强制继承某个基类只需满足可调用实现__call__ 可被 FastAPI 当作依赖注入即可。依赖与默认 Token 鉴权的叠加顺序从create()的源码可见依赖列表的构建顺序为先追加默认的TOKEN鉴权若设置了TOKEN再追加DEPENDENCIES中的自定义依赖。FastAPI 会按列表顺序依次执行这些依赖因此自定义依赖可以作为默认 token 校验通过之后的第二道防线如进一步鉴权用户角色也可以把TOKEN留空、完全由自定义依赖接管认证逻辑。这一设计正是文档所述额外的授权步骤和/或认证方法的具体落地方式。环境变量速查表结合 application.py 与 docker/api/Dockerfile整理 API 定制与运行相关的全部环境变量如下环境变量用途取值示例来源CONFIG指定 YAML 配置文件路径应用启动时读取config.ymlapplication.pyAPI_CLASS指定自定义 API 实现类继承txtai.api.API覆盖默认 APImymodule.MyAPIapplication.pyEXTENSIONS逗号分隔的扩展类路径须继承Extensionpkg.mod.MyExtensionapplication.pyDEPENDENCIES逗号分隔的自定义依赖类路径随每个请求执行pkg.mod.MyAuthapplication.pyTOKEN默认 token 鉴权合法 token 的 SHA-256 哈希不设置则关闭内置鉴权9f86d081884c7d65...application.py定制与 YAML 配置的协同扩展与依赖解决的是代码级定制而 YAML 配置解决的是声明式组装两者互补。完整的 API 顶层配置项path、writable、reindex、cloud、agent、pipeline、workflow等参见 docs/api/configuration.md其中embeddings、agent、各类 pipeline 与workflow均在启动时按 YAML 自动创建扩展端点可以通过application.get()访问这些 YAML 组装好的对象依赖中间件则守护着这些 YAML 暴露出来的所有路由包括内置路由与扩展路由。一个典型的组合用法是用 YAML 声明 embeddings 索引与 pipeline参考 docs/embeddings/configuration、docs/pipeline 与 docs/workflow 的完整配置说明再通过EXTENSIONS暴露一个调用这些组件的高级聚合端点最后用TOKENDEPENDENCIES为该端点及全部路由加上鉴权。实战完整可运行的最小定制服务综合以上机制一个最小可运行的定制 API 服务如下。目录结构myapi/ ├── config.yml # API YAML 配置 ├── ext.py # 扩展与依赖定义 └── Dockerfile # 可选容器化部署config.ymlpath: index writable: true embeddings: path: sentence-transformers/all-MiniLM-L6-v2 content: true summary:ext.pyfrom fastapi import APIRouter, Depends from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from txtai.api import Extension, application # ---------- 自定义依赖 ---------- class HeaderAuth: 校验自定义请求头作为默认 TOKEN 鉴权之外的第二道检查。 def __call__(self, authorization: HTTPAuthorizationCredentials Depends(HTTPBearer())): if not authorization.credentials.startswith(sk-): from fastapi import HTTPException raise HTTPException(status_code403, detailInvalid key prefix) # ---------- 自定义路由 ---------- class HealthRouter: router APIRouter() staticmethod router.get(/health) def health(): return {status: ok} staticmethod router.get(/summarize) def summarize(text: str): return application.get().pipeline(summary, (text,)) # ---------- 自定义扩展 ---------- class AppExtension(Extension): def __call__(self, app): app.include_router(HealthRouter().router)启动export CONFIGconfig.yml export TOKEN$(printf my-secret | sha256sum | cut -d -f1) export DEPENDENCIESext.HeaderAuth export EXTENSIONSext.AppExtension uvicorn --host 0.0.0.0 txtai.api:app此时 API 同时具备内置 embeddings 检索与 summary 流水线路由、/health与/summarize两个自定义端点、以及Bearer token 校验 自定义请求头前缀校验两道请求防线。容器化部署时可直接参考 docker/api/Dockerfile 的模式将config.yml复制进镜像用RUN python -c from txtai.api import API; API(config.yml, False)预缓存模型再以uvicorn txtai.api:app作为入口扩展与依赖类需一并打包进镜像如通过 pip 安装自定义包并通过EXTENSIONS/DEPENDENCIES环境变量在运行时注入。注意事项与最佳实践类路径必须可解析EXTENSIONS/DEPENDENCIES中的类路径依赖 Python import 机制务必确保模块位于进程sys.path中否则Resolver会在启动阶段直接抛错。基类约束差异扩展强制继承ExtensionAPIFactory.get(path, Extension)会做issubclass校验依赖则无基类约束只要求可调用。鉴权安全性内置Authorization采用 SHA-256 哈希比对 hmac.compare_digest常量时间比较切勿把明文 token 直接写入TOKEN自定义依赖若涉及敏感校验也应遵循同样的安全实践。扩展示例参考仓库中的 examples/51_Custom_API_Endpoints.ipynb 提供了自定义端点的完整 Notebook 演示examples/54_API_Authorization_and_Authentication.ipynb 则演示了授权、认证与中间件依赖的完整示例可作为深入学习的起点。回归验证官方测试 testextension.py 与 testauthorization.py 覆盖了扩展挂载、空扩展、无效/有效 token 等关键路径编写自定义扩展与依赖后可参照这两个文件补充自己的单元测试。【免费下载链接】txtai All-in-one AI framework for semantic search, LLM orchestration and language model workflows项目地址: https://gitcode.com/GitHub_Trending/tx/txtai创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考