云数据库 PostgreSQL 基于 OpenClaw 构建 AI Agent 应用

OpenClaw 是一个开源的 AI 助理框架,支持快速构建企业级 AI Agent 应用。结合腾讯云数据库 PostgreSQL 作为记忆和向量后端,可以实现具备长期记忆、知识检索、多轮对话的智能 Agent。

相关产品
云数据库 PostgreSQL 产品页,含版本规格、价格与购买入口。

一、核心特点

·多 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 插件功能介绍。

赞 (0)
上一篇 4天前
下一篇 4天前