1. 理解 Agentic AI 时代数据库角色的根本性转变

在传统的软件架构中，数据库的角色相对清晰且被动：它是一个可靠、持久、用于存储和查询结构化数据的系统。开发者的核心工作是设计表结构、编写高效的 SQL、优化索引和事务。然而，随着以大型语言模型（LLM）为核心的 Agentic AI 的兴起，这种角色定位正在发生根本性的变化。Agentic AI 不再是简单的问答或内容生成工具，而是能够自主感知、规划、决策和执行的智能体。它需要与数据世界进行深度、动态的交互，这使得数据库从后台的“数据仓库”转变为前台的“智能决策引擎”。

这种转变的核心驱动力在于 Agentic AI 运作的两大支柱： 上下文（Context） 和 记忆（Memory） 。

  • 上下文 ：指 AI 在执行特定任务时，需要实时获取和理解的相关信息。这不仅仅是几条用户记录，而是可能跨越多个数据源、包含历史交互、业务规则和实时状态的综合信息视图。数据库需要能够快速、灵活地为 AI 提供这些上下文。
  • 记忆 ：指 AI 在长期交互中需要保留和回溯的信息，例如用户的长期偏好、历史对话、任务执行结果等。这要求数据库不仅能存储数据，还要能管理这些“记忆”的版本、关联性和有效性。

因此，现代数据库在 AI 时代的新使命是：不仅要“记住”数据，更要“理解”数据，并能够主动“调度”数据，为 AI 提供决策支持。这直接定义了 AI 智能的边界——AI 的能力上限，很大程度上取决于它能访问和处理的数据的广度、深度和实时性。

2. 构建支持 Agentic AI 的数据层：核心能力与架构选择

要为 Agentic AI 构建坚实的数据底座，我们需要关注几个超越传统 CRUD 的关键能力。这些能力决定了你的 AI 应用是停留在演示阶段，还是能够稳定处理复杂、真实的业务场景。

2.1 向量检索：从关键词匹配到语义理解

传统数据库依赖精确匹配或模糊查询，而 AI 更需要理解语义。向量数据库（或支持向量检索的数据库）通过将文本、图像等非结构化数据转换为高维向量（嵌入），并计算向量间的相似度，实现了基于语义的搜索。

关键考量：

  • 精度与召回率 ：不同的索引算法（如 HNSW, IVF）在速度和准确性上有权衡。
  • 过滤与混合搜索 ：如何在进行向量检索的同时，高效地结合结构化过滤条件（如时间范围、类别）。
  • 嵌入模型管理 ：向量质量取决于嵌入模型，数据库是否需要集成或管理模型版本。

示例：使用 PostgreSQL 的 pgvector 扩展 

-- 启用扩展并创建带向量类型的表
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE documents (
    id BIGSERIAL PRIMARY KEY,
    content TEXT,
    embedding vector(1536), -- 例如 OpenAI text-embedding-3-small 的维度
    metadata JSONB,
    created_at TIMESTAMP DEFAULT NOW()
);

-- 创建向量索引以加速相似性搜索
CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);

-- 语义搜索：查找与查询向量最相似的文档
SELECT id, content, metadata,
       1 - (embedding <=> '[0.1, 0.2, ...]') AS similarity -- <=> 计算余弦距离
FROM documents
ORDER BY embedding <=> '[0.1, 0.2, ...]' -- 查询向量
LIMIT 10;

-- 混合搜索：结合语义和元数据过滤
SELECT id, content,
       1 - (embedding <=> '[0.1, 0.2, ...]') AS similarity
FROM documents
WHERE metadata->>'category' = 'technical_blog'
ORDER BY embedding <=> '[0.1, 0.2, ...]'
LIMIT 10;

2.2 长上下文与记忆管理

Agent 的记忆不能是无限堆砌的聊天记录。有效的记忆管理需要：

  1. 结构化存储 ：将记忆分类（如用户事实、会话历史、工具调用结果），并用结构化字段（如重要性分数、访问时间、关联实体）存储。
  2. 摘要与压缩 ：对长对话或复杂结果进行总结，存储摘要而非全文，以节省上下文窗口。
  3. 相关性检索 ：根据当前对话或任务，动态从记忆库中检索最相关的片段注入上下文。

示例：设计一个记忆表 

CREATE TABLE agent_memories (
    memory_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    session_id VARCHAR(255) NOT NULL,
    user_id VARCHAR(255),
    entity_type VARCHAR(50), -- 如 ‘user_preference‘， ’conversation‘， ’task_result‘
    entity_id VARCHAR(255),
    content TEXT, -- 原始内容或摘要
    embedding vector(1536), -- 用于基于当前对话检索相关记忆
    importance_score FLOAT DEFAULT 1.0,
    access_count INT DEFAULT 0,
    last_accessed_at TIMESTAMP,
    created_at TIMESTAMP DEFAULT NOW()
);

-- 为记忆内容创建向量索引
CREATE INDEX ON agent_memories USING ivfflat (embedding vector_cosine_ops);

2.3 与 AI 框架的无缝集成：MCP 协议

模型上下文协议（Model Context Protocol, MCP）正在成为连接 AI 应用（如 Claude Desktop, Cursor）与各种数据源和工具的标准。数据库支持 MCP 服务器，意味着开发者可以直接在 AI 开发环境中通过自然语言查询和操作数据库。

核心价值 ：

  • 降低开发门槛 ：数据分析师、产品经理可以直接通过对话查询数据，无需编写复杂 SQL。
  • 加速原型验证 ：在 AI IDE 中快速验证数据获取逻辑。
  • 统一访问层 ：通过 MCP 标准化对各类数据库（PostgreSQL, MySQL, 甚至 S3）的访问。

架构示意 ：

[AI 应用如 Cursor] <--(MCP 协议)--> [MCP 服务器] <--(数据库驱动)--> [你的数据库]

目前，Amazon Aurora、DynamoDB 等已宣布支持 MCP。对于自建数据库，可以考虑使用开源的 MCP 服务器实现来包装你的数据访问层。

2.4 架构路径选择：增强传统库 vs. 原生 AI 数据库

面对 Agentic AI 的需求，技术选型上通常有两条路径：

考量维度 路径一：增强现有传统数据库 路径二：采用原生 AI 数据库/数据平台
核心思路 在已有 PostgreSQL、MySQL 等基础上，通过扩展（如 pgvector ）或旁路系统增加 AI 能力。 使用为 AI 场景从头设计的数据库，如某些云托管的向量数据库或 AI 原生数据平台。
优点 技术栈统一 ，复用现有运维知识、工具链和备份恢复流程。
事务一致性 有保障，适合强一致要求的业务数据。
混合查询 能力强，可轻松关联向量和结构化数据。		 性能优化 ，专为向量操作设计，吞吐量和延迟可能更优。
开箱即用 ，通常集成了更丰富的 AI 工作流（如嵌入管道、多模态）。
Serverless 弹性 更好，更适合不可预测的 AI 工作负载。
缺点 性能瓶颈 ，向量检索可能不如专用系统。
运维复杂度 增加，需要管理扩展和可能的分片策略。
功能可能滞后 于专用系统。		 生态锁定 风险，可能依赖特定云厂商或供应商。
数据同步 成本，如需与核心业务数据库保持同步，需额外管道。
学习成本 ，需要掌握新的查询语言或 API。
适用场景 已有成熟关系型数据库，AI 功能作为补充；对事务一致性要求高；团队熟悉现有技术栈。 全新 AI 项目；对向量检索性能要求极高；希望最小化运维负担，快速上线。

决策建议 ：

  • 传统商业数据库（如 Oracle, SQL Server）用户 ：考虑向开源生态（如 PostgreSQL/MySQL）迁移，以降低成本和获得更活跃的 AI 生态支持。云服务商通常提供便捷的迁移工具。
  • 封装应用（如 ERP）用户 ：不必急于重构底层数据库。优先考虑通过 API 或变更数据捕获（CDC）技术，将现有数据价值释放给 AI 层，进行渐进式改造。
  • 全新项目 ：建议直接拥抱开源技术栈。关系型数据用 PostgreSQL/MySQL，数据湖用 Apache Iceberg 等开放格式。这为未来提供了最大的灵活性和成本优势。

3. 实战：搭建一个支持 Agentic AI 的简易数据层

让我们通过一个具体的例子，搭建一个能为技术问答 Agent 提供支持的后端数据服务。这个 Agent 能回答关于内部技术文档的问题。

3.1 环境准备与依赖

假设我们选择 PostgreSQL + pgvector 作为核心，使用 Python 的 FastAPI 构建服务。

环境要求：

  • PostgreSQL 15+
  • Python 3.10+

项目依赖 ( requirements.txt ): 

fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy==2.0.23
psycopg2-binary==2.9.9
pgvector==0.2.0
openai==1.6.1
langchain==0.1.0
python-dotenv==1.0.0

3.2 数据库 schema 设计与初始化

首先，设计存储技术文档和对话记忆的表。

数据库初始化脚本 ( init_db.sql ): 

-- 启用必要的扩展
CREATE EXTENSION IF NOT EXISTS vector;

-- 技术文档表
CREATE TABLE technical_docs (
    id BIGSERIAL PRIMARY KEY,
    doc_id VARCHAR(255) UNIQUE NOT NULL, -- 外部文档ID
    title TEXT NOT NULL,
    content TEXT NOT NULL,
    content_embedding vector(1536), -- 存储内容向量
    metadata JSONB DEFAULT '{}', -- 存储来源、作者、标签等信息
    chunk_index INT, -- 如果文档被分块，这是块索引
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

-- 为向量搜索创建索引（根据数据量选择 lists 参数）
CREATE INDEX idx_docs_embedding ON technical_docs USING ivfflat (content_embedding vector_cosine_ops) WITH (lists = 100);

-- 对话记忆表（简化版）
CREATE TABLE conversation_memories (
    id BIGSERIAL PRIMARY KEY,
    session_id VARCHAR(255) NOT NULL,
    query TEXT, -- 用户问题
    query_embedding vector(1536), -- 问题的向量，用于寻找相似历史
    answer TEXT, -- Agent 的回答
    retrieved_doc_ids BIGINT[], -- 本次回答引用的文档ID数组
    created_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_memories_session ON conversation_memories(session_id);
CREATE INDEX idx_memories_query_embedding ON conversation_memories USING ivfflat (query_embedding vector_cosine_ops);

3.3 核心服务层实现

我们创建主要的服务文件，包含文档嵌入、向量检索和记忆管理功能。

主服务文件 ( app/main.py ): 

import os
from typing import List, Optional
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
import openai
import numpy as np

# 加载环境变量，如数据库连接串和 OpenAI API Key
load_dotenv()

app = FastAPI(title="TechDoc AI Agent Backend")

# 数据库连接配置
DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://user:password@localhost/agentic_db")
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# OpenAI 客户端初始化（用于生成嵌入）
openai.api_key = os.getenv("OPENAI_API_KEY")
EMBEDDING_MODEL = "text-embedding-3-small"

# 数据模型
class DocumentChunk(BaseModel):
    doc_id: str
    title: str
    content: str
    metadata: Optional[dict] = {}

class QueryRequest(BaseModel):
    session_id: str
    question: str
    top_k: int = 5  # 返回最相关的文档数量

class QueryResponse(BaseModel):
    answer: str
    supporting_docs: List[dict]  # 包含标题和片段
    session_id: str

def get_embedding(text: str) -> List[float]:
    """调用 OpenAI API 获取文本的向量表示"""
    try:
        response = openai.embeddings.create(model=EMBEDDING_MODEL, input=text)
        return response.data[0].embedding
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to get embedding: {str(e)}")

@app.post("/ingest_doc")
async def ingest_document(chunk: DocumentChunk):
    """接收文档块，生成向量并存入数据库"""
    embedding = get_embedding(chunk.content)
    embedding_array = np.array(embedding).tolist()  # 转换为列表

    with SessionLocal() as db:
        # 使用 SQLAlchemy core 或 raw SQL 插入，因为 pgvector 类型需要适配
        insert_stmt = text("""
            INSERT INTO technical_docs (doc_id, title, content, content_embedding, metadata)
            VALUES (:doc_id, :title, :content, :embedding, :metadata)
            ON CONFLICT (doc_id) DO UPDATE SET
                title = EXCLUDED.title,
                content = EXCLUDED.content,
                content_embedding = EXCLUDED.content_embedding,
                metadata = EXCLUDED.metadata,
                updated_at = NOW()
        """)
        db.execute(insert_stmt, {
            "doc_id": chunk.doc_id,
            "title": chunk.title,
            "content": chunk.content,
            "embedding": embedding_array,
            "metadata": chunk.metadata or {}
        })
        db.commit()
    return {"message": f"Document '{chunk.doc_id}' ingested successfully."}

@app.post("/query", response_model=QueryResponse)
async def query_documents(request: QueryRequest):
    """处理用户查询，检索相关文档并生成回答（简化版，实际需集成LLM）"""
    query_embedding = get_embedding(request.question)

    # 1. 向量检索：从技术文档中查找最相关内容
    with SessionLocal() as db:
        # 使用 pgvector 的 <=> 运算符进行余弦相似度搜索
        search_sql = text("""
            SELECT id, title, content,
                   1 - (content_embedding <=> :embedding) as similarity
            FROM technical_docs
            ORDER BY content_embedding <=> :embedding
            LIMIT :top_k
        """)
        result = db.execute(search_sql, {
            "embedding": query_embedding,
            "top_k": request.top_k
        })
        relevant_docs = result.fetchall()

    if not relevant_docs:
        raise HTTPException(status_code=404, detail="No relevant documents found.")

    # 2. 构建上下文（这里简化处理，实际应使用更复杂的提示工程）
    context_parts = []
    retrieved_ids = []
    for doc in relevant_docs:
        context_parts.append(f"Title: {doc.title}\nSnippet: {doc.content[:500]}...")  # 截取片段
        retrieved_ids.append(doc.id)

    context = "\n\n---\n\n".join(context_parts)

    # 3. 调用 LLM 生成答案（此处为模拟，实际应调用如 OpenAI GPT、Claude 等）
    # 提示词示例：`基于以下上下文，请回答问题：{request.question}\n\n上下文：{context}`
    simulated_answer = f"根据检索到的 {len(relevant_docs)} 篇相关文档，您的问题‘{request.question}’的核心要点如下：[此处应是LLM生成的总结性回答]。"
    # 实际项目中，这里应替换为真实的 LLM 调用代码。

    # 4. 保存本次交互到记忆库
    with SessionLocal() as db:
        memory_sql = text("""
            INSERT INTO conversation_memories (session_id, query, query_embedding, answer, retrieved_doc_ids)
            VALUES (:session_id, :query, :query_embedding, :answer, :retrieved_doc_ids)
        """)
        db.execute(memory_sql, {
            "session_id": request.session_id,
            "query": request.question,
            "query_embedding": query_embedding,
            "answer": simulated_answer,
            "retrieved_doc_ids": retrieved_ids
        })
        db.commit()

    # 5. 构造返回结果
    supporting_docs = [{"id": doc.id, "title": doc.title, "similarity": round(doc.similarity, 4)} for doc in relevant_docs]
    return QueryResponse(
        answer=simulated_answer,
        supporting_docs=supporting_docs,
        session_id=request.session_id
    )

@app.get("/memory/{session_id}")
async def get_conversation_history(session_id: str, limit: int = 10):
    """获取指定会话的历史记录"""
    with SessionLocal() as db:
        sql = text("SELECT query, answer, created_at FROM conversation_memories WHERE session_id = :sid ORDER BY created_at DESC LIMIT :limit")
        result = db.execute(sql, {"sid": session_id, "limit": limit})
        history = result.fetchall()
    return {"session_id": session_id, "history": [dict(row) for row in history]}

3.4 运行与验证

  1. 启动服务： 

    uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

  2. 注入文档： 使用 curl 或 Postman 调用 /ingest_doc 接口。

    curl -X POST "http://localhost:8000/ingest_doc" \
-H "Content-Type: application/json" \
-d '{
    "doc_id": "doc_001",
    "title": "如何配置 PostgreSQL 高可用",
    "content": "PostgreSQL 高可用方案通常涉及流复制、自动故障转移和负载均衡...",
    "metadata": {"category": "database", "author": "admin"}
}'

  3. 进行查询： 调用 /query 接口模拟用户提问。

    curl -X POST "http://localhost:8000/query" \
-H "Content-Type: application/json" \
-d '{
    "session_id": "user_123_session",
    "question": "PostgreSQL 的高可用怎么实现？",
    "top_k": 3
}'

    预期返回会包含模拟的答案和检索到的相关文档列表。

  4. 查看记忆： 调用 /memory/{session_id} 查看该会话的历史。

    curl "http://localhost:8000/memory/user_123_session"

4. 生产环境关键考量与最佳实践

上述示例是一个简化模型。要将它用于生产环境，必须解决以下问题：

4.1 性能、扩展性与 Serverless

Agentic AI 的负载往往是突发和不可预测的。一个成功的问答可能触发数十次向量检索和 LLM 调用。

  • 向量索引优化 ： pgvector 的 ivfflat 索引需要足够的数据进行训练（ CREATE INDEX 前应有代表性数据）。对于十亿级向量，需要考虑 hnsw 索引或专用向量数据库。
  • 查询优化 ：避免在循环中进行向量检索。批量处理查询，并利用数据库连接池。
  • 拥抱 Serverless ：对于负载波动大的场景，优先考虑 Serverless 数据库（如 Amazon Aurora Serverless, DynamoDB On-Demand）。它们可以真正做到“按需伸缩，用量计费”，在空闲时成本为零。这是构建经济高效 AI 应用的关键。
  • 缓存策略 ：对频繁查询的相似问题或通用知识片段引入缓存（如 Redis），减少对数据库和 LLM 的重复调用。

4.2 安全、治理与数据沿袭

将企业数据暴露给 AI 模型，安全是首要问题。

  • 权限最小化 ：AI Agent 使用的数据库账号应只有读取特定表和视图的权限，绝不能拥有 DROP 或 UPDATE 关键数据的权限。
  • 数据脱敏与过滤 ：在数据注入向量库前，必须进行脱敏处理（如替换身份证号、手机号）。在检索层，根据用户角色动态添加数据过滤条件（如 WHERE department_id = :user_dept ）。
  • 审计与日志 ：记录所有 AI 发起的查询请求、检索到的数据 ID 以及生成的回答，便于溯源和合规审查。
  • 统一数据治理 ：确保 AI 访问的数据与数据中台、主数据管理（MDM）的定义一致，避免出现“多个真相来源”。

4.3 可观测性与排错

当 AI 回答出现错误或“幻觉”时，你需要快速定位是数据问题、检索问题还是 LLM 问题。

  • 结构化日志 ：记录每次调用的 session_id , query , retrieved_doc_ids , similarity_scores , model_used , token_usage , response_time 。
  • 检索质量监控 ：定期用标准问题集测试，监控检索结果的相关性（Recall@K）和准确性。
  • 链路追踪 ：集成 OpenTelemetry 等工具，追踪一个用户请求从入口到数据库检索、LLM 调用、最终响应的完整链路，便于分析延迟瓶颈。
  • 常见问题排查清单 ：
问题现象 可能原因 检查步骤
AI 回答“未找到相关信息” 1. 查询向量与文档向量不匹配。
2. 相关文档未入库或嵌入失败。
3. 向量索引未正确构建或需要重建。		 1. 检查查询文本的嵌入模型是否与文档嵌入模型一致。
2. 查询数据库，确认目标文档是否存在且 content_embedding 字段非空。
3. 对少量已知文档执行精确 ID 查询和向量相似度查询，对比结果。
检索速度慢 1. 向量索引类型不适合数据规模。
2. 数据库资源（CPU/IO）不足。
3. 查询并发过高。		 1. 使用 EXPLAIN ANALYZE 分析向量查询计划。
2. 监控数据库性能指标（CPU、内存、磁盘 IOPS）。
3. 考虑升级索引（如从 ivfflat 到 hnsw ）或迁移到专用向量数据库。
AI 回答包含过时信息 1. 源数据已更新，但向量库未同步。
2. 记忆库中保留了过时的上下文。		 1. 建立 CDC（变更数据捕获）管道，确保源数据变更能实时或准实时同步到向量库。
2. 为记忆实现 TTL（生存时间）或基于重要性的遗忘机制。
安全性担忧：AI 访问了不该看的数据 1. 数据库账号权限过大。
2. 检索层未实施行级安全过滤。		 1. 审查 AI Agent 使用的数据库角色的权限。
2. 在检索 SQL 中强制加入基于用户上下文的过滤条件。

5. 演进方向：从数据平台到 AI 原生数据智能

最终，支持 Agentic AI 的数据库将演变为一个 AI 原生数据智能平台 。它不仅仅是存储和检索，而是能够：

  • 主动编排数据工作流 ：根据 Agent 的目标，自动从多个数据源（数据库、数据湖、API）拉取、清洗、关联数据，构建上下文。
  • 提供推理能力 ：内置轻量级推理模型，能在数据库内部完成一些简单的数据判断和摘要，减少与外部 LLM 的往返。
  • 持续学习与优化 ：根据 Agent 任务的成功率反馈，自动调整检索策略、向量化模型或数据预处理管道。

对于开发者和架构师而言，当下的任务不是等待一个完美的“AI 数据库”出现，而是开始行动：评估现有数据资产，选择一条务实的技术路径（增强或新建），构建一个可迭代、可观测的数据服务层。真正的竞争优势，将来自于你利用私有数据训练和滋养 AI 的深度与速度，而一个设计良好的数据层，正是这场竞赛的核心引擎。

# 数据库架构
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

终极跨平台串口调试工具：SerialPortAssistant完整使用指南

在嵌入式开发、物联网设备调试和工业控制领域，串口通信调试是每个工程师必备的技能。然而，面对不同操作系统、设备平台和开发环境，寻找一款真正好用且免费的串口调试工具并不容易。今天，我要为你介绍一款功能强大的开源串口调试工具——SerialPortAssistant，这款跨平台串口助手能够完美运行在Windows、Linux、Android和macOS四大主流系统上，让你在任何开发环境中都能得心应手。

avatar AI Agent技术社区

Roboto字体：你的数字设计难题，Google的解决方案

你是否曾为移动应用界面字体选择而烦恼？或者在网页设计中寻找既现代又易读的字体？也许你正在为多语言项目寻找一款能覆盖拉丁、西里尔和希腊字符的字体。这些设计难题，Google的设计团队早已预见，并给出了一个优雅的解决方案：Roboto字体。作为Android系统和Material Design的默认字体，Roboto不仅仅是Google的视觉语言代表，更是一款经过精心设计的现代无衬线字体。它解决了

avatar AI Agent技术社区

Tabby终端终极指南：如何快速提升开发效率300%

Tabby是一款面向现代开发者的高度可配置终端模拟器，集成了SSH、Telnet和串行客户端功能，支持Windows、macOS和Linux三大平台。这款免费开源的终端工具通过智能标签管理、强大的快捷键系统和丰富的插件生态，能够显著提升开发者的工作效率。在前100个字内，我们重点介绍Tabby终端的核心功能：它是一个现代化的终端解决方案，提供了多标签界面、SSH连接管理、主题定制和插件扩展等特性，

avatar AI Agent技术社区