基于 OpenClaw 构建 AI Agent 应用

┌─────────────────────────────────────────────────┐
│                  OpenClaw Agent                 │
├─────────────────────────────────────────────────┤
│  ┌───────────┐  ┌──────────┐  ┌──────────────┐  │
│  │ Skill A   │  │ Skill B  │  │ Skill C      │  │
│  │ (搜索)    │  │ (代码)   │  │ (数据分析)   │  │
│  └─────┬─────┘  └────┬─────┘  └──────┬───────┘  │
│        └──────────────┴───────────────┘         │
│                       │                         │
│  ┌────────────────────▼───────────────────────┐ │
│  │          对话管理 & 记忆检索               │ │
│  └────────────────────┬───────────────────────┘ │
└───────────────────────┼─────────────────────────┘
                        │
          ┌─────────────┼─────────────┐
          ▼             ▼             ▼
   ┌────────────┐ ┌──────────┐ ┌──────────────┐
   │ 腾讯云 PG  │ │ 大模型   │ │ 外部服务     │
   │(记忆/向量) │ │ (推理)   │ │ (API)        │
   └────────────┘ └──────────┘ └──────────────┘

-- 创建 Agent 应用所需的 schema
CREATE 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.knowledge
USING 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.memories
USING hnsw (embedding vector_cosine_ops);

import psycopg2
from pgvector.psycopg2 import register_vector

# 连接腾讯云 PostgreSQL
conn = psycopg2.connect(
    host='your-instance.pg.tencentcdb.com',
    port=5432,
    dbname='agent_db',
    user='your_user',
    password='your_password'  # 生产环境请通过环境变量 / 密钥管理服务注入
)
register_vector(conn)

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()

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