5.2 核心组件：Chain、Agent、Memory、Callback、Retriever

不管你用哪个框架，底层都是由几个核心组件组成的。理解这些组件的本质，比记住某个框架的 API 更重要——因为框架会变，但这些概念的逻辑不会变。这篇文章拆解 AI 应用开发的 5 个核心组件，让你「知其然更知其所以然」。

📑 目录

• Chain：数据处理的流水线
• Agent：自治决策者
• Memory：对话的连续性保障
• Callback：可观测性的基础
• Retriever：知识的大门
• 组件协作关系总览

Chain：数据处理的流水线

一句话定义

将多个处理步骤串联成一个管道，上一步的输出作为下一步的输入。就像工厂的装配流水线——原材料进入，经过各道工序，最终变成成品。

# Chain 的本质就是一个函数组合
class Chain:
def __init__(self, steps):
self.steps = steps # [step1, step2, step3, …]

def invoke(self, input_data):
data = input_data
for step in self.steps:
data = step(data) # 上一步输出 → 下一步输入
return data

# 实际例子：一个 RAG Chain
rag_chain = Chain([
retrieve_documents, # Step 1: 检索相关文档
format_context, # Step 2: 格式化为 Prompt 上下文
call_llm, # Step 3: 调用 LLM 生成回答
parse_output, # Step 4: 解析结构化输出
])

answer = rag_chain.invoke({"query": "什么是 RAG？"})

常见的 Chain 类型

Agent：自治决策者

（详见 4.1 智能体核心，这里从组件视角补充关键点）

# Agent 作为组件的核心接口
from abc import ABC, abstractmethod

class BaseAgent(ABC):
@abstractmethod
def plan(self, task) –> list:
"""制定执行计划"""
pass

@abstractmethod
def execute_step(self, action):
"""执行单个步骤"""
pass

@abstractmethod
def observe(self, result) –> str:
"""观察结果并决定下一步"""
pass

def run(self, task, max_steps=10):
"""主循环：ReAct"""
plan = self.plan(task)
for _ in range(max_steps):
action = self.decide(plan)
result = self.execute_step(action)
if self.is_complete(result):
return result
return "达到最大步数限制"

Memory：对话的连续性保障

一句话定义

负责管理对话历史和跨会话信息的组件。没有 Memory，每次对话都是「失忆状态」。

Memory 组件的类型体系

Memory 组件层次：

┌─────────────────────────────────────┐
│ BaseMemory │
│ （抽象接口：load / save / clear） │
└──────────────┬──────────────────────┘
┌───────┼───────────┐
↓ ↓ ↓
┌──────────┐ ┌──────────┐ ┌──────────┐
│BufferMem │ │EntityMem │ │VectorMem │
│ (缓冲区) │ │ (实体提取)│ │ (向量) │
│ │ │ │ │ │
│ 滑动窗口 │ │ 用户画像 │ │ 语义搜索 │
│ 最近N轮 │ │ 关系图谱 │ │ 跨会话 │
└──────────┘ └──────────┘ └──────────┘

Callback：可观测性的基础

一句话定义

在框架的关键节点插入钩子函数，让你能在不修改核心逻辑的情况下监听事件、收集指标、记录日志。是做调试、追踪、监控的基础设施。

# Callback 的典型用途
class MyCallback:
def on_llm_start(self, prompts):
print(f"[LLM 开始] 发送 {len(prompts)} 条 prompt")

def on_llm_end(self, response):
tokens = response.token_usage
print(f"[LLM 结束] 用了 {tokens} tokens")

def on_tool_start(self, tool_name, args):
print(f"[Tool] 调用 {tool_name}({args})")

def on_tool_end(self, tool_name, result):
print(f"[Tool] {tool_name} 返回 {len(str(result))} 字符")

def on_chain_end(self, outputs):
print(f"[Chain 完成] 最终输出已生成")

# 使用：注入到 Chain/Agent 中
chain = build_rag_chain(callbacks=[MyCallback()])
result = chain.invoke(query)
# 控制台自动打印完整的调用链日志！

Callback 的实际价值

Retriever：知识的大门

一句话定义

统一的检索接口。不管背后用的是 FAISS、Chroma 还是 Milvus，对外暴露的都是 query(str) -> list[Document] 这一个方法。

from abc import ABC, abstractmethod

class BaseRetriever(ABC):
@abstractmethod
def get_relevant_documents(self, query: str) –> list:
"""核心接口：查询 → 返回相关文档列表"""
pass

# 不同后端的 Retriever 实现
class VectorRetriever(BaseRetriever):
def __init__(self, vector_store):
self.store = vector_store

def get_relevant_documents(self, query):
return self.store.search(embed(query), top_k=5)

class HybridRetriever(BaseRetriever):
def __init__(self, vector_store, bm25_store):
self.vs = vector_store
self.bm25 = bm25_store

def get_relevant_documents(self, query):
vec_results = self.vs.search(embed(query), top_k=10)
kw_results = self.bm25.search(query, top_k=10)
return rrf_merge(vec_results, kw_results, k=5)

# 切换检索后端只需换 Retriever 实现，不影响上层逻辑

组件协作关系总览

┌────────────────────────────────────────────┐
│ AI 应用的组件架构 │
│ │
│ 用户 Query │
│ ↓ │
│ ┌──────────┐ │
│ │Retriever │ ← 从知识库检索相关内容 │
│ │ (大门) │ │
│ └────┬─────┘ │
│ ↓ Documents │
│ ┌──────────┐ ┌──────────────────┐ │
│ │ Memory │←──→│ Chain │ │
│ │ (历史) │ │ 组装 Prompt+检索 │ │
│ └──────────┘ │ 结果 → 调用 LLM │ │
│ └────────┬─────────┘ │
│ ↓ │
│ ┌──────────────────┐ │
│ │ Agent │ │
│ │ 决策/工具调用/ │ │
│ │ 多轮推理 │ │
│ └────────┬─────────┘ │
│ ↓ │
│ ┌──────────────────┐ │
│ │ Callback │ │
│ │ 全程追踪/日志 │ │
│ └──────────────────┘ │
│ │
│ 关键认知： │
│ Retriever取料 → Memory补上下文 → Chain组装 │
│ → Agent决策 → Callback记录全过程 │
└────────────────────────────────────────────┘

❌ 常见误区

• ❌ 组件越多越好 — 每多一个组件就多一层抽象和潜在故障点
• ❌ 必须用框架提供的组件 — 很多时候自定义组件更简单直接
• Callback 只用于调试 — 生产环境的监控、计费、审计都靠它