Hindsight Extensions 扩展机制实战指南从多租户隔离到自定义 MCP 工具【免费下载链接】hindsightHindsight: Agent Memory That Learns项目地址: https://gitcode.com/GitHub_Trending/hindsight2/hindsightHindsight 通过一套基于环境变量驱动的扩展Extensions体系允许在不修改核心代码的前提下自定义服务端行为覆盖多租户隔离、自定义鉴权、额外 HTTP 端点与操作钩子四类场景。本文以 hindsight-docs/versioned_docs/version-0.6/developer/extensions.md 为主线结合hindsight-api-slim中扩展基类的源码实现完整讲解四种内置扩展类型、自定义扩展的编写范式、延迟操作DeferOperation语义与 Docker/裸机部署方式让读者可以独立实现并上线一个生产可用的 Hindsight 扩展。扩展机制总览环境变量即插件入口Hindsight 扩展的本质是通过环境变量指定的 Python 类。其加载逻辑位于 extensions/loader.py 的load_extension函数系统读取HINDSIGHT_API_TYPE_EXTENSION环境变量格式为module.path:ClassName动态导入模块、校验类继承关系、收集配置并实例化。扩展配置的传递遵循前缀剥离规则。load_extension会扫描所有以HINDSIGHT_API_TYPE_开头的环境变量排除EXTENSION变量本身剥离前缀并转为小写后作为config字典传给扩展构造函数HINDSIGHT_API_OPERATION_VALIDATOR_EXTENSIONmypackage.validators:MyValidator HINDSIGHT_API_OPERATION_VALIDATOR_MAX_REQUESTS100 # 扩展收到 config {max_requests: 100}所有扩展类都继承自 extensions/base.py 中的抽象基类Extension并共享两个生命周期钩子on_startup()—— 应用启动时调用适合连接外部服务、预热缓存on_shutdown()—— 应用关闭时调用适合释放连接、清理资源。扩展还可以通过set_context()获得一个ExtensionContext见 extensions/context.py它提供受控的系统访问 API而非直接暴露内部组件run_migration(schema)—— 为指定 schema 运行数据库迁移自动创建 schema、执行全部待执行迁移并通过 advisory lock 协调分布式 workerget_memory_engine()—— 获取MemoryEngineInterface用于执行 retain、recall、reflect 等记忆操作。DefaultExtensionContext是内置实现它在迁移完成后还会调用租户扩展声明的provision_bank_tables使扩展自有表与核心 schema 保持同一生命周期。四大扩展类型与内置实现TenantExtension多租户隔离与鉴权TenantExtension负责处理多租户和 API Key 鉴权。它校验入站请求并决定数据库操作使用哪个 PostgreSQL schema从而实现数据库层面的租户隔离。其抽象基类定义在 extensions/tenant.py核心方法包括authenticate(context: RequestContext) - TenantContext—— 校验请求并返回TenantContext(schema_name...)后续所有查询都会使用schema_name.memory_units这样的全限定表名list_tenants() - list[Tenant]—— 供 worker 发现所有租户 schema进行任务轮询authenticate_mcp(context)—— MCP 端点的鉴权入口默认委托给authenticate()可覆写以实现 HTTP 与 MCP 的差异化鉴权。鉴权失败时抛出AuthenticationError(reason, headersNone)其headers字典会被透传到 HTTP 与 MCP 的错误响应中非常适合返回WWW-Authenticate等自定义头。内置实现一ApiKeyTenantExtension源码见 extensions/builtin/tenant.py。它校验请求的 API Key 是否与HINDSIGHT_API_TENANT_API_KEY匹配并为所有通过鉴权的请求使用同一个 schema默认public可用HINDSIGHT_API_DATABASE_SCHEMA覆盖HINDSIGHT_API_TENANT_EXTENSIONhindsight_api.extensions.builtin.tenant:ApiKeyTenantExtension HINDSIGHT_API_TENANT_API_KEYyour-secret-key从源码还可以看到两个补充细节未配置任何租户扩展时系统默认启用DefaultTenantExtension免鉴权的单租户模式使用HINDSIGHT_API_DATABASE_SCHEMA默认public而HINDSIGHT_API_TENANT_MCP_AUTH_DISABLEDtrue可以单独关闭 MCP 端点的鉴权用于向后兼容既有 MCP 服务。内置实现二SupabaseTenantExtension验证 Supabase JWT 并提供多租户记忆隔离。每个通过鉴权的用户拥有独立的 PostgreSQL schema{prefix}_{user_id}默认前缀user实现完全的数据分离。它采用JWKS 本地验签默认从/auth/v1/.well-known/jwks.json拉取公钥在本地验证 token每请求零网络调用仅对不支持 JWKS 的旧 HS256 项目才回退到每请求调用/auth/v1/user接口验证此时才需要 service key。该扩展作为独立包发布在 hindsight-extensions/supabase-tenant实现位于 hindsight_ext_supabase_tenant/extension.py内置了 JWKS 缓存 TTL600 秒、最小刷新间隔30 秒、请求超时10 秒与密钥轮换处理等生产级细节HINDSIGHT_API_TENANT_EXTENSIONhindsight_ext_supabase_tenant:SupabaseTenantExtension HINDSIGHT_API_TENANT_SUPABASE_URLhttps://your-project.supabase.co # 可选仅旧 HS256 项目或健康检查需要 HINDSIGHT_API_TENANT_SUPABASE_SERVICE_KEYyour-service-role-key对于其他每租户独立 schema的自定义方案例如基于自有 JWT 的鉴权则应自行实现TenantExtension见下文示例。仓库中 hindsight-extensions/static-keys-tenant 还提供了一个静态 Key 映射到租户 schema 的参考实现可对照学习。TenantExtension还支持两个进阶能力get_tenant_config()可在分层配置解析时注入租户级配置覆盖如{llm_model: gpt-4}键为 Python 字段名而非环境变量名get_allowed_config_fields()可控制租户/银行在 bank config API 上允许修改的字段子集返回None表示允许全部、空集合表示只读。HttpExtension在/ext/下挂载自定义端点HttpExtension用于添加自定义 HTTP 端点便于为特定业务域提供与记忆引擎集成的 API。基类定义在 extensions/http.py提供两个路由方法get_router(memory: MemoryEngine) - APIRouter—— 返回挂载在/ext/前缀下的 FastAPI 路由get_root_router(memory) - APIRouter | None—— 可选返回挂载在应用根路径的路由用于/.well-known/等必须位于固定路径的端点默认返回None。get_router中定义的路由最终会以/ext/hello、/ext/custom/{bank_id}/action的形式对外暴露。该方法接收MemoryEngine实例可在处理器内通过memory._get_pool()获取连接池执行数据库操作或直接调用memory.health_check()等核心方法。该类型无内置实现需要自行编写HINDSIGHT_API_HTTP_EXTENSIONmypackage.ext:MyHttpExtensionOperationValidatorExtension操作校验与监控钩子OperationValidatorExtension挂钩 retain / recall / reflect / consolidate 等操作典型用途包括限流与配额执行、权限检查与内容过滤、审计日志与用量追踪、自定义指标采集。其完整契约定义在 extensions/operation_validator.py无内置实现HINDSIGHT_API_OPERATION_VALIDATOR_EXTENSIONmypackage.validators:MyValidator该扩展的钩子执行顺序为validate_*操作前→ 操作执行 →on_*_complete操作后。validate_*钩子有三个返回值语义结果表达方式效果接受ValidationResult.accept()放行操作接受并富化ValidationResult.accept_with(contents..., tags..., tags_match..., tag_groups...)引擎使用返回的富化数据替代原始请求参数如注入 tags / tag_groups拒绝ValidationResult.reject(reason, status_code403)拒绝操作上游抛出OperationValidationError除文档示例中要求的validate_retain/validate_recall/validate_reflect三个抽象方法外源码还提供了大量可选钩子覆盖更细粒度的场景precheck在请求体反序列化之前执行的廉价预检查可基于Content-Length做成本预估而不读取 body、validate_consolidate、validate_mental_model_get/refresh、validate_bank_read/write、validate_create_bank、filter_bank_list查询后的列表过滤而非阻断、filter_mcp_tools按用户裁剪 MCP 工具可见集以及对应的on_*_complete后置钩子含on_file_convert_complete文件转换计费、mental model 的 token 用量统计等。后置钩子的上下文对象携带丰富的执行结果例如RetainResult中不仅包含unit_ids、success、error还包括llm_input_tokens/llm_output_tokens/llm_cached_input_tokens提示词缓存命中部分/llm_thoughts_tokens推理 token等计费关键数据以及processed_content_tokens去重后实际进入 LLM 提取的 token 数——这些都是实现按用量计费扩展的直接依据。MCPExtension为 MCP 服务器注册自定义工具MCPExtension允许外部包在不改核心代码的前提下向 Hindsight MCP 服务器注册额外的 Model Context Protocol 工具。基类定义在 extensions/mcp.py唯一抽象方法是register_tools(mcp: FastMCP, memory: MemoryEngine)其中mcp是 FastMCP 服务器实例可直接使用mcp.tool()装饰器注册工具HINDSIGHT_API_MCP_EXTENSIONmypackage.mcp:MyMCPExtension该类型无内置实现但扩展包内可访问MemoryEngine来执行记忆操作如自定义搜索、批量 retain 等。编写自定义扩展四个完整示例扩展基础加载与配置约定所有扩展统一遵循HINDSIGHT_API_TYPE_EXTENSIONmypackage.module:MyExtensionClass配置通过前缀环境变量传入键会被剥离前缀并转为小写HINDSIGHT_API_TYPE_SOME_CONFIGvalue # 扩展收到: {some_config: value}生命周期钩子on_startup()/on_shutdown()对所有扩展可用ExtensionContext提供run_migration(schema)与get_memory_engine()两个系统 API。示例一基于 JWT 的自定义 TenantExtension以下示例校验 JWT 中的tenant_id声明并映射到tenant_{tenant_id}schemaimport jwt from hindsight_api.extensions import TenantExtension, TenantContext, AuthenticationError class JwtTenantExtension(TenantExtension): def __init__(self, config: dict[str, str]): super().__init__(config) self.jwt_secret config.get(jwt_secret) if not self.jwt_secret: raise ValueError(HINDSIGHT_API_TENANT_JWT_SECRET is required) async def authenticate(self, context: RequestContext) - TenantContext: token context.api_key if not token: # 可选的 headers 字典会透传到 HTTP/MCP 错误响应 raise AuthenticationError(Bearer token required) try: payload jwt.decode(token, self.jwt_secret, algorithms[HS256]) tenant_id payload.get(tenant_id) if not tenant_id: raise AuthenticationError(Missing tenant_id in token) return TenantContext(schema_nameftenant_{tenant_id}) except jwt.InvalidTokenError as e: raise AuthenticationError(str(e))AuthenticationError接受可选的headers参数该参数会同时透传到 HTTP 与 MCP 错误响应适合返回WWW-Authenticate等自定义头raise AuthenticationError( Authorization required, headers{WWW-Authenticate: Bearer realmexample}, )补充说明默认情况下只有Authorization头以RequestContext.api_key形式会进入authenticate()。若部署环境需要基于其他请求头鉴权如代理透传的身份断言可设置HINDSIGHT_API_EXTENSION_PASSTHROUGH_HEADERSx-user-assertion将其显式纳入RequestContext.extra_headersHTTP 与 MCP 传输均生效未设置时扩展不接收任何请求头数据。示例二自定义 HttpExtensionfrom fastapi import APIRouter from hindsight_api.extensions import HttpExtension class MyHttpExtension(HttpExtension): def get_router(self, memory: MemoryEngine) - APIRouter: router APIRouter() router.get(/hello) async def hello(): return {message: Hello from extension!} router.post(/custom/{bank_id}/action) async def custom_action(bank_id: str): # 访问 memory engine 执行数据库操作 pool await memory._get_pool() # ... custom logic return {status: ok} return router def get_root_router(self, memory: MemoryEngine) - APIRouter | None: 可选在应用根路径挂载路由而非 /ext/ 下。 router APIRouter() router.get(/.well-known/my-metadata) async def metadata(): return {version: 1.0} return routerget_router中的路由可通过/ext/hello、/ext/custom/{bank_id}/action访问get_root_router的路由挂在应用根路径如/.well-known/my-metadata。示例三自定义 OperationValidatorExtension 与延迟操作from hindsight_api.extensions import ( OperationValidatorExtension, ValidationResult, RetainContext, RecallContext, ReflectContext, RetainResult, ) class MyValidator(OperationValidatorExtension): # 操作前校验必须实现 async def validate_retain(self, ctx: RetainContext) - ValidationResult: # 实现你的校验逻辑 return ValidationResult.accept() # 或者拒绝: return ValidationResult.reject(Reason) async def validate_recall(self, ctx: RecallContext) - ValidationResult: return ValidationResult.accept() async def validate_reflect(self, ctx: ReflectContext) - ValidationResult: return ValidationResult.accept() # 操作后钩子可选 async def on_retain_complete(self, result: RetainResult) - None: # 记录用量、更新指标、发送通知等 pass延迟操作DeferOperation除了accept和rejectvalidate_*钩子还可以通过抛出DeferOperation请求 worker重新排队该操作到未来某个时间点执行。这适用于背压场景上游被限流、配额窗口未开启、依赖服务正在预热——与重试不同延迟不会递增retry_count也不会写入error_message。worker 会将next_retry_at设置为你指定的exec_date在此之前任务对 claim 查询不可见from datetime import datetime, timedelta, timezone from hindsight_api.extensions import ( DeferOperation, OperationValidatorExtension, RetainContext, ValidationResult, ) class QuotaAwareValidator(OperationValidatorExtension): async def validate_retain(self, ctx: RetainContext) - ValidationResult: if not await self._quota_available(ctx.bank_id): raise DeferOperation( exec_datedatetime.now(timezone.utc) timedelta(minutes5), reasonbank quota window exhausted, ) return ValidationResult.accept()DeferOperation定义在 hindsight-api-slim/hindsight_api/worker/exceptions.py由 worker 轮询器worker/poller.py捕获处理。重要限制它是worker 专用的——不要从同步 HTTP 请求路径上的validate_recall或validate_reflect中抛出它因为同步路径没有可延迟的队列该异常会以 500 错误浮出。示例四自定义 MCPExtensionfrom mcp.server.fastmcp import FastMCP from hindsight_api.extensions import MCPExtension from hindsight_api.engine import MemoryEngine class MyMCPExtension(MCPExtension): async def register_tools(self, mcp: FastMCP, memory: MemoryEngine) - None: mcp.tool() async def custom_search(query: str) - str: Custom MCP tool for specialized search. # 访问 memory engine 执行操作 pool await memory._get_pool() # ... custom logic return fResults for: {query}部署自定义扩展Docker 部署方式一将扩展包作为 volume 挂载并设置环境变量# docker-compose.yml services: hindsight-api: image: vectorize/hindsight-api:latest volumes: - ./my_extensions:/app/my_extensions environment: - HINDSIGHT_API_TENANT_EXTENSIONmy_extensions.auth:JwtTenantExtension - HINDSIGHT_API_TENANT_JWT_SECRET${JWT_SECRET} - PYTHONPATH/app方式二构建包含扩展的自定义镜像FROM vectorize/hindsight-api:latest COPY my_extensions /app/my_extensions ENV PYTHONPATH/app仓库中的 hindsight-extensions/supabase-tenant/Dockerfile 正是基于 Hindsight 镜像叠加扩展包的官方参考可对照使用。裸机部署将扩展包安装到与 Hindsight 相同的 Python 环境中# 安装 Hindsight pip install hindsight-api # 安装你的扩展包 pip install ./my-extensions # 或 pip install my-extensions-package # 配置 export HINDSIGHT_API_TENANT_EXTENSIONmy_extensions.auth:JwtTenantExtension export HINDSIGHT_API_TENANT_JWT_SECRETyour-secret # 运行 hindsight-api将扩展贡献回社区如果你构建了解决常见问题的扩展欢迎将其贡献到 Hindsight 的hindsight_api.extensions.builtin包中。官方尤其欢迎以下类别的扩展认证提供方OAuth、SAML、API 网关限流与配额管理审计日志集成指标导出器Datadog、New Relic 等面向特定平台的自定义 HTTP 端点。投稿前建议先阅读仓库根目录的 CONTRIBUTING.md 与 AGENTS.md了解代码规范与测试要求扩展相关测试可参考 hindsight-api-slim/tests/test_extensions.py 及 hindsight-extensions/supabase-tenant/tests 中的既有用例。小结Hindsight 的扩展体系可以用一句话概括环境变量声明插件前缀环境变量传递配置抽象基类约束契约ExtensionContext 提供受控系统访问。TenantExtension解决谁在用、用哪个 schemaHttpExtension解决如何暴露自定义 APIOperationValidatorExtension解决操作能否执行、执行后做什么MCPExtension解决如何扩展 Agent 工具面。结合DeferOperation的背压语义与accept_with的请求富化能力这套机制足以支撑生产级的租户隔离、配额计费、审计与平台化扩展需求而无需触碰 Hindsight 的任何核心代码。【免费下载链接】hindsightHindsight: Agent Memory That Learns项目地址: https://gitcode.com/GitHub_Trending/hindsight2/hindsight创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Hindsight Extensions 扩展机制实战指南：从多租户隔离到自定义 MCP 工具
Hindsight Extensions 扩展机制实战指南从多租户隔离到自定义 MCP 工具【免费下载链接】hindsightHindsight: Agent Memory That Learns项目地址: https://gitcode.com/GitHub_Trending/hindsight2/hindsightHindsight 通过一套基于环境变量驱动的扩展Extensions体系允许在不修改核心代码的前提下自定义服务端行为覆盖多租户隔离、自定义鉴权、额外 HTTP 端点与操作钩子四类场景。本文以 hindsight-docs/versioned_docs/version-0.6/developer/extensions.md 为主线结合hindsight-api-slim中扩展基类的源码实现完整讲解四种内置扩展类型、自定义扩展的编写范式、延迟操作DeferOperation语义与 Docker/裸机部署方式让读者可以独立实现并上线一个生产可用的 Hindsight 扩展。扩展机制总览环境变量即插件入口Hindsight 扩展的本质是通过环境变量指定的 Python 类。其加载逻辑位于 extensions/loader.py 的load_extension函数系统读取HINDSIGHT_API_TYPE_EXTENSION环境变量格式为module.path:ClassName动态导入模块、校验类继承关系、收集配置并实例化。扩展配置的传递遵循前缀剥离规则。load_extension会扫描所有以HINDSIGHT_API_TYPE_开头的环境变量排除EXTENSION变量本身剥离前缀并转为小写后作为config字典传给扩展构造函数HINDSIGHT_API_OPERATION_VALIDATOR_EXTENSIONmypackage.validators:MyValidator HINDSIGHT_API_OPERATION_VALIDATOR_MAX_REQUESTS100 # 扩展收到 config {max_requests: 100}所有扩展类都继承自 extensions/base.py 中的抽象基类Extension并共享两个生命周期钩子on_startup()—— 应用启动时调用适合连接外部服务、预热缓存on_shutdown()—— 应用关闭时调用适合释放连接、清理资源。扩展还可以通过set_context()获得一个ExtensionContext见 extensions/context.py它提供受控的系统访问 API而非直接暴露内部组件run_migration(schema)—— 为指定 schema 运行数据库迁移自动创建 schema、执行全部待执行迁移并通过 advisory lock 协调分布式 workerget_memory_engine()—— 获取MemoryEngineInterface用于执行 retain、recall、reflect 等记忆操作。DefaultExtensionContext是内置实现它在迁移完成后还会调用租户扩展声明的provision_bank_tables使扩展自有表与核心 schema 保持同一生命周期。四大扩展类型与内置实现TenantExtension多租户隔离与鉴权TenantExtension负责处理多租户和 API Key 鉴权。它校验入站请求并决定数据库操作使用哪个 PostgreSQL schema从而实现数据库层面的租户隔离。其抽象基类定义在 extensions/tenant.py核心方法包括authenticate(context: RequestContext) - TenantContext—— 校验请求并返回TenantContext(schema_name...)后续所有查询都会使用schema_name.memory_units这样的全限定表名list_tenants() - list[Tenant]—— 供 worker 发现所有租户 schema进行任务轮询authenticate_mcp(context)—— MCP 端点的鉴权入口默认委托给authenticate()可覆写以实现 HTTP 与 MCP 的差异化鉴权。鉴权失败时抛出AuthenticationError(reason, headersNone)其headers字典会被透传到 HTTP 与 MCP 的错误响应中非常适合返回WWW-Authenticate等自定义头。内置实现一ApiKeyTenantExtension源码见 extensions/builtin/tenant.py。它校验请求的 API Key 是否与HINDSIGHT_API_TENANT_API_KEY匹配并为所有通过鉴权的请求使用同一个 schema默认public可用HINDSIGHT_API_DATABASE_SCHEMA覆盖HINDSIGHT_API_TENANT_EXTENSIONhindsight_api.extensions.builtin.tenant:ApiKeyTenantExtension HINDSIGHT_API_TENANT_API_KEYyour-secret-key从源码还可以看到两个补充细节未配置任何租户扩展时系统默认启用DefaultTenantExtension免鉴权的单租户模式使用HINDSIGHT_API_DATABASE_SCHEMA默认public而HINDSIGHT_API_TENANT_MCP_AUTH_DISABLEDtrue可以单独关闭 MCP 端点的鉴权用于向后兼容既有 MCP 服务。内置实现二SupabaseTenantExtension验证 Supabase JWT 并提供多租户记忆隔离。每个通过鉴权的用户拥有独立的 PostgreSQL schema{prefix}_{user_id}默认前缀user实现完全的数据分离。它采用JWKS 本地验签默认从/auth/v1/.well-known/jwks.json拉取公钥在本地验证 token每请求零网络调用仅对不支持 JWKS 的旧 HS256 项目才回退到每请求调用/auth/v1/user接口验证此时才需要 service key。该扩展作为独立包发布在 hindsight-extensions/supabase-tenant实现位于 hindsight_ext_supabase_tenant/extension.py内置了 JWKS 缓存 TTL600 秒、最小刷新间隔30 秒、请求超时10 秒与密钥轮换处理等生产级细节HINDSIGHT_API_TENANT_EXTENSIONhindsight_ext_supabase_tenant:SupabaseTenantExtension HINDSIGHT_API_TENANT_SUPABASE_URLhttps://your-project.supabase.co # 可选仅旧 HS256 项目或健康检查需要 HINDSIGHT_API_TENANT_SUPABASE_SERVICE_KEYyour-service-role-key对于其他每租户独立 schema的自定义方案例如基于自有 JWT 的鉴权则应自行实现TenantExtension见下文示例。仓库中 hindsight-extensions/static-keys-tenant 还提供了一个静态 Key 映射到租户 schema 的参考实现可对照学习。TenantExtension还支持两个进阶能力get_tenant_config()可在分层配置解析时注入租户级配置覆盖如{llm_model: gpt-4}键为 Python 字段名而非环境变量名get_allowed_config_fields()可控制租户/银行在 bank config API 上允许修改的字段子集返回None表示允许全部、空集合表示只读。HttpExtension在/ext/下挂载自定义端点HttpExtension用于添加自定义 HTTP 端点便于为特定业务域提供与记忆引擎集成的 API。基类定义在 extensions/http.py提供两个路由方法get_router(memory: MemoryEngine) - APIRouter—— 返回挂载在/ext/前缀下的 FastAPI 路由get_root_router(memory) - APIRouter | None—— 可选返回挂载在应用根路径的路由用于/.well-known/等必须位于固定路径的端点默认返回None。get_router中定义的路由最终会以/ext/hello、/ext/custom/{bank_id}/action的形式对外暴露。该方法接收MemoryEngine实例可在处理器内通过memory._get_pool()获取连接池执行数据库操作或直接调用memory.health_check()等核心方法。该类型无内置实现需要自行编写HINDSIGHT_API_HTTP_EXTENSIONmypackage.ext:MyHttpExtensionOperationValidatorExtension操作校验与监控钩子OperationValidatorExtension挂钩 retain / recall / reflect / consolidate 等操作典型用途包括限流与配额执行、权限检查与内容过滤、审计日志与用量追踪、自定义指标采集。其完整契约定义在 extensions/operation_validator.py无内置实现HINDSIGHT_API_OPERATION_VALIDATOR_EXTENSIONmypackage.validators:MyValidator该扩展的钩子执行顺序为validate_*操作前→ 操作执行 →on_*_complete操作后。validate_*钩子有三个返回值语义结果表达方式效果接受ValidationResult.accept()放行操作接受并富化ValidationResult.accept_with(contents..., tags..., tags_match..., tag_groups...)引擎使用返回的富化数据替代原始请求参数如注入 tags / tag_groups拒绝ValidationResult.reject(reason, status_code403)拒绝操作上游抛出OperationValidationError除文档示例中要求的validate_retain/validate_recall/validate_reflect三个抽象方法外源码还提供了大量可选钩子覆盖更细粒度的场景precheck在请求体反序列化之前执行的廉价预检查可基于Content-Length做成本预估而不读取 body、validate_consolidate、validate_mental_model_get/refresh、validate_bank_read/write、validate_create_bank、filter_bank_list查询后的列表过滤而非阻断、filter_mcp_tools按用户裁剪 MCP 工具可见集以及对应的on_*_complete后置钩子含on_file_convert_complete文件转换计费、mental model 的 token 用量统计等。后置钩子的上下文对象携带丰富的执行结果例如RetainResult中不仅包含unit_ids、success、error还包括llm_input_tokens/llm_output_tokens/llm_cached_input_tokens提示词缓存命中部分/llm_thoughts_tokens推理 token等计费关键数据以及processed_content_tokens去重后实际进入 LLM 提取的 token 数——这些都是实现按用量计费扩展的直接依据。MCPExtension为 MCP 服务器注册自定义工具MCPExtension允许外部包在不改核心代码的前提下向 Hindsight MCP 服务器注册额外的 Model Context Protocol 工具。基类定义在 extensions/mcp.py唯一抽象方法是register_tools(mcp: FastMCP, memory: MemoryEngine)其中mcp是 FastMCP 服务器实例可直接使用mcp.tool()装饰器注册工具HINDSIGHT_API_MCP_EXTENSIONmypackage.mcp:MyMCPExtension该类型无内置实现但扩展包内可访问MemoryEngine来执行记忆操作如自定义搜索、批量 retain 等。编写自定义扩展四个完整示例扩展基础加载与配置约定所有扩展统一遵循HINDSIGHT_API_TYPE_EXTENSIONmypackage.module:MyExtensionClass配置通过前缀环境变量传入键会被剥离前缀并转为小写HINDSIGHT_API_TYPE_SOME_CONFIGvalue # 扩展收到: {some_config: value}生命周期钩子on_startup()/on_shutdown()对所有扩展可用ExtensionContext提供run_migration(schema)与get_memory_engine()两个系统 API。示例一基于 JWT 的自定义 TenantExtension以下示例校验 JWT 中的tenant_id声明并映射到tenant_{tenant_id}schemaimport jwt from hindsight_api.extensions import TenantExtension, TenantContext, AuthenticationError class JwtTenantExtension(TenantExtension): def __init__(self, config: dict[str, str]): super().__init__(config) self.jwt_secret config.get(jwt_secret) if not self.jwt_secret: raise ValueError(HINDSIGHT_API_TENANT_JWT_SECRET is required) async def authenticate(self, context: RequestContext) - TenantContext: token context.api_key if not token: # 可选的 headers 字典会透传到 HTTP/MCP 错误响应 raise AuthenticationError(Bearer token required) try: payload jwt.decode(token, self.jwt_secret, algorithms[HS256]) tenant_id payload.get(tenant_id) if not tenant_id: raise AuthenticationError(Missing tenant_id in token) return TenantContext(schema_nameftenant_{tenant_id}) except jwt.InvalidTokenError as e: raise AuthenticationError(str(e))AuthenticationError接受可选的headers参数该参数会同时透传到 HTTP 与 MCP 错误响应适合返回WWW-Authenticate等自定义头raise AuthenticationError( Authorization required, headers{WWW-Authenticate: Bearer realmexample}, )补充说明默认情况下只有Authorization头以RequestContext.api_key形式会进入authenticate()。若部署环境需要基于其他请求头鉴权如代理透传的身份断言可设置HINDSIGHT_API_EXTENSION_PASSTHROUGH_HEADERSx-user-assertion将其显式纳入RequestContext.extra_headersHTTP 与 MCP 传输均生效未设置时扩展不接收任何请求头数据。示例二自定义 HttpExtensionfrom fastapi import APIRouter from hindsight_api.extensions import HttpExtension class MyHttpExtension(HttpExtension): def get_router(self, memory: MemoryEngine) - APIRouter: router APIRouter() router.get(/hello) async def hello(): return {message: Hello from extension!} router.post(/custom/{bank_id}/action) async def custom_action(bank_id: str): # 访问 memory engine 执行数据库操作 pool await memory._get_pool() # ... custom logic return {status: ok} return router def get_root_router(self, memory: MemoryEngine) - APIRouter | None: 可选在应用根路径挂载路由而非 /ext/ 下。 router APIRouter() router.get(/.well-known/my-metadata) async def metadata(): return {version: 1.0} return routerget_router中的路由可通过/ext/hello、/ext/custom/{bank_id}/action访问get_root_router的路由挂在应用根路径如/.well-known/my-metadata。示例三自定义 OperationValidatorExtension 与延迟操作from hindsight_api.extensions import ( OperationValidatorExtension, ValidationResult, RetainContext, RecallContext, ReflectContext, RetainResult, ) class MyValidator(OperationValidatorExtension): # 操作前校验必须实现 async def validate_retain(self, ctx: RetainContext) - ValidationResult: # 实现你的校验逻辑 return ValidationResult.accept() # 或者拒绝: return ValidationResult.reject(Reason) async def validate_recall(self, ctx: RecallContext) - ValidationResult: return ValidationResult.accept() async def validate_reflect(self, ctx: ReflectContext) - ValidationResult: return ValidationResult.accept() # 操作后钩子可选 async def on_retain_complete(self, result: RetainResult) - None: # 记录用量、更新指标、发送通知等 pass延迟操作DeferOperation除了accept和rejectvalidate_*钩子还可以通过抛出DeferOperation请求 worker重新排队该操作到未来某个时间点执行。这适用于背压场景上游被限流、配额窗口未开启、依赖服务正在预热——与重试不同延迟不会递增retry_count也不会写入error_message。worker 会将next_retry_at设置为你指定的exec_date在此之前任务对 claim 查询不可见from datetime import datetime, timedelta, timezone from hindsight_api.extensions import ( DeferOperation, OperationValidatorExtension, RetainContext, ValidationResult, ) class QuotaAwareValidator(OperationValidatorExtension): async def validate_retain(self, ctx: RetainContext) - ValidationResult: if not await self._quota_available(ctx.bank_id): raise DeferOperation( exec_datedatetime.now(timezone.utc) timedelta(minutes5), reasonbank quota window exhausted, ) return ValidationResult.accept()DeferOperation定义在 hindsight-api-slim/hindsight_api/worker/exceptions.py由 worker 轮询器worker/poller.py捕获处理。重要限制它是worker 专用的——不要从同步 HTTP 请求路径上的validate_recall或validate_reflect中抛出它因为同步路径没有可延迟的队列该异常会以 500 错误浮出。示例四自定义 MCPExtensionfrom mcp.server.fastmcp import FastMCP from hindsight_api.extensions import MCPExtension from hindsight_api.engine import MemoryEngine class MyMCPExtension(MCPExtension): async def register_tools(self, mcp: FastMCP, memory: MemoryEngine) - None: mcp.tool() async def custom_search(query: str) - str: Custom MCP tool for specialized search. # 访问 memory engine 执行操作 pool await memory._get_pool() # ... custom logic return fResults for: {query}部署自定义扩展Docker 部署方式一将扩展包作为 volume 挂载并设置环境变量# docker-compose.yml services: hindsight-api: image: vectorize/hindsight-api:latest volumes: - ./my_extensions:/app/my_extensions environment: - HINDSIGHT_API_TENANT_EXTENSIONmy_extensions.auth:JwtTenantExtension - HINDSIGHT_API_TENANT_JWT_SECRET${JWT_SECRET} - PYTHONPATH/app方式二构建包含扩展的自定义镜像FROM vectorize/hindsight-api:latest COPY my_extensions /app/my_extensions ENV PYTHONPATH/app仓库中的 hindsight-extensions/supabase-tenant/Dockerfile 正是基于 Hindsight 镜像叠加扩展包的官方参考可对照使用。裸机部署将扩展包安装到与 Hindsight 相同的 Python 环境中# 安装 Hindsight pip install hindsight-api # 安装你的扩展包 pip install ./my-extensions # 或 pip install my-extensions-package # 配置 export HINDSIGHT_API_TENANT_EXTENSIONmy_extensions.auth:JwtTenantExtension export HINDSIGHT_API_TENANT_JWT_SECRETyour-secret # 运行 hindsight-api将扩展贡献回社区如果你构建了解决常见问题的扩展欢迎将其贡献到 Hindsight 的hindsight_api.extensions.builtin包中。官方尤其欢迎以下类别的扩展认证提供方OAuth、SAML、API 网关限流与配额管理审计日志集成指标导出器Datadog、New Relic 等面向特定平台的自定义 HTTP 端点。投稿前建议先阅读仓库根目录的 CONTRIBUTING.md 与 AGENTS.md了解代码规范与测试要求扩展相关测试可参考 hindsight-api-slim/tests/test_extensions.py 及 hindsight-extensions/supabase-tenant/tests 中的既有用例。小结Hindsight 的扩展体系可以用一句话概括环境变量声明插件前缀环境变量传递配置抽象基类约束契约ExtensionContext 提供受控系统访问。TenantExtension解决谁在用、用哪个 schemaHttpExtension解决如何暴露自定义 APIOperationValidatorExtension解决操作能否执行、执行后做什么MCPExtension解决如何扩展 Agent 工具面。结合DeferOperation的背压语义与accept_with的请求富化能力这套机制足以支撑生产级的租户隔离、配额计费、审计与平台化扩展需求而无需触碰 Hindsight 的任何核心代码。【免费下载链接】hindsightHindsight: Agent Memory That Learns项目地址: https://gitcode.com/GitHub_Trending/hindsight2/hindsight创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
