Context Engineering 实战：Weaviate 构建智能体系统完整指南（附代码）
本文介绍了如何利用Weaviate向量数据库构建智能体系统的上下文管理系统。通过ContextEngineering理念，将上下文管理分为采集、存储、检索和更新四个环节，借助Weaviate的语义检索和向量存储能力，为智能体提供精准的信息支撑。文章详细演示了从环境搭建、数据导入到智能体交互的全流程实现，包括Weaviate初始化、上下文数据嵌入、语义检索以及与LLM的集成方法。同时提供了避坑指南和
在大模型智能体（Agent）开发中，“上下文工程（Context Engineering）” 是决定智能体决策质量的核心 —— 通过高效的上下文获取、组织与检索，让智能体具备精准的信息支撑。Weaviate 作为开源向量数据库，凭借强大的语义检索、向量存储与知识图谱能力，成为智能体上下文管理的优选工具。本文将结合 Context Engineering 理念，带大家用 Weaviate 搭建完整的智能体系统，覆盖从环境搭建、数据导入、上下文设计到智能体交互的全流程，代码精简可直接复用，新手也能快速落地。
一、核心概念与架构设计
1. 关键术语说明
|
术语 |
核心作用 |
|
Context Engineering |
上下文工程，含上下文采集、结构化存储、智能检索、动态更新四大环节 |
|
Weaviate |
开源向量数据库，支持向量存储、语义搜索、知识图谱查询，适配上下文管理 |
|
智能体系统 |
具备 “感知 - 决策 - 执行” 能力的系统，核心依赖上下文信息完成任务 |
|
向量嵌入 |
将文本、图片等数据转化为向量，实现语义层面的相似度匹配与检索 |
|
知识图谱 |
以图结构存储实体关系，为智能体提供结构化知识支撑 |
2. 智能体系统架构（基于 Weaviate）
3. 核心优势
- 上下文精准性：Weaviate 的语义检索能力，确保智能体获取最相关的上下文信息；
- 实时性：支持上下文动态更新，适配任务场景的实时变化；
- 扩展性：兼容多模态数据（文本、图片），支持知识图谱扩展，满足复杂任务需求；
- 轻量化：部署简单，API 友好，可快速集成到现有智能体系统。
二、前置准备（10 分钟搞定）
1. 环境配置要求
|
组件 |
版本要求 |
说明 |
|
Python |
3.8+ |
核心开发语言 |
|
Weaviate |
1.23.0+ |
向量数据库（支持 Docker 部署） |
|
依赖库 |
weaviate-client==4.5.0、openai==1.3.5、python-dotenv==1.0.0 |
Weaviate 客户端 + 向量嵌入 + 环境变量管理 |
|
大模型 API |
OpenAI API 密钥（或其他支持嵌入的大模型） |
用于生成数据向量嵌入 |
2. 环境搭建步骤
（1）部署 Weaviate（Docker 快速部署）
- 创建docker-compose.yml文件：
version: '3.4'
services:
weaviate:
image: semitechnologies/weaviate:1.23.3
ports:
- "8080:8080" # 访问端口
environment:
- AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED=true # 匿名访问（开发环境）
- PERSISTENCE_DATA_PATH=/var/lib/weaviate
- DEFAULT_VECTORIZER_MODULE=none # 禁用内置向量器（使用外部嵌入）
volumes:
- weaviate_data:/var/lib/weaviate
volumes:
weaviate_data:
- 启动 Weaviate：
docker-compose up -d
- 验证部署：访问http://localhost:8080/v1/meta，返回 Weaviate 元信息即部署成功。
（2）安装依赖库
pip install weaviate-client openai python-dotenv
（3）获取关键凭证
- OpenAI API 密钥（用于生成向量嵌入，获取地址：https://platform.openai.com/）；
- Weaviate 访问地址（本地部署默认：http://localhost:8080）。
三、核心步骤：Weaviate 构建智能体上下文系统
1. 步骤 1：初始化 Weaviate 客户端与数据模式
定义 Weaviate 数据模式（Schema），用于存储智能体的上下文数据（以 “任务上下文” 为例）：
import weaviate
from weaviate.classes.config import Property, DataType
from dotenv import load_dotenv
import os
# 加载环境变量
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
WEAVIATE_URL = os.getenv("WEAVIATE_URL", "http://localhost:8080")
# 1. 初始化Weaviate客户端
client = weaviate.Client(
url=WEAVIATE_URL,
additional_headers={"X-OpenAI-Api-Key": OPENAI_API_KEY} # 用于Weaviate内置OpenAI嵌入（可选）
)
# 2. 定义数据模式（Schema）：存储任务上下文
schema = {
"class": "TaskContext", # 类名：任务上下文
"properties": [
{
"name": "taskId",
"dataType": ["string"],
"description": "任务ID"
},
{
"name": "taskName",
"dataType": ["string"],
"description": "任务名称"
},
{
"name": "contextContent",
"dataType": ["text"],
"description": "上下文内容（如任务要求、历史交互、相关知识）"
},
{
"name": "timestamp",
"dataType": ["date"],
"description": "上下文创建时间"
}
],
"vectorizer": "none" # 手动传入向量（外部生成）
}
# 3. 创建Schema（不存在则创建）
if not client.schema.exists("TaskContext"):
client.schema.create_class(schema)
print("Weaviate Schema创建成功！")
else:
print("Weaviate Schema已存在，跳过创建")
2. 步骤 2：上下文数据导入（含向量嵌入）
将任务上下文数据转化为向量并导入 Weaviate，为智能体提供检索数据源：
from openai import OpenAI
from datetime import datetime
# 初始化OpenAI客户端（生成向量嵌入）
openai_client = OpenAI(api_key=OPENAI_API_KEY)
def generate_embedding(text: str) -> list[float]:
"""生成文本的向量嵌入（使用OpenAI Embeddings）"""
response = openai_client.embeddings.create(
input=text,
model="text-embedding-3-small" # 轻量版嵌入模型，适合上下文检索
)
return response.data[0].embedding
# 示例上下文数据（智能体任务相关）
context_data_list = [
{
"taskId": "task_001",
"taskName": "用户订单查询",
"contextContent": "用户需要查询2024年10月的订单，订单号前缀为OD202410，支持按手机号或订单号查询，需返回订单状态、金额、物流信息"
},
{
"taskId": "task_002",
"taskName": "产品售后申请",
"contextContent": "售后申请需满足：收货后7天内，产品未拆封、无质量问题不支持退货；质量问题需提供检测报告，退货物流费由商家承担"
},
{
"taskId": "task_003",
"taskName": "会员积分兑换",
"contextContent": "会员积分100分可兑换1元优惠券，积分有效期为1年，兑换的优惠券30天内有效，支持兑换实物商品或服务"
}
]
# 导入数据到Weaviate
for context_data incontext_data_list:
# 生成上下文内容的向量嵌入
embedding = generate_embedding(context_data["contextContent"])
# 构造Weaviate数据对象
data_object = {
"taskId": context_data["taskId"],
"taskName": context_data["taskName"],
"contextContent": context_data["contextContent"],
"timestamp": datetime.now().isoformat() # 当前时间
}
# 导入数据（关联向量）
client.data_object.create(
data_object=data_object,
class_name="TaskContext",
vector=embedding # 手动传入向量
)
print(f"成功导入{len(context_data_list)}条上下文数据到Weaviate！")
3. 步骤 3：上下文检索（智能体获取相关信息）
实现基于语义的上下文检索，让智能体在处理任务时快速获取相关上下文：
def retrieve_context(query: str, top_k: int = 2) -> list[dict]:
"""
语义检索相关上下文
:param query: 智能体当前任务查询（如用户提问）
:param top_k: 返回最相关的top_k条上下文
:return: 相关上下文列表
"""
# 生成查询的向量嵌入
query_embedding = generate_embedding(query)
# Weaviate向量检索
response = client.data_object.get_with_vector(
vector=query_embedding,
class_name="TaskContext",
limit=top_k,
return_metadata=["distance"] # 返回相似度距离（越小越相似）
)
# 处理检索结果
context_list = []
for item in response.objects:
context = {
"taskName": item.properties["taskName"],
"contextContent": item.properties["contextContent"],
"similarity": 1 - item.metadata["distance"] # 转换为相似度（0-1）
}
context_list.append(context)
return context_list
# 测试检索：用户提问“如何用积分兑换优惠券”
user_query = "如何用积分兑换优惠券"
relevant_contexts = retrieve_context(user_query)
print(f"用户查询：{user_query}")
print("检索到的相关上下文：")
for idx, context in enumerate(relevant_contexts, 1):
print(f"\n[{idx}] 任务名称：{context['taskName']}")
print(f"相似度：{context['similarity']:.4f}")
print(f"上下文内容：{context['contextContent']}")
4. 步骤 4：构建智能体核心逻辑（上下文 + 大模型）
整合 Weaviate 上下文检索与大模型，实现智能体的决策与响应：
def agent_response(user_query: str) -> str:
"""
智能体响应逻辑：检索上下文→拼接提示词→大模型生成响应
"""
# 1. 检索相关上下文
relevant_contexts = retrieve_context(user_query)
# 2. 构建提示词（整合上下文+用户查询）
context_text = "\n\n".join([f"【{c['taskName']}】{c['contextContent']}" for c in relevant_contexts])
prompt = f"""
你是一个智能客服代理，基于以下上下文信息回答用户问题：
{context_text}
要求：
1. 仅使用上下文信息回答，不编造内容；
2. 回答简洁明了，贴合用户查询；
3. 若上下文无相关信息，回复“抱歉，暂无相关信息可提供”。
用户问题：{user_query}
"""
# 3. 大模型生成响应
response = openai_client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
temperature=0.3 # 降低随机性，确保回答准确
)
return response.choices[0].message.content
# 测试智能体响应
user_queries = [
"如何用积分兑换优惠券？",
"售后退货需要满足什么条件？",
"我想查询2024年10月的订单，需要提供什么信息？",
"如何办理会员升级？" # 无相关上下文
]
for query in user_queries:
print(f"\n用户：{query}")
print(f"智能体：{agent_response(query)}")
5. 步骤 5：上下文动态更新（智能体反馈闭环）
智能体执行任务后，将新的上下文数据更新到 Weaviate，实现上下文迭代：
def update_context(task_id: str, task_name: str, context_content: str):
"""
动态更新上下文到Weaviate
"""
# 生成新上下文的向量嵌入
embedding = generate_embedding(context_content)
# 构造数据对象
data_object = {
"taskId": task_id,
"taskName": task_name,
"contextContent": context_content,
"timestamp": datetime.now().isoformat()
}
# 导入Weaviate（若taskId已存在，可先删除再添加，或更新）
client.data_object.create(
data_object=data_object,
class_name="TaskContext",
vector=embedding
)
print(f"成功更新上下文：{task_name}")
# 示例：新增“会员升级”上下文（智能体无相关信息后补充）
update_context(
task_id="task_004",
task_name="会员升级",
context_content="会员升级条件：普通会员累计消费满1000元可升级为银卡会员，满5000元升级为金卡会员，升级后可享受对应等级折扣和专属服务"
)
# 重新测试“会员升级”查询
print("\n用户：如何办理会员升级？")
print(f"智能体：{agent_response('如何办理会员升级？')}")
四、实战场景：智能客服智能体完整代码
整合上述步骤，实现一个完整的智能客服智能体，支持上下文检索、响应与动态更新：
# 完整智能客服智能体代码（可直接运行）
import weaviate
from openai import OpenAI
from dotenv import load_dotenv
import os
from datetime import datetime
# 加载环境变量
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
WEAVIATE_URL = os.getenv("WEAVIATE_URL", "http://localhost:8080")
# 初始化客户端
weaviate_client = weaviate.Client(url=WEAVIATE_URL)
openai_client = OpenAI(api_key=OPENAI_API_KEY)
# -------------------------- 核心工具函数 --------------------------
def generate_embedding(text: str) -> list[float]:
"""生成文本向量嵌入"""
response = openai_client.embeddings.create(
input=text, model="text-embedding-3-small"
)
return response.data[0].embedding
def init_weaviate_schema():
"""初始化Weaviate Schema"""
schema = {
"class": "TaskContext",
"properties": [
{"name": "taskId", "dataType": ["string"], "description": "任务ID"},
{"name": "taskName", "dataType": ["string"], "description": "任务名称"},
{"name": "contextContent", "dataType": ["text"], "description": "上下文内容"},
{"name": "timestamp", "dataType": ["date"], "description": "创建时间"}
],
"vectorizer": "none"
}
if not weaviate_client.schema.exists("TaskContext"):
weaviate_client.schema.create_class(schema)
print("Schema初始化成功")
def import_initial_contexts():
"""导入初始上下文数据"""
initial_contexts = [
{
"taskId": "task_001",
"taskName": "用户订单查询",
"contextContent": "用户查询2024年10月订单，订单号前缀OD202410，支持手机号或订单号查询，返回状态、金额、物流"
},
{
"taskId": "task_002",
"taskName": "产品售后申请",
"contextContent": "售后7天内可申请，未拆封无质量问题不退货；质量问题需检测报告，运费商家承担"
},
{
"taskId": "task_003",
"taskName": "会员积分兑换",
"contextContent": "100积分=1元优惠券，积分有效期1年，优惠券30天有效，可兑换实物或服务"
}
]
for ctx in initial_contexts:
embedding = generate_embedding(ctx["contextContent"])
weaviate_client.data_object.create(
data_object={**ctx, "timestamp": datetime.now().isoformat()},
class_name="TaskContext",
vector=embedding
)
print("初始上下文导入成功")
def retrieve_context(query: str, top_k: int = 2) -> list[dict]:
"""检索相关上下文"""
query_embedding = generate_embedding(query)
response = weaviate_client.data_object.get_with_vector(
vector=query_embedding, class_name="TaskContext", limit=top_k, return_metadata=["distance"]
)
return [
{
"taskName": item.properties["taskName"],
"contextContent": item.properties["contextContent"],
"similarity": 1 - item.metadata["distance"]
}
for item in response.objects
]
def agent_response(user_query: str) -> str:
"""智能体响应"""
contexts = retrieve_context(user_query)
context_text = "\n\n".join([f"【{c['taskName']}】{c['contextContent']}" for c in contexts])
prompt = f"""
基于以下上下文回答用户问题，仅用上下文信息，不编造，简洁明了：
{context_text}
用户问题：{user_query}
无相关信息则回复“抱歉，暂无相关信息可提供”
"""
response = openai_client.chat.completions.create(
model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}], temperature=0.3
)
return response.choices[0].message.content
def update_context(task_id: str, task_name: str, context_content: str):
"""更新上下文"""
embedding = generate_embedding(context_content)
weaviate_client.data_object.create(
data_object={
"taskId": task_id,
"taskName": task_name,
"contextContent": context_content,
"timestamp": datetime.now().isoformat()
},
class_name="TaskContext",
vector=embedding
)
print(f"上下文更新成功：{task_name}")
# -------------------------- 运行智能体 --------------------------
if __name__ == "__main__":
# 初始化
init_weaviate_schema()
if not weaviate_client.data_object.get(class_name="TaskContext", limit=1).objects:
import_initial_contexts()
# 交互测试
print("智能客服智能体已启动，输入'quit'退出")
while True:
user_input = input("\n用户：")
if user_input.lower() == "quit":
print("智能体：再见！")
break
response = agent_response(user_input)
print(f"智能体：{response}")
# 示例：若查询“会员升级”，自动更新上下文（实际场景可由人工或系统触发）
if "会员升级" in user_input and not any("会员升级" in c["taskName"] for c in retrieve_context(user_input)):
update_context(
task_id="task_004",
task_name="会员升级",
context_content="普通会员消费满1000元升银卡，5000元升金卡，享对应折扣和专属服务"
)
五、避坑指南与最佳实践
1. 常见问题与解决方案
（1）Weaviate 连接失败
- 错误提示：weaviate.exceptions.WeaviateConnectionError；
- 解决方法：
-
- 检查 Docker 容器是否启动（docker ps查看 weaviate 容器状态）；
-
- 确认 Weaviate 访问地址正确（本地默认http://localhost:8080）；
-
- 关闭防火墙或开放 8080 端口，确保网络通畅。
（2）向量嵌入生成失败
- 错误提示：openai.APIError或密钥无效；
- 解决方法：
-
- 检查 OpenAI API 密钥是否正确，是否有余额；
-
- 若无法访问 OpenAI，可替换为国内大模型嵌入（如通义千问、讯飞星火）；
-
- 降低嵌入模型复杂度（如用text-embedding-ada-002替代text-embedding-3-large）。
（3）检索结果不相关
- 错误原因：向量嵌入模型不匹配，或上下文数据质量低；
- 解决方法：
-
- 统一嵌入模型（检索与导入使用同一模型）；
-
- 优化上下文内容（简洁明了，突出核心信息）；
-
- 调整top_k参数（适当增加返回条数），或使用 Weaviate 过滤条件（如按任务类型过滤）。
（4）Weaviate 数据量过大导致检索变慢
- 解决方法：
-
- 启用 Weaviate 索引优化（如 HNSW 索引参数调整）；
-
- 定期清理过期上下文数据（按timestamp字段删除）；
-
- 分片存储（大规模场景下，按任务类型分 Class 存储）。
2. 最佳实践
（1）上下文设计原则
- 结构化：上下文内容需包含 “任务类型 + 核心规则 + 操作步骤”，便于大模型理解；
- 轻量化：单条上下文长度控制在 500 字内，避免冗余信息影响检索精度；
- 时效性：通过timestamp字段标记上下文，检索时优先返回最新数据。
（2）Weaviate 优化配置
- 向量维度：选择合适的嵌入模型维度（如text-embedding-3-small的 1536 维，平衡精度与性能）；
- 索引参数：调整 HNSW 索引参数（如efConstruction=128、ef=64），优化检索速度；
- 批量操作：批量导入 / 更新上下文数据，减少 API 调用次数（使用client.data_object.batch_create）。
（3）智能体与上下文协同
- 上下文拼接策略：仅将最相关的 top 2-3 条上下文传入大模型，避免上下文过长导致性能下降；
- 反馈闭环：智能体响应后，通过用户评价或系统检测，动态更新优质上下文、删除无效信息；
- 多模态扩展：若需处理图片类上下文（如产品故障图片），可使用 Weaviate 的多模态向量存储能力。
（4）安全性建议
- 敏感上下文数据（如用户隐私）需加密存储（Weaviate 支持数据加密）；
- 限制 Weaviate 访问权限（生产环境禁用匿名访问，启用 API 密钥认证）；
- 大模型提示词中添加安全限制，避免泄露上下文敏感信息。
六、总结
基于 Context Engineering 理念与 Weaviate 向量数据库，构建智能体系统的核心是 “高效的上下文管理”—— 通过 Weaviate 实现上下文的结构化存储、语义检索与动态更新，让智能体在处理任务时能精准获取所需信息，提升决策质量与响应准确性。
本文的核心要点可总结为：
- 架构设计：明确 “数据导入→向量存储→上下文检索→大模型决策→反馈更新” 的核心流程；
- 核心操作：Weaviate Schema 设计、向量嵌入生成、语义检索、上下文动态更新，代码精简可复用；
- 实战价值：完整实现智能客服智能体，支持常见业务场景，可快速迁移至其他智能体系统（如办公助手、工业控制智能体）；
- 优化方向：通过调整嵌入模型、Weaviate 配置、上下文设计，平衡智能体的响应速度与准确性。
无论是开发轻量级智能体，还是构建复杂的多任务智能体系统，Weaviate 都能提供强大的上下文支撑。如果需要扩展功能（如知识图谱关联、多模态上下文处理、分布式部署），或对接特定大模型（如本地部署的 LLaMA 3），可以告诉我具体需求，将提供针对性的实现方案！
来源地址：@BiRd.GzNuOsHuI.cOm@||@BiRd.HuLiAnDaI.cOm@||
火山引擎开发者社区是火山引擎打造的AI技术生态平台，聚焦Agent与大模型开发，提供豆包系列模型（图像/视频/视觉）、智能分析与会话工具，并配套评测集、动手实验室及行业案例库。社区通过技术沙龙、挑战赛等活动促进开发者成长，新用户可领50万Tokens权益，助力构建智能应用。
更多推荐
8
0- 0
已为社区贡献21条内容
所有评论(0)