拦截与自动捕获:register 如何劫持 LLM 调用

30 秒导读: 你只写一行 memori.llm.register(client),之后照常调用 OpenAI / Anthropic / Gemini。Memori 在背后把你的 client.chat.completions.create 悄悄换成了自己的包装方法——每一次调用它都会:请求发出前把历史与召回的记忆注进 prompt,调用真实的官方方法,拿到回复后再攒出一份"待处理载荷"交给记忆管线。整个过程你的业务代码一个字都不用改。这就是所谓的零改代码捕获。

本章只回答一个问题:Memori 怎么在不改你代码的前提下,把一次 LLM 调用变成一份待处理载荷?

• 载荷之后如何被炼成记忆 → 见 02-augmentation.md。
• 注进 prompt 的那些"召回记忆"是从哪检索来的 → 见 03-recall.md。本章只讲注入发生的时机和位置,不讲检索逻辑。
• 顶层全景 → 见 index.md。

---

1. 这是什么(零基础也能懂)

一句话定义

Memori 的捕获层是一套 monkey patch(猴子补丁,运行时把一个对象的方法替换成你自己的实现) 机制:它把官方 SDK 客户端上"发起对话"的那个方法,替换成一个功能相同、但前后多做了两件事的包装方法。

它要解决的问题

你想给一个已经写好的 AI 应用加"长期记忆",但不想重写调用代码。传统做法是自己在每次调用前后插桩:调用前查数据库拼 prompt、调用后把对话存起来。这既繁琐又容易漏。

Memori 的思路是:你只注册一次,之后所有调用自动被拦截。

用起来什么样

from openai import OpenAI
from memori import Memori

client = OpenAI()
memori = Memori(...)          # 建立记忆实例
memori.llm.register(client)   # ← 关键的一行:劫持这个 client

# 之后照常用,一字不改。但每次 create 都被 Memori 接管了
resp = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "我住在东京"}],
)

注册那一行返回后,client.chat.completions.create 已经不是 OpenAI 原来的方法了——它指向了 Memori 的包装器。你调用它的姿势不变,但行为多了"记忆注入"和"对话捕获"。

一句话直觉

把它想成给水管中间接了个透明的三通阀:水(你的请求/回复)照样流过,但阀门每次都能取一瓢(捕获对话)、也能掺一点(注入记忆),而上下游的水管(你的代码、官方 SDK)都感觉不到它的存在。

---

2. 顶层全景(捕获层怎么转)

捕获层内部分三步走,对应三个模块:

阶段	干什么	核心文件
① 注册	认出你传进来的是哪家 client,把它的方法替换成包装器	memori/llm/_registry.py、memori/llm/clients/direct.py
② 拦截	每次调用时,请求前注入、调用真方法、按返回类型分流	memori/llm/invoke/invoke.py
③ 捕获	响应回来后组装成标准载荷,交给记忆管线	memori/llm/pipelines/post_invoke.py

主线走一遍(从左到右,一次 LLM 调用的生命周期):

       ┌───────────── 注册期(register,只发生一次)─────────────┐
你的代码 ─register(client)─▶ 匹配器认出 provider ─▶ 把 create 换成 Invoke.invoke
       └───────────────────────────────────────────────────────┘

       ┌───────────── 调用期(每次 create 都发生)──────────────┐
                        ① 注入召回记忆(recall)
你调 create ─▶ Invoke.invoke ─▶ ② 注入历史对话(conversation)
                        ③ 调用真实的官方 create  ← self._method
                        ④ 看返回类型分流
                             ├─ 同步对象 ─▶ 立刻捕获
                             └─ 流式迭代器 ─▶ 边迭代边攒,结束时捕获
                        ⑤ handle_post_response ─▶ 组装载荷 ─▶ 交给记忆管线(→02)
       └───────────────────────────────────────────────────────┘

怎么读这张图: 上半区是"注册"——一次性动作,把方法换掉;下半区是"调用"——之后每次 create 都会跑一遍 ①→⑤。①②是本章的"注入时机",⑤是本章的终点"载荷"。

---

3. 注册:register 如何认出 client 并替换方法

3.1 它要解决的小问题

register(client) 收到的可能是 OpenAI、Anthropic、Google、xAI、LiteLLM 里的任意一个,每家 SDK 的方法路径都不一样(OpenAI 是 client.chat.completions.create,Anthropic 是 client.messages.create……)。第一步得先认出这是谁,再决定去 patch 哪个方法。

3.2 思路:matcher 注册表 + 按注册顺序匹配

Memori 用一张注册表解决"认人"的问题。每个 provider 的包装类头上挂一个装饰器,把"一个判断函数(matcher)"和"这个类"配成一对,存进 Registry._clients 字典。

真源码 memori/llm/_registry.py:33 的 Registry.register_client:

@classmethod
def register_client(cls, matcher):        # matcher: 一个 (client) -> bool 的判断函数
    def decorator(client_class):
        cls._clients[matcher] = client_class   # 把"判断函数 → 包装类"存进注册表
        return client_class
    return decorator

这段的意思:它是个装饰器工厂,把 matcher(判断某个 client 是不是本家)和对应的包装类登记进 _clients。

包装类就这样自我登记。看 memori/llm/clients/direct.py:24 和 :155:

@Registry.register_client(client_is_anthropic)   # 用 client_is_anthropic 判断
class Anthropic(BaseClient): ...

@Registry.register_client(client_is_openai)      # 用 client_is_openai 判断
class OpenAi(BaseClient): ...

matcher 本身很朴素——就是看 client 的模块名前缀。memori/llm/_utils.py:35 的 client_is_anthropic:

def client_is_anthropic(client) -> bool:
    return _client_module(client).startswith("anthropic")   # 看 __module__ 是不是以 anthropic 开头

3.3 匹配:遍历注册表,第一个命中就用

注册进来的 client 交给 Registry.client() 解析。memori/llm/_registry.py:51:

def client(self, client_obj, config):
    for matcher, client_class in self._clients.items():   # 按注册顺序遍历
        if matcher(client_obj):                            # 第一个返回 True 的
            return client_class(config)                    # 就实例化它的包装类
    ...
    raise UnsupportedLLMProviderError(provider)            # 都不匹配就报错

关键细节: 匹配是按注册顺序的线性扫描,谁先命中用谁(注释在 _registry.py:24 明确写了 "selected by matcher registration order")。所以 matcher 之间要互斥、别重叠,否则顺序会决定归属。

register_llm(_registry.py:77)是对外总入口:它先分流(直连 client 走 Registry().client(...);Agno / LangChain 这类框架走命名参数 openai_chat= / chatopenai=,委托给 memori.agno.register / memori.langchain.register),然后调 client_handler.register(client) 完成真正的替换(_registry.py:176-178)。

支持的 provider / adapter 一览:

家族	client matcher(_utils.py)	直连包装类(clients/direct.py)	响应 adapter(adapters/*/_adapter.py)
OpenAI	client_is_openai	OpenAi	openai/_adapter.py
Anthropic	client_is_anthropic	Anthropic	anthropic/_adapter.py
Google	client_is_google	Google	google/_adapter.py
xAI	client_is_xai	XAi	xai/_adapter.py
Bedrock	(经 LangChain 命名参数)	—	bedrock/_adapter.py
LiteLLM	client_is_litellm	LiteLLM	复用 openai adapter
PydanticAi	client_is_pydantic_ai	PydanticAi	复用 openai 系

adapter 是"怎么把这家的响应结构解析成统一 messages"的解析器,和 client 包装类一一对应但各管一段。adapter 也用同一套注册表(register_adapter,_registry.py:41;在 memori/llm/__init__.py:13 于 import 时统一注册)。本章只讲到它"存在且按 provider 选中",解析细节属于载荷成型,不展开。

3.4 替换动作:备份原方法,装上包装器

真正的 "劫持" 在各包装类的 register() 里。以 Anthropic 为例(direct.py:26):它先把原方法存一份备份(client._messages_create = client.messages.create),再调 _wrap_method 把 create 换成包装器。备份很关键——包装器最终还得调用它。

替换逻辑集中在 memori/llm/_base.py:35 的 BaseClient._wrap_method:

original = getattr(backup_obj, backup_attr)      # 取回刚备份的原方法
is_async = inspect.iscoroutinefunction(original) or \
           type(obj).__name__.startswith("Async")   # 判断同步还是异步
wrapper_class = ... Invoke / InvokeAsync / InvokeStream ...   # 按同步/异步/流式选包装类
setattr(obj, method_name,                        # 把 create 指向...
        wrapper_class(self.config, original)     # ...包装器(内部持有 original)
        .set_client(provider, llm_provider, version)
        .invoke)                                 # 暴露出去的是 .invoke 方法

这段重点看两处:它先自动判断同步/异步再选包装类;替换后 create 指向的是 wrapper.invoke,而原方法被包装器攥在手里(_method = original,见 _base.py:88),稍后调用期才会用到。

一个防重入的小细节:每个 register() 都用 hasattr(client, "_memori_installed") 做幂等标记(如 direct.py:30、:60),重复注册同一个 client 不会叠加多层包装。

---

4. 拦截:Invoke.invoke 的主循环

这是捕获层的心脏。替换后你调用的 create,实际执行的是 memori/llm/invoke/invoke.py:25 的 Invoke.invoke。它的骨架只有三段:请求前注入 → 调真方法 → 按返回类型分流。

4.1 请求前:先注召回,再注历史

invoke.py:28:

kwargs = inject_conversation_messages(
    self,
    inject_recalled_facts(self, self.configure_for_streaming_usage(kwargs)),
)

读法是从里往外:先 inject_recalled_facts(把召回的记忆片段拼进 system prompt),外面再套 inject_conversation_messages(把本会话的历史消息插到 messages 前面)。两者都只是在改 kwargs——也就是你即将发给模型的那份请求参数。

• inject_recalled_facts(pipelines/recall_injection.py:107):把召回结果包成 <memori_context>…</memori_context> 文本,按 provider 塞到正确位置——Anthropic/Bedrock 拼进 system 字段,Google 走 system_instruction,OpenAI Responses API 拼进 instructions,普通 Chat 则插一条 role: system 消息(recall_injection.py:201-224)。注意:本章只讲"注入发生在这里、位置是 system/instructions",召回结果怎么检索出来的属于 03-recall.md。
• inject_conversation_messages(pipelines/conversation_injection.py:156):从存储读回本会话历史,_inject_messages_by_provider 按 provider 的消息格式拼到请求最前面(conversation_injection.py:91)。它还记下"注入了几条"到 invoke._injected_message_count,好让捕获时把注入的历史排除掉、不重复入库。

一句话:注入 = 只动 kwargs,不动你的代码;时机 = 真调用之前;位置 = system/instructions/messages 头部。

4.2 调真方法

invoke.py:38:

raw_response = self._method(**kwargs)   # self._method 就是注册期备份的官方原方法

这里终于用上了替换时攥在手里的 original。对上下游而言,这一步和你直接调官方 SDK 完全等价——只是参数已被①②悄悄增强过。

4.3 按返回类型分流

模型的返回可能是一次性对象,也可能是流式。Invoke.invoke 用返回值的类型来分三条路(invoke.py:40-65):

raw_response 是什么?
├─ Iterator / generator ──────▶ 包成 MemoriIterator(边迭代边攒,结束时捕获)
├─ Bedrock 且 body 是 EventStream ─▶ 包成 MemoriIterable
│   └─ Bedrock 且 body 非流 ──────▶ 包成 MemoriStreamingBody(read 时捕获)
└─ 其它(同步整块响应)──────────▶ handle_post_response(立刻捕获)

返回类型	判断依据	包装/捕获类	何时触发捕获
同步整块	落到 else 分支	直接 handle_post_response	函数返回前立刻
通用流式	isinstance(Iterator) 或 isgenerator	invoke/iterator.py 的 Iterator	迭代到 StopIteration 时
Bedrock EventStream	client_is_bedrock 且 body 是 EventStream	invoke/iterable.py 的 Iterable	__iter__ 的 finally 块
Bedrock 非流 body	client_is_bedrock 且 body 非流	invoke/streaming.py 的 StreamingBody	调用方 .read() 时

为什么流式要单独处理? 流式响应是一段一段吐出来的,直到最后一块才算完整。所以 Memori 不能在调用点立刻捕获——它把真迭代器包一层,原样把每个 chunk 透传给你(你的循环照常跑),同时在内部用 process_chunk 把碎片攒成完整响应(_base.py:170 的 BaseIterator.process_chunk);等迭代终止(StopIteration / finally),再一次性组装载荷。看 invoke/iterator.py:61 的 Iterator.__next__:

def __next__(self):
    try:
        chunk = next(self.source_iterator)   # 拿真迭代器的下一块
        self.set_raw_response()
        self.process_chunk(chunk)            # 攒进 self.raw_response
        return chunk                         # 原样透传给调用方
    except StopIteration:
        MemoryManager(self.config).execute(format_payload(...))  # 流结束 → 才捕获
        raise

InvokeAsync(invoke.py:68)是同一套逻辑的异步版:await self._method(...),再看返回是不是 AsyncIterator / 有 __aiter__ / 是 gRPC 的 UnaryStreamCall,分流到 MemoriAsyncIterator。

---

5. 捕获:handle_post_response 攒出一份待处理载荷

分流的所有路径最终都汇到同一个终点:memori/llm/pipelines/post_invoke.py:94 的 handle_post_response。它做的事就是把"请求 kwargs + 原始响应"翻译成一份标准 JSON 载荷,然后交给记忆管线。

5.1 载荷长什么样

核心在 post_invoke.py:24 的 format_payload。它组装出这样一份 dict(BYODB 本地模式):

顶层字段	装什么	来源
attribution	entity.id / process.id——这轮对话归谁、属哪个流程	config.entity_id / config.process_id
conversation	client(provider/title/version)+ query(请求)+ response(响应)	入参 + convert_to_json(response)
session	session.uuid——归到哪个会话	config.session_id
meta	api key、sdk 版本、成功状态	config
time	本轮 start/end 时间戳	Invoke 记录的 start
messages	从上面 payload 解析出的规整 (role, text) 列表	parse_payload_conversation_messages(payload)

format_payload 内部还会调 parse_payload_conversation_messages(payload) 把请求/响应拍平成统一的 messages 列表(post_invoke.py:64),这一步正是各家 adapter 的 get_formatted_query / get_formatted_response 发挥作用的地方。

Cloud vs 本地的分叉: 若 config.cloud is True,format_payload 只回传精简版(attribution + messages + session,post_invoke.py:67-75),把重活留给云端;否则回传上面的完整本地载荷。两条路的存储差异见 05-storage.md。

5.2 attribution / session 语义(点到为止)

载荷里的三个身份字段回答"这段对话归属于谁":

• entity——记忆的主体(通常是"用户"),召回时按它检索。
• process——产生这段对话的流程/应用。
• session——一次会话,历史注入(§4.1)就按它读回。

本章只需知道它们在捕获这一刻被盖章进载荷。它们如何驱动后续的记忆生成与归并,属于 02-augmentation.md。

5.3 交棒:载荷离开本章

handle_post_response 的最后两步就是本章与下一章的交接线(post_invoke.py:130-144):

MemoryManager(invoke.config).execute(payload)          # ① 载荷进记忆管线
if invoke.config.augmentation is not None:             # ② 若开了增强
    handle_augmentation(config=..., payload=aug_input, ...)  # 交给增强管线(→02)

到这里,一次 LLM 调用已经被完整地"翻译"成了一份待处理载荷,并递交出去。载荷之后如何被炼成结构化记忆,是 02-augmentation.md 的事。 本章的旅程到此结束。

---

6. 巧妙之处(可借鉴的技术)

• 用注册表 + matcher 做开放扩展。 加一家新 provider,只需写一个 matcher 和一个包装类、挂上 @Registry.register_client,主流程零改动(_registry.py:33)。这是典型的开闭原则落地。
• 备份原方法再替换,包装器攥着它。 _wrap_method 先 getattr 存原方法,再 setattr 装包装器,原方法通过 _method 保命(_base.py:67-84)。这让"拦截"始终能回落到真实行为,不改变功能语义。
• 注入只碰 kwargs,不碰调用点。 记忆注入全部表现为"改请求参数"(invoke.py:28),因此对你的业务代码完全透明——这正是"零改代码"的技术根因。
• 流式响应延迟捕获。 不在调用点捕获,而是包一层迭代器透传 chunk、内部攒全,终止时才组载荷(iterator.py:61、iterable.py:35),既不破坏你的流式体验,又拿到完整响应。
• 用 _injected_message_count 防污染。 注入的历史会被打上计数,捕获时 adapter 用 _exclude_injected_messages 把它们剔除(_base.py:234 的 BaseLlmAdaptor),避免把"我们刚注进去的历史"又当成新对话重复入库。

7. 边界与局限(诚实)

• 靠 monkey patch,依赖 SDK 内部结构。 包装点是写死的方法路径(如 client.chat.completions.create)。上游 SDK 若改了方法路径或类名,匹配/替换可能失效——这是运行时 patch 的固有脆弱性。
• matcher 靠模块名前缀。 client_is_openai 等只看 __module__.startswith(...)(_utils.py:35-54),一个伪装/代理成相同模块前缀的对象可能被误判。
• matcher 顺序敏感。 匹配是首个命中即用(_registry.py:52),matcher 若不互斥,注册顺序会悄悄决定归属。
• LangChain 必须用命名参数。 直接 register(langchain_model) 会被显式拒绝并提示改用 register(chatopenai=...)(_registry.py:56-63),因为框架模型需要靠命名参数区分底层 provider。

8. 代码地图(导航索引)

主题	文件路径	符号名
注册表 / matcher 存取	memori/llm/_registry.py	Registry.register_client / register_adapter / client / adapter
对外注册总入口	memori/llm/_registry.py	register_llm
直连 client 包装类	memori/llm/clients/direct.py	Anthropic / OpenAi / Google / XAi / LiteLLM
框架(Agno/LangChain)注册	memori/llm/clients/frameworks.py	Agno.register / LangChain.register
provider 分流	memori/llm/_providers.py	Agno / Anthropic / Google / LangChain / OpenAi / XAi
client / provider matcher	memori/llm/_utils.py	client_is_anthropic / client_is_openai / llm_is_openai …
方法替换(patch)	memori/llm/_base.py	BaseClient._wrap_method
包装器基类	memori/llm/_base.py	BaseInvoke / BaseIterator / BaseLlmAdaptor
拦截主循环(同步/异步)	memori/llm/invoke/invoke.py	Invoke.invoke / InvokeAsync.invoke
流式捕获包装器	memori/llm/invoke/iterator.py iterable.py streaming.py	Iterator / Iterable / StreamingBody
请求前:注召回	memori/llm/pipelines/recall_injection.py	inject_recalled_facts
请求前:注历史	memori/llm/pipelines/conversation_injection.py	inject_conversation_messages / _inject_messages_by_provider
响应后:组载荷并交棒	memori/llm/pipelines/post_invoke.py	handle_post_response / format_payload
响应解析 adapter	memori/llm/adapters/openai/_adapter.py 等	Adapter.get_formatted_query / get_formatted_response