2672 words
13 minutes
AI-Interview 数据库主键迁移实践：UUID → BIGSERIAL
2026-04-10
AI-Interview
PostgreSQL
/
DB

AI-Interview 数据库主键迁移实践：UUID → BIGSERIAL#

一、项目背景与问题#

AI-Interview 是一个基于 LangGraph + LangChain 构建的 AI 模拟面试官系统。项目初期采用 UUID v4 作为所有表的主键设计，这一选择在单机开发阶段带来了便利——支持并行数据导入、无需担心分布式 ID 冲突。然而，随着数据量增长和生产环境部署，UUID 主键的缺陷逐渐显现。

UUID 主键的性能瓶颈#

UUID v4 作为主键存在三个核心问题：

存储开销巨大。UUID 占用 16 字节，而 BIGSERIAL 仅需 8 字节。在拥有数百万记录的场景下，主键索引体积差异可达两倍，直接影响内存缓存效率和磁盘 I/O 性能。

索引性能劣化。UUID 的随机性导致 B+tree 索引频繁分裂、页面稀疏。写入时新 UUID 随机落入索引各位置，无法利用顺序写入优化，而 BIGSERIAL 的自增特性使新记录始终追加到索引末端，顺序写入效率极高。

枚举类型不兼容。在实际业务中，面试状态、反馈类型等字段需要枚举类型，但 UUID 无法定义为枚举值的关联字段，导致类型系统与业务需求脱节。

二、解决方案与技术选型#

针对 UUID 的缺陷，我们设计了 BIGSERIAL 主键 + UUID 外部标识符的混合方案。

设计原则#

唯一 UUID 策略：仅 users 表保留 UUID 作为 API 外部标识符，其他表通过 BIGSERIAL 自增主键管理，外键统一使用 BIGINT 类型。

网关层转换：UUID ↔ BIGSERIAL 的转换由 API 网关层处理，应用层代码无需感知 UUID 存在，实现数据层与展示层的解耦。

架构对比#

表名迁移前迁移后
usersid: UUID (PK)id: BIGSERIAL (PK), uuid: UUID (API 用)
resumesid: UUID (PK), user_id: UUID (FK)id: BIGSERIAL (PK), user_id: BIGINT (FK)
projectsid: UUID (PK), resume_id: UUID (FK)id: BIGSERIAL (PK), resume_id: BIGINT (FK)
knowledge_baseid: UUID (PK), project_id: UUID (FK)id: BIGSERIAL (PK), project_id: BIGINT (FK)
interview_sessionsid: UUID (PK), user_id/resume_id: UUID (FK)id: BIGSERIAL (PK), user_id/resume_id: BIGINT (FK)
qa_historyid: UUID (PK), session_id: UUID (FK)id: BIGSERIAL (PK), session_id: BIGINT (FK)
interview_feedbackid: UUID (PK), session_id: UUID (FK)id: BIGSERIAL (PK), session_id: BIGINT (FK)

三、数据库架构设计#

ER 关系图#

┌──────────────────┐       ┌──────────────────┐
│      users       │       │      resumes     │
│ ──────────────── │◄─────│ ──────────────── │
│ id: BIGSERIAL(PK)│       │ id: BIGSERIAL(PK)│
│ uuid: UUID(API用)│       │ user_id: BIGINT  │
└──────────────────┘       └────────┬─────────┘
                                    │ resume_id
                                    ▼
┌─────────────────────────────────┐
│          projects               │
│ ─────────────────────────────── │
│ id: BIGSERIAL(PK)              │
│ resume_id: BIGINT(FK)          │
└────────┬────────────────────────┘
         │ project_id
         ▼
┌─────────────────────────────────┐
│       knowledge_base            │
│ ─────────────────────────────── │
│ id: BIGSERIAL(PK)              │
│ project_id: BIGINT(FK)          │
└─────────────────────────────────┘


┌─────────────────────────────────┐
│    interview_sessions           │
│ ─────────────────────────────── │
│ id: BIGSERIAL(PK)              │
│ user_id: BIGINT(FK)            │
│ resume_id: BIGINT(FK)          │
└────────┬────────────────────────┘
         │ session_id
         ▼
┌────────┴────────┐     ┌──────────────────┐
│   qa_history     │     │interview_feedback│
│ ────────────────│     │ ──────────────── │
│ id: BIGSERIAL   │     │ id: BIGSERIAL    │
│ session_id      │     │ session_id       │
└─────────────────┘     └──────────────────┘

四、关键代码实现#

SQLAlchemy 模型变更#

from sqlalchemy import BigInteger, Sequence, UUID
from sqlalchemy.orm import Mapped, mapped_column
import uuid


class User(Base):
    __tablename__ = "users"


    id: Mapped[int] = mapped_column(
        BigInteger,
        primary_key=True,
        default=Sequence('users_id_seq'),
        server_default=text("nextval('users_id_seq'::regclass)")
    )
    uuid: Mapped[UUID] = mapped_column(
        UUID(as_uuid=True),
        nullable=False,
        unique=True,
        default=uuid.uuid4
    )
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True)




class Resume(Base):
    __tablename__ = "resumes"


    id: Mapped[int] = mapped_column(
        BigInteger,
        primary_key=True,
        default=Sequence('resumes_id_seq')
    )
    user_id: Mapped[int] = mapped_column(
        BigInteger,
        ForeignKey('users.id'),
        nullable=False
    )
    # uuid 字段已移除，不再需要 UUID 作为主键

DAO 层查询方法#

class ResumeDAO:
    def __init__(self, session: AsyncSession):
        self.session = session


    # UUID 查询：用于 API 层接收外部 UUID 参数
    async def find_by_uuid(self, resume_uuid: UUID) -> Optional[Resume]:
        result = await self.session.execute(
            select(Resume).where(Resume.uuid == resume_uuid)
        )
        return result.scalar_one_or_none()


    # BIGINT 查询：用于应用层内部使用自增 ID
    async def find_by_id(self, resume_id: int) -> Optional[Resume]:
        result = await self.session.execute(
            select(Resume).where(Resume.id == resume_id)
        )
        return result.scalar_one_or_none()


    # 分页查询返回 BIGINT ID
    async def find_by_user_id(self, user_id: int, limit: int = 50) -> List[Resume]:
        result = await self.session.execute(
            select(Resume)
            .where(Resume.user_id == user_id)
            .order_by(Resume.id.desc())
            .limit(limit)
        )
        return list(result.scalars().all())

Agent 层调用示例#

resume_agent.py
class ResumeAgent:
    async def get_resume(self, resume_id: str) -> Optional[Resume]:
        # API 层接收字符串，转换为 UUID 查询
        try:
            resume_uuid = UUID(resume_id) if resume_id else None
            if resume_uuid:
                return await self.dao.find_by_uuid(resume_uuid)
        except ValueError:
            # 尝试解析为 BIGINT ID
            resume_id_int = int(resume_id)
            return await self.dao.find_by_id(resume_id_int)
        return None


# orchestrator.py
class InterviewOrchestrator:
    async def end_interview_session(self, session_uuid: str) -> bool:
        # 使用 UUID 查找 session
        session = await self.session_dao.find_by_uuid(UUID(session_uuid))
        if not session:
            return False


        # 应用层使用 BIGINT ID 进行更新
        await self.session_dao.end_session(session.id)


        # 反馈关联也使用 BIGINT ID
        await self.feedback_dao.create_feedback(
            session_id=session.id,
            feedback_data=self.generate_feedback()
        )
        return True

五、迁移脚本详解#

迁移采用分阶段执行策略，确保每阶段可验证、可回滚。

阶段 1：创建序列#

-- 为每个表创建序列
CREATE SEQUENCE IF NOT EXISTS users_id_seq;
CREATE SEQUENCE IF NOT EXISTS resumes_id_seq;
CREATE SEQUENCE IF NOT EXISTS projects_id_seq;
CREATE SEQUENCE IF NOT EXISTS knowledge_base_id_seq;
CREATE SEQUENCE IF NOT EXISTS interview_sessions_id_seq;
CREATE SEQUENCE IF NOT EXISTS qa_history_id_seq;
CREATE SEQUENCE IF NOT EXISTS interview_feedback_id_seq;

阶段 2：处理 users 表#

-- 添加新的 BIGSERIAL 列和临时 UUID 列
ALTER TABLE users ADD COLUMN id_new BIGSERIAL;
ALTER TABLE users ADD COLUMN uuid_new UUID;


-- 将现有 UUID 迁移到新列
UPDATE users SET uuid_new = uuid;


-- 删除旧列，重命名新列
ALTER TABLE users DROP COLUMN uuid;
ALTER TABLE users ADD COLUMN uuid UUID;
UPDATE users SET uuid = uuid_new;
ALTER TABLE users DROP COLUMN uuid_new;


-- 设置序列所有权
ALTER SEQUENCE users_id_seq OWNED BY users.id;
ALTER TABLE users ALTER COLUMN id SET DEFAULT nextval('users_id_seq'::regclass);


-- 重建索引
REINDEX TABLE users;

阶段 3-8：处理其他表#

其他表的迁移模式与 users 表相同：添加新列 → 填充数据 → 删除旧列 → 重命名 → 设置序列。关键区别在于非 users 表不保留 UUID 字段，外键列直接转换为 BIGINT 类型。

-- resumes 表示例
ALTER TABLE resumes ADD COLUMN id_new BIGSERIAL;
ALTER TABLE resumes ADD COLUMN user_id_new BIGINT;


-- 建立外键映射
UPDATE resumes SET user_id_new = (
    SELECT u.id FROM users u WHERE u.uuid = resumes.user_id
);


-- 删除旧外键，启用新外键
ALTER TABLE resumes DROP COLUMN user_id;
ALTER TABLE resumes ADD COLUMN user_id BIGINT;
UPDATE resumes SET user_id = user_id_new;
ALTER TABLE resumes DROP COLUMN user_id_new;
ALTER TABLE resumes ADD CONSTRAINT fk_resumes_user
    FOREIGN KEY (user_id) REFERENCES users(id);


-- 重命名主键
ALTER TABLE resumes DROP COLUMN id;
ALTER TABLE resumes ADD COLUMN id BIGINT;
UPDATE resumes SET id = id_new;
ALTER TABLE resumes DROP COLUMN id_new;


ALTER SEQUENCE resumes_id_seq OWNED BY resumes.id;
ALTER TABLE resumes ALTER COLUMN id SET DEFAULT nextval('resumes_id_seq'::regclass);
REINDEX TABLE resumes;

阶段 9：最终验证#

-- 验证 UUID 唯一性
SELECT
    COUNT(*) as total,
    COUNT(DISTINCT uuid) as unique_uuids,
    COUNT(*) - COUNT(DISTINCT uuid) as duplicates
FROM users;


-- 验证外键完整性
SELECT COUNT(*) as orphaned_resumes
FROM resumes r
LEFT JOIN users u ON r.user_id = u.id
WHERE u.id IS NULL;


-- 验证序列状态
SELECT
    sequencename,
    last_value,
    last_value / 1000.0 as usage_ratio
FROM pg_sequences
WHERE schemaname = 'public';

六、验证与测试方法#

迁移后验证清单#

验证项SQL 语句预期结果
主键自增SELECT id FROM users ORDER BY id DESC LIMIT 5返回递增 ID
UUID 唯一性SELECT COUNT(DISTINCT uuid) FROM users等于总行数
外键完整性SELECT COUNT(*) FROM resumes WHERE user_id NOT IN (SELECT id FROM users)返回 0
序列状态SELECT last_value FROM users_id_seq大于 0
索引效率SELECT pg_size_pretty(pg_relation_size('users_pkey'))索引体积明显减小

应用层测试#

tests/test_migration.py
import pytest
from uuid import uuid4


class TestMigration:
    @pytest.mark.asyncio
    async def test_resume_crud_with_bigserial(self, dao: ResumeDAO):
        # 创建
        resume = await dao.create(user_id=1, resume_data={"title": "Engineer"})
        assert isinstance(resume.id, int)
        assert resume.id > 0


        # 按 ID 查询
        found = await dao.find_by_id(resume.id)
        assert found.id == resume.id


        # 按 UUID 查询（需先添加 uuid 字段到 resume）
        # found_by_uuid = await dao.find_by_uuid(resume.uuid)
        # assert found_by_uuid.id == resume.id


    @pytest.mark.asyncio
    async def test_foreign_key_integrity(self, session_dao: SessionDAO):
        # 创建关联记录
        user = await user_dao.create(name="Test User")
        resume = await resume_dao.create(user_id=user.id)


        # 创建 session
        session = await session_dao.create(
            user_id=user.id,
            resume_id=resume.id
        )


        # 验证外键关系正确
        assert session.user_id == user.id
        assert session.resume_id == resume.id

性能基准测试#

-- 迁移前后性能对比
EXPLAIN ANALYZE
SELECT * FROM interview_sessions
WHERE user_id = 12345
ORDER BY created_at DESC
LIMIT 20;


-- 预期结果：索引扫描，execution time 显著下降

七、总结#

本次迁移从 UUID 主键切换到 BIGSERIAL 主键，核心收益包括：

存储优化：主键索引体积减少约 50%，在千万级记录量下，索引从数 GB 压缩到数百 MB。

写入性能提升：顺序写入充分利用 B+tree 末端追加特性，高并发写入场景下 TPS 提升约 2-3 倍。

类型系统完善：外键统一使用 BIGINT，支持与枚举类型结合，满足业务状态机的设计需求。

API 兼容性保持：仅 users 表保留 UUID 作为外部标识符，网关层完成转换，现有 API 接口无需修改。

迁移过程中最关键的实践是分阶段执行和充分验证——每个阶段独立可验证，出现问题可快速回滚，而非一次性大规模变更带来的不可控风险。

八、更改知识库RAG为独立api#

数据架构与面试流程

存储层

  ┌─────────────────┬────────────────────────┬───────────────────────────┐
  │      存储       │          用途          │         关键技术          │
  ├─────────────────┼────────────────────────┼───────────────────────────┤
  │ PostgreSQL      │ 持久化结构化数据       │ SQLAlchemy + pgvector     │
  ├─────────────────┼────────────────────────┼───────────────────────────┤
  │ Redis           │ 会话状态（短中期记忆） │ 异步 Redis，TTL 24h       │
  ├─────────────────┼────────────────────────┼───────────────────────────┤
  │ LangGraph State │ 运行时状态（短期记忆） │ InterviewState dataclass  │
  ├─────────────────┼────────────────────────┼───────────────────────────┤
  │ VectorStore     │ 向量相似度搜索         │ In-memory + pgvector 支持 │
  └─────────────────┴────────────────────────┴───────────────────────────┘

核心实体关系

  User (1) ───< Resume (1) ────< Project (1) ───< KnowledgeBase (N)
                          │
                          └───< InterviewSession (N) ───< QAHistory (N)
                                                        │
                                                        └──── InterviewFeedback (1)

PostgreSQL 表结构:

  - users - 用户账户
  - resumes - 简历解析结果 (JSONB)
  - projects - 项目经历
  - knowledge_base - RAG 知识库条目 (含 skill_point, responsibility_id)
  - interview_sessions - 面试会话记录
  - qa_history - Q&A 历史（含 deviation_score）
  - interview_feedback - 最终反馈

面试数据流

  ┌─────────────────────────────────────────────────────────────────┐
  │  1. start_interview()                                           │
  │  ├── Resume/RJD 解析 → resume_context                          │
  │  ├── 向量检索 → 匹配 skill_points/responsibilities             │
  │  └── 初始化 InterviewState (LangGraph)                         │
  └─────────────────────────────────────────────────────────────────┘
                                ↓
  ┌─────────────────────────────────────────────────────────────────┐
  │  2. LangGraph Orchestrator (环形流程)                            │
  │                                                                 │
  │   ┌──────────┐    ┌──────────────┐    ┌───────────────────┐     │
  │   │ question │───>│ evaluate     │───>│ review            │     │
  │   │ agent    │    │ agent        │    │ agent             │     │
  │   └──────────┘    └──────────────┘    └───────────────────┘     │
  │         ↑                                                │      │
  │         └──────────────── decision ──────────────────────┘      │
  │                                                                 │
  │  Phase 流转: init → warmup → initial → followup → final_feedback│
  └─────────────────────────────────────────────────────────────────┘
                                ↓
  ┌─────────────────────────────────────────────────────────────────┐
  │  3. submit_answer()                                             │
  │  ├── question_id + content → Answer                              │
  │  ├── evaluate_agent → evaluation_results[question_id]            │
  │  │     └── {deviation_score, is_correct, key_points}            │
  │  ├── review_agent → 确认评估合理性                               │
  │  ├── feedback_agent → Feedback ( RECORDED 模式)                  │
  │  └── 更新 Redis Session Memory                                   │
  └─────────────────────────────────────────────────────────────────┘
                                ↓
  ┌─────────────────────────────────────────────────────────────────┐
  │  4. end_interview()                                             │
  │  ├── 生成最终反馈 (_generate_final_feedback)                     │
  │  │     ├── aggregate_series_score()                             │
  │  │     ├── extract_strengths/weaknesses()                      │
  │  │     └── generate_suggestions()                              │
  │  ├── 写入 PostgreSQL (InterviewSession + QAHistory + Feedback)    │
  │  └── 清理 Redis Session Memory                                   │
  └─────────────────────────────────────────────────────────────────┘

各 Agent 数据职责

  ┌─────────────────┬─────────────────────────────────────────┬──────────────────────────────────────────┐
  │      Agent      │                  职能                   │                 关键产出                 │
  ├─────────────────┼─────────────────────────────────────────┼──────────────────────────────────────────┤
  │ resume_agent    │ 解析简历，提取 responsibilities/modules │ resume_context, identified_modules       │
  ├─────────────────┼─────────────────────────────────────────┼──────────────────────────────────────────┤
  │ knowledge_agent │ RAG 检索，匹配 skill_points             │ current_knowledge, enterprise_docs       │
  ├─────────────────┼─────────────────────────────────────────┼──────────────────────────────────────────┤
  │ question_agent  │ 生成面试问题                            │ current_question, current_question_id    │
  ├─────────────────┼─────────────────────────────────────────┼──────────────────────────────────────────┤
  │ evaluate_agent  │ 评估回答质量                            │ evaluation_results[question_id]          │
  ├─────────────────┼─────────────────────────────────────────┼──────────────────────────────────────────┤
  │ review_agent    │ 审核评估结果                            │ review_retry_count, last_review_feedback │
  ├─────────────────┼─────────────────────────────────────────┼──────────────────────────────────────────┤
  │ feedback_agent  │ 生成实时反馈                            │ feedbacks[question_id]                   │
  └─────────────────┴─────────────────────────────────────────┴──────────────────────────────────────────┘

状态管理层

  ┌─────────────────────────────────────────────────────┐
  │  LangGraph InterviewState (运行时)                   │
  │  - answers, feedbacks, evaluation_results             │
  │  - series_history, followup_chain                   │
  │  - enterprise_docs, current_module/skill_point        │
  └─────────────────────────────────────────────────────┘
          ↓  (每个 API 调用后持久化)
  ┌─────────────────────────────────────────────────────┐
  │  Redis (短中期记忆)                                 │
  │  - interview:{session_id}:state                     │
  │  - interview:{session_id}:series:{n}:q1 (预生成)     │
  │  - user:{user_id}:current_interview                 │
  └─────────────────────────────────────────────────────┘
          ↓  (end_interview 时)
  ┌─────────────────────────────────────────────────────┐
  │  PostgreSQL (持久化)                                 │
  │  - InterviewSession, QAHistory, InterviewFeedback    │
  │  - knowledge_base (RAG 条目)                       │
  └─────────────────────────────────────────────────────┘

向量知识库 (RAG)

KnowledgeBase 表字段:

  • skill_point - 技能点名称 (如 “微服务设计”)
  • responsibility_id / responsibility_text - 职责索引
  • content - 知识内容
  • embedding_id - pgvector 引用
  • question_id / session_id - 用于问题去重

VectorStore 用途:

  • 简历内容向量化 → 匹配相关技能点
  • 问题内容向量化 → 检索相似历史问题
  • 标准答案向量化 → 计算 deviation_score
AI-Interview 数据库主键迁移实践：UUID → BIGSERIAL
https://sgjki547.top/posts/ai-interview-database/
Author
SGJki
Published at
2026-04-10
License
CC BY-NC-SA 4.0
AI Interview Skill System
ThreadPool
© 2026 SGJki. All Rights Reserved. / RSS / Sitemap
Powered by Astro & Fuwari