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 存在，实现数据层与展示层的解耦。

架构对比

表名	迁移前	迁移后
users	id: UUID (PK)	id: BIGSERIAL (PK), uuid: UUID (API 用)
resumes	id: UUID (PK), user_id: UUID (FK)	id: BIGSERIAL (PK), user_id: BIGINT (FK)
projects	id: UUID (PK), resume_id: UUID (FK)	id: BIGSERIAL (PK), resume_id: BIGINT (FK)
knowledge_base	id: UUID (PK), project_id: UUID (FK)	id: BIGSERIAL (PK), project_id: BIGINT (FK)
interview_sessions	id: UUID (PK), user_id/resume_id: UUID (FK)	id: BIGSERIAL (PK), user_id/resume_id: BIGINT (FK)
qa_history	id: UUID (PK), session_id: UUID (FK)	id: BIGSERIAL (PK), session_id: BIGINT (FK)
interview_feedback	id: UUID (PK), session_id: UUID (FK)	id: BIGSERIAL (PK), session_id: BIGINT (FK)

三、数据库架构设计

ER 关系图

erDiagram
    users ||--o{ resumes : "user_id"
    resumes ||--o{ projects : "resume_id"
    projects ||--o{ knowledge_base : "project_id"
    users ||--o{ interview_sessions : "user_id"
    resumes ||--o{ interview_sessions : "resume_id"
    interview_sessions ||--o{ qa_history : "session_id"
    interview_sessions ||--o{ interview_feedback : "session_id"

    users {
        bigint id PK
        uuid uuid "API用"
    }
    resumes {
        bigint id PK
        bigint user_id FK
    }
    projects {
        bigint id PK
        bigint resume_id FK
    }
    knowledge_base {
        bigint id PK
        bigint project_id FK
    }
    interview_sessions {
        bigint id PK
        bigint user_id FK
        bigint resume_id FK
    }
    qa_history {
        bigint id PK
        bigint session_id FK
    }
    interview_feedback {
        bigint id PK
        bigint session_id FK
    }

四、关键代码实现

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 支持

---

核心实体关系

erDiagram
    User ||--o{ Resume : ""
    Resume ||--o{ Project : ""
    Project ||--o{ KnowledgeBase : ""
    Resume ||--o{ InterviewSession : ""
    InterviewSession ||--o{ QAHistory : ""
    InterviewSession ||--o| InterviewFeedback : ""

PostgreSQL 表结构:

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

---

面试数据流

graph TD
    A1["1. start_interview()"] --> B1["Resume/RJD 解析 → resume_context"]
    A1 --> C1["向量检索 → 匹配 skill_points/responsibilities"]
    A1 --> D1["初始化 InterviewState (LangGraph)"]
    D1 --> E2["2. LangGraph Orchestrator (环形流程)"]
    E2 --> F2["question agent"]
    F2 --> G2["evaluate agent"]
    G2 --> H2["review agent"]
    H2 -->|decision| F2
    E2 --> I2["Phase 流转: init → warmup → initial → followup → final_feedback"]
    I2 --> J3["3. submit_answer()"]
    J3 --> K3["question_id + content → Answer"]
    J3 --> L3["evaluate_agent → evaluation_results[question_id]"]
    L3 --> M3["{deviation_score, is_correct, key_points}"]
    J3 --> N3["review_agent → 确认评估合理性"]
    J3 --> O3["feedback_agent → Feedback (RECORDED 模式)"]
    J3 --> P3["更新 Redis Session Memory"]
    P3 --> Q4["4. end_interview()"]
    Q4 --> R4["生成最终反馈 (_generate_final_feedback)"]
    R4 --> S4["aggregate_series_score()"]
    R4 --> T4["extract_strengths/weaknesses()"]
    R4 --> U4["generate_suggestions()"]
    Q4 --> V4["写入 PostgreSQL (InterviewSession + QAHistory + Feedback)"]
    Q4 --> W4["清理 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]

---

状态管理层

graph TD
    A["LangGraph InterviewState (运行时)<br/>answers, feedbacks, evaluation_results<br/>series_history, followup_chain<br/>enterprise_docs, current_module/skill_point"]
    A -->|每个 API 调用后持久化| B["Redis (短中期记忆)<br/>interview:{session_id}:state<br/>interview:{session_id}:series:{n}:q1 (预生成)<br/>user:{user_id}:current_interview"]
    B -->|end_interview 时| C["PostgreSQL (持久化)<br/>InterviewSession, QAHistory, InterviewFeedback<br/>knowledge_base (RAG 条目)"]

---

向量知识库 (RAG)

KnowledgeBase 表字段:

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

VectorStore 用途:

• 简历内容向量化 → 匹配相关技能点
• 问题内容向量化 → 检索相似历史问题
• 标准答案向量化 → 计算 deviation_score

Sources

No external sources for this entry.

Related

2026-04-19 · blog

AI Interview 中的实体关系分析

梳理 AI Interview 项目的 PostgreSQL 数据模型：核心业务表、pgvector 向量存储的五种嵌入类型，以及 Mermaid 实体关系图与关联关系。专题系列 · 3 min read 2026-04-11 · blog

RL-Scheduler 中的 Master-Worker 机制可用性分析

Master，Worker维持任务的最终一致性分析专题系列 · 4 min read 2026-04-10 · blog

AI Interview Skill System

AI-Interview 多 Agent 系统的 Context-Aware Skill 加载机制：按 phase、action、condition 三维触发按需注入方法论文档，将 Prompt token 消耗降低约 44%。专题系列 · 8 min read