一、核心特点
·多 Skills 架构:通过 Skills 扩展 Agent 能力(搜索、代码、数据分析等)。
·对话管理:支持多轮对话、上下文保持。
·插件化:支持自定义 Skills 开发。
·多渠道接入:支持 IM(企业微信 / 飞书)、Web、API 等。
二、架构图
┌─────────────────────────────────────────────────┐│ OpenClaw Agent │├─────────────────────────────────────────────────┤│ ┌───────────┐ ┌──────────┐ ┌──────────────┐ ││ │ Skill A │ │ Skill B │ │ Skill C │ ││ │ (搜索) │ │ (代码) │ │ (数据分析) │ ││ └─────┬─────┘ └────┬─────┘ └──────┬───────┘ ││ └──────────────┴───────────────┘ ││ │ ││ ┌────────────────────▼───────────────────────┐ ││ │ 对话管理 & 记忆检索 │ ││ └────────────────────┬───────────────────────┘ │└───────────────────────┼─────────────────────────┘ │ ┌─────────────┼─────────────┐ ▼ ▼ ▼ ┌────────────┐ ┌──────────┐ ┌──────────────┐ │ 腾讯云 PG │ │ 大模型 │ │ 外部服务 │ │(记忆/向量) │ │ (推理) │ │ (API) │ └────────────┘ └──────────┘ └──────────────┘
三、前提条件
·腾讯云数据库 PostgreSQL 实例。
·Python 3.10+。
·已安装 pgvector 和 tencentdb_ai 扩展。
四、数据库初始化
— 创建 Agent 应用所需的 schemaCREATE SCHEMA IF NOT EXISTS agent_app;– 对话历史表CREATE TABLE agent_app.conversations ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), user_id TEXT NOT NULL, agent_id TEXT NOT NULL, messages JSONB[] DEFAULT '{}', created_at TIMESTAMPTZ DEFAULT NOW(), updated_at TIMESTAMPTZ DEFAULT NOW());– 知识库表CREATE TABLE agent_app.knowledge ( id BIGSERIAL PRIMARY KEY, title TEXT, content TEXT NOT NULL, embedding vector(1024), source TEXT, metadata JSONB DEFAULT '{}', created_at TIMESTAMPTZ DEFAULT NOW());CREATE INDEX idx_knowledge_embedding ON agent_app.knowledgeUSING hnsw (embedding vector_cosine_ops);– 长期记忆表CREATE TABLE agent_app.memories ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), user_id TEXT NOT NULL, content TEXT NOT NULL, embedding vector(1024), importance FLOAT DEFAULT 0.5, created_at TIMESTAMPTZ DEFAULT NOW());CREATE INDEX idx_memories_embedding ON agent_app.memoriesUSING hnsw (embedding vector_cosine_ops);
五、集成开发
以下示例的大模型名 <your-llm-model> 仅为占位;请替换为实际开通的模型,如腾讯混元(hunyuan-*)等。
字符串中的换行 \\n 在 PostgreSQL 标准字符串里是字面字符;如需真换行,建议使用 E'\\n'(转义字符串)或 chr(10)。
六、连接配置
import psycopg2from pgvector.psycopg2 import register_vector# 连接腾讯云 PostgreSQLconn = psycopg2.connect( host='your-instance.pg.tencentcdb.com', port=5432, dbname='agent_db', user='your_user', password='your_password' # 生产环境请通过环境变量 / 密钥管理服务注入)register_vector(conn)
七、记忆存储 Skill
class MemorySkill: """基于 PostgreSQL 的长期记忆 Skill""" def store_memory(self, user_id: str, content: str): """存储一条记忆""" with conn.cursor() as cur: # 使用参数化查询,禁止字符串拼接 cur.execute(""" INSERT INTO agent_app.memories (user_id, content, embedding) VALUES (%s, %s, (SELECT tencentdb_ai.get_embedding('bge-m3', %s)) ) """, (user_id, content, content)) conn.commit() def recall_memories(self, user_id: str, query: str, limit: int = 5): """检索相关记忆""" with conn.cursor() as cur: cur.execute(""" SELECT content, importance, 1 – (embedding <=> (SELECT tencentdb_ai.get_embedding('bge-m3', %s))) AS relevance FROM agent_app.memories WHERE user_id = %s ORDER BY embedding <=> (SELECT tencentdb_ai.get_embedding('bge-m3', %s)) LIMIT %s """, (query, user_id, query, limit)) return cur.fetchall()
八、RAG 知识检索 Skill
class RAGSkill: """基于 PostgreSQL 的 RAG 检索 Skill""" def search_knowledge(self, query: str, limit: int = 3): """语义搜索知识库""" with conn.cursor() as cur: cur.execute(""" SELECT title, content, 1 – (embedding <=> (SELECT tencentdb_ai.get_embedding('bge-m3', %s))) AS score FROM agent_app.knowledge ORDER BY embedding <=> (SELECT tencentdb_ai.get_embedding('bge-m3', %s)) LIMIT %s """, (query, query, limit)) return cur.fetchall() def rag_answer(self, query: str): """RAG: 检索 + 生成""" # 1. 检索相关文档 docs = self.search_knowledge(query) context = '\\n'.join([f'- {d[1]}' for d in docs]) # 2. 调用大模型生成回答(使用参数化查询) with conn.cursor() as cur: cur.execute(""" SELECT tencentdb_ai.chat_completions(%s, '基于以下参考资料回答用户问题。' || E'\\n参考资料:\\n' || %s || E'\\n\\n用户问题:' || %s ) """, ('<your-llm-model>', context, query)) return cur.fetchone()[0]
九、企业微信对接
# 企业微信消息回调处理def handle_wecom_message(user_id: str, message: str): # 1. 检索用户历史记忆 memories = memory_skill.recall_memories(user_id, message) # 2. 构建上下文 context = '\\n'.join([f'[记忆] {m[0]}' for m in memories]) # 3. RAG 检索知识 knowledge = rag_skill.search_knowledge(message) # 4. 生成回答 answer = generate_answer(message, context, knowledge) # 5. 存储本次交互为新记忆 memory_skill.store_memory(user_id, f'用户问: {message},回答: {answer[:100]}') return answer
十、相关文档
·Agent 长期记忆功能概述。
·多 Agent 共享记忆与项目协作。
·tencentdb_ai 插件功能介绍。