Day3 FAISS仅内存存储，重启丢失向量、无元数据、不支持过滤、无内置去重逻辑；
Chroma专为LLM RAG设计，核心优势：
| 客户端类型 | 存储位置 | 适用场景 |
|---|---|---|
| EphemeralClient | 内存 | 单元测试、临时调试 |
| PersistentClient | 本地文件夹磁盘 | 单机本地知识库、学习项目 |
| AsyncHttpClient | 远程Chroma服务 | 多实例分布式线上服务 |
filter={"type":"image","source":"产品手册"}传统文本Embedding只能编码文字；SigLIP/CLIP构建统一共享向量空间：
type=text/image元数据在Day6网关基础上新增multimodal_search工具，标准化输入：
网关统一封装SigLIP向量化、Chroma检索、元数据过滤逻辑，上层ReAct Agent无需关心图文底层差异。
用户提问（支持图文描述）
解决方案：
torch.no_grad()关闭梯度，大幅提速解决方案：
type元数据区分图文search_type过滤，按需只查文本/图片解决方案：
解决方案：
pip install chromadb torch transformers pillow scikit-learn
import asyncio
import chromadb
from chromadb.utils.embedding_functions import create_langchain_embedding
import torch
from PIL import Image
from transformers import AutoProcessor, AutoModel
import numpy as np
from typing import List, Dict, Optional
# ========== 1. SigLIP多模态嵌入本地模型 ==========
device = "cuda" if torch.cuda.is_available() else "cpu"
processor = AutoProcessor.from_pretrained("google/siglip-so400m-patch14-384")
siglip_model = AutoModel.from_pretrained("google/siglip-so400m-patch14-384").to(device).eval()
def text_to_vector(text: str) -> List[float]:
"""文本转多模态向量"""
with torch.no_grad():
inputs = processor(text=text, return_tensors="pt").to(device)
vec = siglip_model.get_text_embedding(**inputs)
return vec.squeeze().cpu().numpy().tolist()
def image_to_vector(img_path: str) -> List[float]:
"""图片转多模态向量"""
img = Image.open(img_path).convert("RGB")
with torch.no_grad():
inputs = processor(images=img, return_tensors="pt").to(device)
vec = siglip_model.get_image_embedding(**inputs)
return vec.squeeze().cpu().numpy().tolist()
# ========== 2. 持久化Chroma多模态向量库封装 ==========
class MultiModalChromaStore:
def __init__(self, persist_path="./chroma_mm_db", coll_name="mm_kb"):
# 本地持久化客户端
self.client = chromadb.PersistentClient(path=persist_path)
self.collection = self.client.get_or_create_collection(
name=coll_name,
metadata={"hnsw:space": "cosine"} # 使用余弦相似度
)
# 批量插入文本
async def add_text(self, text_list: List[str], source: str = "文档"):
ids = [f"txt_{i}" for i in range(len(text_list))]
vecs = [text_to_vector(t) for t in text_list]
metas = [{"type": "text", "source": source} for _ in text_list]
self.collection.add(
embeddings=vecs,
documents=text_list,
metadatas=metas,
ids=ids
)
# 插入单张图片
async def add_image(self, img_path: str, desc: str, source: str = "图片库"):
vec = image_to_vector(img_path)
self.collection.add(
embeddings=[vec],
documents=[desc],
metadatas=[{"type": "image", "source": source, "img_path": img_path}],
ids=[f"img_{img_path.replace('/','_')}"]
)
# 多模态检索
async def search(self, query: str, search_type: str = "all", top_k=3, use_mmr=False):
query_vec = text_to_vector(query)
where_filter = {}
if search_type == "text":
where_filter = {"type": "text"}
elif search_type == "image":
where_filter = {"type": "image"}
if use_mmr:
res = self.collection.max_marginal_relevance_search(
query_embeddings=[query_vec],
n_results=top_k,
where=where_filter
)
else:
res = self.collection.query(
query_embeddings=[query_vec],
n_results=top_k,
where=where_filter
)
output = []
for doc, meta, dist in zip(res["documents"][0], res["metadatas"][0], res["distances"][0]):
output.append({
"content": doc,
"meta": meta,
"distance": round(float(dist), 4)
})
return output
# 测试多模态库
async def test_chroma_mm():
mm_store = MultiModalChromaStore()
# 写入文本知识库
await mm_store.add_text([
"多模态RAG使用SigLIP实现图文统一向量检索",
"Chroma支持本地持久化，重启不丢失向量数据",
"ReAct Agent可以自主调用图文检索工具"
], source="Day7学习文档")
# 文字检索全部内容
ret = await mm_store.search("什么是多模态RAG", search_type="all")
print("检索结果：", ret)
if __name__ == "__main__":
asyncio.run(test_chroma_mm())
import asyncio
from pydantic import BaseModel, Field
from tool_gateway import gateway, ToolMeta
from chroma_mm_store import MultiModalChromaStore
# 全局多模态向量库实例
mm_chroma = MultiModalChromaStore()
# 多模态检索参数模型
class MultiModalSearchParams(BaseModel):
query: str = Field(description="检索关键词/图片描述")
search_type: str = Field(description="检索类型 all/text/image")
top_k: int = Field(default=2, description="返回条数")
use_mmr: bool = Field(default=False, description="是否开启去重检索")
# 新增图文检索异步工具
async def multimodal_search(query: str, search_type: str, top_k: int, use_mmr: bool):
res = await mm_chroma.search(query, search_type, top_k, use_mmr)
if not res:
return "未检索到相关图文资料"
out_text = ""
for item in res:
t = item["meta"]["type"]
src = item["meta"]["source"]
out_text += f"【{t}|{src}】相似度距离：{item['distance']}\n内容：{item['content']}\n\n"
return out_text
# 注册新工具到网关
gateway.register_tool(ToolMeta(
name="multimodal_search",
description="多模态知识库检索，可查询文本资料、产品截图、流程图等图片信息；描述图表/图片使用image类型，查询理论知识使用text类型",
params_model=MultiModalSearchParams,
run_func=multimodal_search
))
import asyncio
from react_agent import ReActAgent
from tool_gateway_mm import gateway
class MultiModalReActAgent(ReActAgent):
def __init__(self, model_name="qwen-turbo"):
super().__init__(model_name)
# 更新系统提示词，新增多模态工具规则
self.system_prompt = """
你是多模态ReAct智能体，遵循Thought -> Action -> Observation循环。
可用工具：
1. calculator：数学加减乘除运算
2. multimodal_search：图文混合知识库检索
- 查询理论、课程文字资料：search_type="text"
- 查询流程图、产品截图、图表描述：search_type="image"
- 全部资料一起查：search_type="all"
规则：
1. 涉及图片、图表类问题必须调用multimodal_search并指定image类型
2. 数学计算禁止心算，必须调用calculator
3. 知识库无内容禁止编造信息
4. 每轮执行工具后自动反思信息是否充足
"""
self.tools_schema = gateway.get_openai_tools_schema()
# 测试多模态Agent
async def test_mm_agent():
agent = MultiModalReActAgent()
await agent.init_memory()
# 写入测试图文知识库
from chroma_mm_store import mm_chroma
await mm_chroma.add_text(["SigLIP是跨模态图文嵌入模型，用于多模态RAG检索"], source="AI知识库")
# 复合型问题：先计算，再查询多模态RAG知识
query = "计算 (50+10)*3，再介绍多模态RAG和SigLIP作用"
res = await agent.chat(session_id="mm_test_001", user_query=query)
print("最终回答：\n", res["answer"])
await agent.close_memory()
if __name__ == "__main__":
asyncio.run(test_mm_agent())
from fastapi import FastAPI, Query
import asyncio
from react_mm_agent import MultiModalReActAgent
app = FastAPI(title="Day7 多模态ReAct Agent｜Chroma图文知识库")
agent = MultiModalReActAgent()
@app.on_event("startup")
async def startup():
await agent.init_memory()
@app.on_event("shutdown")
async def shutdown():
await agent.close_memory()
@app.get("/agent/mm_chat")
async def mm_chat(
session_id: str = Query(..., description="会话ID"),
prompt: str = Query(..., description="用户提问，可描述图片、数学、理论知识")
):
result = await agent.chat(session_id, prompt)
return result
if __name__ == "__main__":
import uvicorn
uvicorn.run("main_mm_react:app", reload=True)
chroma_mm_store.py，测试文本入库、跨模态检索、元数据过滤功能，重启程序验证向量持久化add_image存入向量库，用文字描述实现以文搜图search_type分别为text/image/all，对比检索返回内容差异FAISS vs Chroma
FAISS仅内存存储、无元数据、无过滤、无MMR；适合临时测试小批量向量。
Chroma支持磁盘持久化、元数据过滤、MMR去重、异步客户端、多集合隔离；适合单机本地RAG学习/小型业务落地。
SigLIP跨模态嵌入
统一图文共享向量空间，图片与文字编码到相同维度向量；语义相近图文向量距离近，因此输入文字描述可检索匹配图片。
Chroma客户端
Ephemeral内存临时；Persistent本地磁盘持久单机；AsyncHttpClient连接远程Chroma服务，分布式多实例使用。
MMR最大边际相关性
平衡相似度与结果多样性，过滤高度重复的知识库片段，避免检索返回内容高度雷同，提升问答信息丰富度。
多模态RAG新增能力
支持图片、图文混合知识库存储检索；可通过文字描述查找图片资料；适配手册截图、流程图、图表类私有资料。
元数据过滤
入库时给每条数据添加type字段区分text/image；检索传入where过滤条件，按需只返回文本或图片片段。
Chroma存储优化
拆分多Collection隔离图文；定期清理无用过期向量；检索使用MMR减少冗余结果；单机大数据切换Milvus/Qdrant。
SigLIP推理优化
使用GPU加速；关闭梯度计算torch.no_grad()；压缩输入图片分辨率；提前缓存模型本地离线加载。
ReAct区分图文检索
System提示词明确规则，图表/截图描述调用image检索；理论文字使用text检索；搭配Few-shot示例规范工具入参。
大规模向量库选型
开源分布式：Milvus、Qdrant；云托管向量库：阿里云DashVector、Pinecone；Chroma仅适合单机小规模场景。
OCR+向量混合检索
图片先用OCR提取文字存入文本向量库；原图SigLIP存入多模态库；检索同时召回OCR文本与图片向量，合并结果重排。
多模态会话持久化
Redis存储对话文字；图片路径、图片描述存入消息元数据；用户再次提问时，根据历史图片描述向量召回对应图片知识库。
Day7完成向量库生产级升级（Chroma替代FAISS）与多模态RAG完整落地，打通图文混合知识库检索能力。整合前六天异步LLM、SSE、Function Calling、统一工具网关、ReAct动态反思、Redis分层记忆全套底层体系，实现能处理图文、数学、私有文档的全能本地智能体。
下一日学习：Agent工作流编排、多智能体协作、工具链批量自动化任务。
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。