文档

文档

Weaviate + ModelRiver

由 ModelRiver 大语言模型嵌入驱动的开源向量搜索。为表示向量的生成以及 RAG 查询提供双重自动故障转移机制。

概览

Weaviate 是一个支持自定义向量化引擎 (custom vectorisers) 的开源向量搜索引擎。通过将嵌入生成 (embedding) 和生成式调用路由到 ModelRiver，您将能够在整个检索搜索管道获得模型提供商故障转移机制。

快速上手

安装相关依赖

Bash
pip install weaviate-client openai

初始化配置

PYTHON
1import weaviate
2from openai import OpenAI
3 
4# ModelRiver 客户端
5client = OpenAI(
6 base_url="https://api.modelriver.com/v1",
7 api_key="mr_live_YOUR_API_KEY",
8)
9 
10# Weaviate 客户端
11wv_client = weaviate.connect_to_local() # 或使用 connect_to_weaviate_cloud()

带有 ModelRiver 表示向量集合的创建 (Create collection with ModelRiver embeddings)

PYTHON
1def embed(texts: list[str]) -> list[list[float]]:
2 """通过 ModelRiver 生成表示向量。"""
3 response = client.embeddings.create(
4 model="my-embedding-workflow",
5 input=texts,
6 )
7 return [d.embedding for d in response.data]
8 
9# 创建集合
10collection = wv_client.collections.create(
11 name="Documents",
12 properties=[
13 weaviate.classes.config.Property(name="text", data_type=weaviate.classes.config.DataType.TEXT),
14 weaviate.classes.config.Property(name="source", data_type=weaviate.classes.config.DataType.TEXT),
15 ],
16)

收录文档数据 (Ingest documents)

PYTHON
1documents = [
2 {"text": "ModelRiver routes AI requests across providers.", "source": "docs"},
3 {"text": "Workflows configure provider and fallback settings.", "source": "docs"},
4 {"text": "Structured outputs guarantee JSON schema compliance.", "source": "docs"},
5]
6 
7# 生成向量并写入
8texts = [doc["text"] for doc in documents]
9vectors = embed(texts)
10 
11with collection.batch.dynamic() as batch:
12 for doc, vector in zip(documents, vectors):
13 batch.add_object(
14 properties=doc,
15 vector=vector,
16 )

语义搜索与检索增强生成 (Semantic search + RAG)

PYTHON
1def search_and_answer(question: str, top_k: int = 3) -> str:
2 """搜索 Weaviate 集合，然后通过 ModelRiver 产出问题的回答。"""
3 
4 # 嵌入查询语句
5 query_vector = embed([question])[0]
6 
7 # 执行 Weaviate 检索
8 results = collection.query.near_vector(
9 near_vector=query_vector,
10 limit=top_k,
11 return_properties=["text", "source"],
12 )
13 
14 # 提取构建上下文环境
15 context = "\n\n".join([obj.properties["text"] for obj in results.objects])
16 
17 # 总结生成回答
18 response = client.chat.completions.create(
19 model="my-chat-workflow",
20 messages=[
21 {"role": "system", "content": f"请严格基于以下上下文作答：\n\n{context}"},
22 {"role": "user", "content": question},
23 ],
24 )
25 
26 return response.choices[0].message.content
27 
28answer = search_and_answer("How does ModelRiver handle failover?")
29print(answer)

最佳实践 (Best practices)

  1. 分离嵌入和聊天工作流 (Separate embedding and chat workflows): 以实现独立的规模伸缩和速率调优。
  2. 在处理大数据集时尽量使用批量导入 (Use batch imports): Weaviate 的 batch API 在此时会快得多。
  3. 存储来源元数据 (Store source metadata): 这便于引用溯源及出处跟踪。
  4. 监控嵌入相关的花销成本 (Monitor embedding costs): 使用 请求日志 分离查询与监控。

下一步探究