企业官网建设流程全解析

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/txtai

txtai 的 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 应用实例。扩展的全部工作就是在该方法内完成两件事之一：

1. 通过app.include_router(router)注册自定义路由（推荐方式）；
2. 直接修改 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 EXTENSIONS="testapi.testextension.SampleExtension"

启动 API 服务后，GET /sample?text=Test%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.py：

class 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_code=401, detail="Invalid 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、正确 token（Bearer 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_code=403, detail="Forbidden")

启用方式：

export DEPENDENCIES="myapp.auth.CustomAuth"

create()中的装配逻辑如下（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 定制与运行相关的全部环境变量如下：

定制与 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暴露一个调用这些组件的高级聚合端点，最后用TOKEN+DEPENDENCIES为该端点及全部路由加上鉴权。

实战：完整可运行的最小定制服务

综合以上机制，一个最小可运行的定制 API 服务如下。

目录结构：

myapi/ ├── config.yml # API YAML 配置 ├── ext.py # 扩展与依赖定义 └── Dockerfile # 可选：容器化部署

config.yml：

path: index writable: true embeddings: path: sentence-transformers/all-MiniLM-L6-v2 content: true summary:

ext.py：

from 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_code=403, detail="Invalid 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 CONFIG="config.yml" export TOKEN="$(printf 'my-secret' | sha256sum | cut -d' ' -f1)" export DEPENDENCIES="ext.HeaderAuth" export EXTENSIONS="ext.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会在启动阶段直接抛错。
• 基类约束差异：扩展强制继承Extension（APIFactory.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