做一个 RAG 知识助理

让用户上传 PDF / Markdown 给 Agent，Agent 用 RAG (检索增强生成) 回答问题。 关键设计：用户文档明文不进 Hashee 后端；embedding 与原文都存在 Agent 自己的后端（用户授权后访问）。

用户故事

“我想做一个产品助理，每个用户能给它喂自己的产品手册 PDF。Agent 用 RAG 答问。我们对用户的隐私承诺：原文存在用户自己的 grant-scoped store，撤销关系后立即清理。“

架构

用户上传文档（通过 Hashee app 发文件给 Agent）
    │
    ▼ Hashee Layer 1 加密 + 上传 R2
    │
    ▼ Agent SDK 收 file 消息
    │
    ├─ agent.downloadAttachment() 拿明文 PDF
    ├─ PDF → 切块 → embedding
    ├─ 存到 Agent 自己的 PG (per user table 分区)
    │     └─ 索引 vector + 原文 chunk
    │
    ▼ 用户问问题
    │
    ├─ 问题 embedding → 检索 top-k chunks
    ├─ 构造 prompt: system + retrieved + user_question
    ├─ LLM 调用
    ├─ 回复时带"参考来源"卡片（artifact 引用原文位置）
    │
    ▼ 用户撤销关系
    │
    └─ relation.revoked → 清理该 user 的所有 embedding + chunk

完整代码

import { HasheeAgent } from "@hasheeai/agent-sdk-ts";
import OpenAI from "openai";
import { Pool } from "pg";
import { extractText } from "@/lib/pdf";   // 你的 pdf-parse 封装

const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });
const db = new Pool({ connectionString: process.env.DATABASE_URL });

const agent = await HasheeAgent.init({
  agentId: process.env.HASHEE_AGENT_ID!,
  token: process.env.HASHEE_AGENT_TOKEN!,
  baseUrl: "https://api.hashee.ai",
  connectionMode: "websocket",
});

// === 接收文件 → 入库 ===

agent.addMessageHandler(async (msg) => {
  const payload = msg.payload;
  if (payload?.type === "file" && payload.mime === "application/pdf") {
    await ingestPdf(msg, payload);
    return;
  }
  if (payload?.type === "text") {
    await answerQuestion(msg, payload.text);
    return;
  }
});

async function ingestPdf(msg: any, payload: any) {
  await agent.send(msg.conversation_id, {
    type: "text",
    text: `📥 收到 ${payload.filename}（${formatBytes(payload.size)}），正在建索引...`,
  });

  // 1. 下载并解密
  const bytes = await agent.downloadAttachment({
    attachment_id: payload.attachment_id,
    conversation_id: msg.conversation_id,
  });

  // 2. 解析文本
  const text = await extractText(bytes);

  // 3. 切块（朴素：每 ~500 字符一块，重叠 100）
  const chunks = chunkText(text, 500, 100);

  // 4. embedding
  const embeds = await openai.embeddings.create({
    model: "text-embedding-3-small",
    input: chunks,
  });

  // 5. 入库（per user 分区）
  await db.query("BEGIN");
  try {
    for (let i = 0; i < chunks.length; i++) {
      await db.query(
        `INSERT INTO doc_chunks (user_id, source_filename, chunk_idx, text, embedding)
         VALUES ($1, $2, $3, $4, $5)`,
        [msg.sender_id, payload.filename, i, chunks[i], embeds.data[i].embedding],
      );
    }
    await db.query("COMMIT");
  } catch (e) {
    await db.query("ROLLBACK");
    throw e;
  }

  await agent.send(msg.conversation_id, {
    type: "text",
    text: `✓ 已索引 ${chunks.length} 个段落。直接问问题就好。`,
  });
}

// === 回答问题 ===

async function answerQuestion(msg: any, question: string) {
  await agent.typing(msg.conversation_id);

  // 1. embed 问题
  const qEmbed = await openai.embeddings.create({
    model: "text-embedding-3-small",
    input: question,
  });

  // 2. 检索 top-5（pgvector）
  const r = await db.query(
    `SELECT chunk_idx, source_filename, text,
            (embedding <=> $1) AS distance
     FROM doc_chunks
     WHERE user_id = $2
     ORDER BY embedding <=> $1
     LIMIT 5`,
    [qEmbed.data[0].embedding, msg.sender_id],
  );
  if (r.rows.length === 0) {
    await agent.send(msg.conversation_id, {
      type: "text",
      text: "我还没有你的文档。先发个 PDF / Markdown 给我吧。",
    });
    return;
  }

  // 3. 构造 RAG prompt
  const context = r.rows.map((row, i) =>
    `[${i + 1}] (${row.source_filename}, 段 ${row.chunk_idx})\n${row.text}`,
  ).join("\n\n");

  const completion = await openai.chat.completions.create({
    model: "gpt-4o-mini",
    messages: [
      {
        role: "system",
        content: `你是知识助理。基于以下用户文档片段回答问题。
回答末尾用 [n] 标注引用来源。如果片段不足以回答，直说"信息不足"。

---
${context}
---`,
      },
      { role: "user", content: question },
    ],
  });

  const answer = completion.choices[0]?.message?.content ?? "(空回复)";

  // 4. 发送 + artifact 引用卡
  await agent.send(msg.conversation_id, { type: "text", text: answer });

  await agent.sendArtifact(msg.conversation_id, {
    artifact: {
      subtype: "info",
      title: "参考来源",
      payload: {
        body: r.rows.map((row, i) =>
          `**[${i + 1}]** ${row.source_filename}（段 ${row.chunk_idx}）\n> ${row.text.slice(0, 150)}...`,
        ).join("\n\n"),
      },
    },
  });
}

// === 关系撤销 → 清理 ===

agent.addEventHandler(async (event) => {
  if (event.type !== "relation.revoked") return;
  const userId = event.payload.user_id;
  await db.query("DELETE FROM doc_chunks WHERE user_id = $1", [userId]);
  console.log(`[cleanup] removed all chunks for user ${userId}`);
});

// === 工具 ===

function chunkText(text: string, size: number, overlap: number): string[] {
  const chunks: string[] = [];
  for (let i = 0; i < text.length; i += size - overlap) {
    chunks.push(text.slice(i, i + size));
  }
  return chunks;
}

function formatBytes(n: number): string {
  if (n < 1024) return `${n} B`;
  if (n < 1024 * 1024) return `${(n / 1024).toFixed(1)} KB`;
  return `${(n / 1024 / 1024).toFixed(1)} MB`;
}

agent.addStatusHandler((s) => console.log(`[status] ${s}`));
console.log("[rag-bot] up");

DB schema (PG + pgvector)

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE doc_chunks (
  id SERIAL PRIMARY KEY,
  user_id TEXT NOT NULL,
  source_filename TEXT NOT NULL,
  chunk_idx INT NOT NULL,
  text TEXT NOT NULL,
  embedding vector(1536) NOT NULL,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX ON doc_chunks(user_id);
CREATE INDEX ON doc_chunks USING ivfflat (embedding vector_cosine_ops);

隐私与数据主权

• ✅ 文档明文只在 Agent 自己的 PG 里（不在 Hashee 后端）
• ✅ Hashee 后端只看到密文 + R2 对象引用
• ✅ 用户撤销 H2A → relation.revoked → 自动 DELETE 该 user 的所有 chunks
• ✅ 提供 export tool（业务侧实现 GET /export/:user_id 返回 user 所有原文 + embeddings）
• ✅ 隐私政策明示：“你的文档明文只在我们 Agent 的服务器，不进 Hashee 后端”

升级路径

想做	改哪
跨用户共享某些公开文档	加 public BOOLEAN；检索时 WHERE user_id = $1 OR public = true
Embedding model 升级	双写 embedding → 灰度切换
接 Grant API（V2）	走 agent.readUserData({ scope: "knowledge:read" }) 拿用户授权的 Hashee 历史消息
Hybrid search（vector + BM25）	加 tsvector 列；rerank 用 cross-encoder
大文档支持（>100MB）	Hashee 单文件 100MB 限制；让用户分多个 PDF 上传

部署

VPS / 容器（推荐）：见 自托管 VPS。 Cloudflare Workers：需要外部 PG（如 Neon），见 部署到 Cloudflare Workers。

相关页面

• 文件上传 — downloadAttachment 详解
• Data Grants — V2 grant API
• 更新与撤销 — GDPR 合规清理
• Production Checklist