22 KiB
22 KiB
Story Extractor — 后端架构设计文档
版本：v1.3 · 2026-04-21
一、架构总览
1.1 技术栈
| 层级 | 技术 | 选型理由 |
|---|---|---|
| Web 框架 | FastAPI | 异步支持、自动 API 文档、类型提示、高性能 |
| 前端 | Vue 3 + Element Plus | 组件丰富、中文生态好、适合管理类交互 |
| 数据库 | SQLite | 轻量零安装、局域网场景够用、后续可迁移 |
| LLM 接口 | OpenAI Python SDK（兼容协议） | 一套代码适配 OpenAI / DeepSeek / 通义千问 / Ollama |
| DOCX 处理 | python-docx | 读写 DOCX 的标准 Python 库 |
| 部署 | Docker Compose | 一键启动，局域网分发方便 |
设计原则：极简架构，不引入 Redis、Celery、Nginx 等外部依赖，全部基于 FastAPI 原生能力。
1.2 系统架构图
┌──────────────────┐
│ 浏览器 (Vue 3) │
└────────┬─────────┘
│ HTTP / SSE
┌────────┴─────────┐
│ │
┌──────┴──────┐ ┌────────┴──┐
│ FastAPI │ │ SQLite │
│ (Web + API │ │ (数据库) │
│ + 静态托管 │ │ │
│ + 后台任务) │ │ │
└──────┬──────┘ └───────────┘
│
┌───────────┼───────────┐
│ │ │
┌─────┴─────┐ ┌──┴───┐ ┌────┴────┐
│ 预处理器 │ │ LLM │ │ DOCX │
│ + 提取器 │ │ 服务 │ │ 生成器 │
└───────────┘ └──────┘ └─────────┘
极简设计：
- ❌ 无 Redis（进度追踪用内存）
- ❌ 无 Celery（后台任务用 asyncio）
- ❌ 无 Nginx（FastAPI 托管静态文件）
- ✅ 单容器部署，一键启动
1.3 项目目录结构
story-extractor/
├── docker-compose.yml
├── Dockerfile
├── requirements.txt
├── .env.example
│
├── backend/
│ ├── app/
│ │ ├── __init__.py
│ │ ├── main.py # FastAPI 入口，路由注册，SSE
│ │ ├── config.py # 配置管理（读取 .env）
│ │ ├── database.py # SQLite 连接与 ORM
│ │ ├── state.py # 内存状态管理（进度追踪）
│ │ │
│ │ ├── api/ # API 路由层
│ │ │ ├── __init__.py
│ │ │ ├── files.py # 文件上传/删除/列表
│ │ │ ├── extraction.py # 提取任务管理
│ │ │ ├── stories.py # 故事 CRUD（含删除）
│ │ │ ├── matching.py # 匹配关系管理
│ │ │ ├── export.py # 预览与导出
│ │ │ └── settings.py # 系统配置
│ │ │
│ │ ├── models/ # 数据模型（SQLAlchemy）
│ │ │ ├── __init__.py
│ │ │ ├── transcript.py
│ │ │ ├── person.py
│ │ │ ├── story.py
│ │ │ └── config.py
│ │ │
│ │ ├── schemas/ # Pydantic 请求/响应模型
│ │ │ ├── __init__.py
│ │ │ ├── file.py
│ │ │ ├── story.py
│ │ │ ├── match.py
│ │ │ └── config.py
│ │ │
│ │ └── services/ # 业务逻辑层
│ │ ├── __init__.py
│ │ ├── file_service.py # 文件存储与管理
│ │ ├── docx_parser.py # DOCX 解析
│ │ ├── preprocessor.py # 预处理（过滤非故事内容）
│ │ ├── extractor.py # 故事提取核心（三阶段 Pipeline）
│ │ ├── llm_client.py # LLM 抽象层
│ │ ├── match_service.py # 匹配关系管理
│ │ └── docx_generator.py # 最终 DOCX 生成
│ │
│ ├── uploads/ # 上传文件存储
│ │ ├── transcripts/
│ │ └── persons/
│ │
│ └── exports/ # 导出文件存储
│
├── frontend/ # Vue 3 前端项目
│ ├── src/
│ │ ├── views/
│ │ │ ├── WelcomeView.vue
│ │ │ ├── UploadView.vue
│ │ │ ├── ExtractView.vue
│ │ │ ├── MatchView.vue
│ │ │ └── ExportView.vue
│ │ ├── components/
│ │ │ ├── Stepper.vue
│ │ │ ├── BottomNav.vue
│ │ │ ├── UploadZone.vue
│ │ │ ├── StoryCard.vue
│ │ │ ├── PersonCard.vue
│ │ │ ├── MatchPanel.vue
│ │ │ ├── DocPreview.vue
│ │ │ ├── SettingsModal.vue
│ │ │ └── Toast.vue
│ │ ├── api/
│ │ │ └── index.js
│ │ ├── stores/
│ │ │ └── app.js
│ │ ├── App.vue
│ │ └── main.js
│ ├── package.json
│ └── vite.config.js
│
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── .env.example
二、API 设计
2.1 API 总览
所有 API 以 /api/v1 为前缀。
| 模块 | 方法 | 路径 | 说明 |
|---|---|---|---|
| 文件 | POST | /api/v1/files/transcripts |
上传文字稿（支持批量） |
| POST | /api/v1/files/persons |
上传讲述者信息（支持批量） | |
| GET | /api/v1/files/transcripts |
获取文字稿列表 | |
| GET | /api/v1/files/persons |
获取讲述者列表 | |
| DELETE | /api/v1/files/transcripts/{id} |
删除文字稿 | |
| DELETE | /api/v1/files/persons/{id} |
删除讲述者信息 | |
| GET | /api/v1/files/persons/{id}/preview |
预览讲述者信息 | |
| 提取 | POST | /api/v1/extraction/start |
启动提取任务（后台异步） |
| GET | /api/v1/extraction/status |
获取提取进度 | |
| GET | /api/v1/extraction/status/stream |
SSE 实时进度推送 | |
| POST | /api/v1/extraction/stop |
停止提取任务 | |
| 故事 | GET | /api/v1/stories |
获取故事列表（支持筛选） |
| GET | /api/v1/stories/{id} |
获取故事详情 | |
| PUT | /api/v1/stories/{id} |
编辑故事 | |
| DELETE | /api/v1/stories/{id} |
删除故事 | |
| 匹配 | POST | /api/v1/matches |
创建匹配关系 |
| DELETE | /api/v1/matches/{story_id} |
取消匹配 | |
| GET | /api/v1/matches |
获取所有匹配关系 | |
| 导出 | GET | /api/v1/export/preview/{person_id} |
预览文档 |
| GET | /api/v1/export/download/{person_id} |
下载单个 DOCX | |
| GET | /api/v1/export/download-all |
批量下载 ZIP | |
| GET | /api/v1/export/list |
获取导出列表 | |
| 设置 | GET | /api/v1/settings |
获取系统配置 |
| PUT | /api/v1/settings |
更新系统配置 | |
| POST | /api/v1/settings/test-llm |
测试 LLM 连接 |
2.2 关键 API 详细设计
启动提取任务
POST /api/v1/extraction/start
响应：
{
"task_id": "uuid",
"status": "started",
"total_files": 10
}
SSE 实时进度推送
GET /api/v1/extraction/status/stream
实现方式：FastAPI StreamingResponse + asyncio.Queue
@router.get("/extraction/status/stream")
async def stream_progress():
async def event_generator():
while True:
data = await extraction_state.get_update()
yield f"event: progress\ndata: {json.dumps(data)}\n\n"
if data.get("status") == "completed":
break
return StreamingResponse(event_generator(), media_type="text/event-stream")
SSE 事件格式：
event: progress
data: {
"percent": 35,
"current_file": "2026年1月21日 第10营训练营.docx",
"action": "故事识别：检测学员发言...",
"files_read": 3,
"files_total": 10,
"segments_found": 42,
"stories_found": 2
}
event: story_found
data: {
"id": "uuid",
"title": "代际传承的体罚创伤",
"speaker_nickname": "14班+HCM",
"summary": "从小被妈妈体罚，长大后用同样方式对待女儿...",
"confidence": 0.85,
"source": "2026年1月21日 第10营训练营.docx"
}
event: completed
data: {
"total_files": 10,
"total_stories": 15,
"duration_seconds": 186
}
停止提取任务
POST /api/v1/extraction/stop
说明：通过 asyncio.Event 取消后台任务。
三、核心服务设计
3.1 内存状态管理（state.py）
替代 Redis 的方案：用 Python 模块级变量 + asyncio.Queue 管理进度。
class ExtractionState:
"""提取任务状态管理（内存中，无需 Redis）"""
def __init__(self):
self.is_running: bool = False
self.cancel_event: asyncio.Event = asyncio.Event()
self.progress: dict = {
"percent": 0,
"current_file": "",
"action": "",
"files_read": 0,
"files_total": 0,
"segments_found": 0,
"stories_found": 0,
}
self.queue: asyncio.Queue = asyncio.Queue()
def update(self, **kwargs):
"""更新进度，同时推送到 SSE 队列"""
self.progress.update(kwargs)
self.queue.put_nowait(self.progress.copy())
async def get_update(self) -> dict:
"""SSE 消费端：等待新的进度更新"""
return await self.queue.get()
def reset(self):
"""重置状态"""
self.is_running = False
self.cancel_event.clear()
self.progress = {k: 0 if k != "" else "" for k in self.progress}
self.queue = asyncio.Queue()
# 全局单例
extraction_state = ExtractionState()
局限性：
- 服务重启后进度丢失（可接受，重新提取即可）
- 单进程内有效（FastAPI 单进程部署，无问题）
3.2 后台任务管理
替代 Celery 的方案：FastAPI asyncio.create_task()
# main.py
from app.services.extractor import run_extraction
@router.post("/extraction/start")
async def start_extraction():
if extraction_state.is_running:
return {"error": "提取任务正在进行中"}
extraction_state.reset()
extraction_state.is_running = True
asyncio.create_task(run_extraction(extraction_state))
return {"task_id": str(uuid4()), "status": "started"}
@router.post("/extraction/stop")
async def stop_extraction():
extraction_state.cancel_event.set()
return {"status": "stopping"}
# extractor.py
async def run_extraction(state: ExtractionState):
"""后台提取任务主循环"""
try:
transcripts = get_all_transcripts()
state.update(files_total=len(transcripts))
for i, transcript in enumerate(transcripts):
if state.cancel_event.is_set():
break
state.update(
current_file=transcript.filename,
action="预处理：过滤课堂管理内容...",
files_read=i + 1,
)
paragraphs = parse_transcript(transcript.file_path)
filtered = preprocess(paragraphs)
segments = await segment_text(filtered, state)
stories = await identify_and_extract(segments, state)
save_stories(stories, transcript.id)
state.update(
percent=round((i + 1) / len(transcripts) * 100),
action=f"已完成：{transcript.filename}",
)
state.update(status="completed", percent=100)
except Exception as e:
state.update(status="error", action=f"错误：{str(e)}")
finally:
state.is_running = False
3.3 文字稿解析器（docx_parser.py）
def parse_transcript(file_path: str) -> TranscriptData:
"""
解析直播回放文字稿 DOCX
格式：发言者(时间戳): 内容
返回：
- paragraphs: [{index, speaker, timestamp, content, raw_text}]
- total_lines: 总行数
"""
关键处理：
- 解析每段的发言者、时间戳、内容
- 识别发言者类型：老师（卢慧老师、静静、督导）vs 学员
- 建立行号映射关系
3.4 预处理器（preprocessor.py）
class Preprocessor:
"""预处理：过滤非故事内容"""
def filter_classroom_management(self, paragraphs: list) -> list:
"""过滤课堂管理内容（改昵称、纪律提醒、进组讨论）"""
def filter_short_responses(self, paragraphs: list) -> list:
"""过滤学员短回应（"好的""收到"等，长度<10字）"""
def filter_healing_exercises(self, paragraphs: list) -> list:
"""过滤疗愈练习重复内容（连续"我愿意看见且释放"模式）"""
def extract_student_segments(self, paragraphs: list) -> list:
"""提取学员发言段落，保留上下文"""
过滤规则：
- 课堂管理关键词：昵称、纪律、进组、会议室、扣三个一
- 短回应：长度 < 10 字，且不含情感词
- 疗愈练习：连续 3 段以上匹配"我愿意看见且释放"模式
3.5 故事提取器（extractor.py）
核心算法：三阶段 Pipeline
阶段1: 预处理 → 阶段2: 故事识别 → 阶段3: 故事提取
阶段一：预处理（规则过滤）
def preprocess(self, paragraphs: list) -> list:
"""
1. 过滤课堂管理内容
2. 过滤短回应
3. 过滤疗愈练习重复
4. 提取学员发言段落
"""
阶段二：故事识别（LLM）
分块策略：
- 按 300-500 行分块，块间重叠 30 行
- 每块送入 LLM 判断是否包含学员故事
Prompt 模板：
你是一个文本分析助手。以下是某心理成长训练营直播的文字稿片段。
每段格式为：发言者(时间戳): 内容
请判断该片段是否包含"学员个人故事"。
【学员个人故事】的定义：
1. 学员（非老师）讲述的自己或家人的真实经历
2. 涉及具体的事件、人物、时间线
3. 有情感色彩（痛苦、挣扎、觉醒等）
4. 主题通常围绕：原生家庭、亲子关系、婚姻情感、财富限制、身体健康、职业发展
【不属于学员故事】的内容：
1. 老师的理论讲解和方法论
2. 课堂管理（改昵称、纪律提醒）
3. 疗愈练习引导语（"我愿意看见且释放..."）
4. 学员的简短回应（"好的""明白了"）
5. 老师引用的案例（非现场学员讲述）
请返回 JSON：
{
"has_story": true/false,
"story_speaker": "学员昵称（如14班+HCM）",
"story_topic": "一句话概括主题（15字以内）",
"story_start_line": 起始行号,
"story_end_line": 结束行号,
"confidence": 0.0-1.0
}
阶段三：故事提取（LLM）
Prompt 模板：
以下是某学员在训练营中讲述的个人故事（对话片段）。
请完成以下任务：
1. 整理出完整的故事叙述（去掉老师的提问和引导语，只保留学员的讲述）
2. 为故事起一个简洁的标题（10字以内）
3. 写一段故事摘要（80-150字，第三人称）
4. 识别故事的核心主题标签
【重要】只提取学员讲述的内容，过滤掉：
- 老师的提问、引导、追问
- 疗愈练习引导语
- 其他学员的发言
【输出格式】JSON：
{
"title": "故事标题",
"summary": "故事摘要",
"content": "整理后的完整故事叙述（第一人称，仅学员讲述部分）",
"themes": ["原生家庭", "体罚", "代际传承"],
"emotion_tone": "痛苦/挣扎/觉醒/释然/..."
}
3.6 LLM 客户端（llm_client.py）
class LLMClient:
"""LLM 统一调用接口"""
def __init__(self, config: LLMConfig):
self.client = AsyncOpenAI(
api_key=config.api_key,
base_url=config.base_url,
)
self.model = config.model
async def chat(self, messages: list[dict], temperature: float = 0.3) -> str:
"""发送对话请求"""
async def chat_json(self, messages: list[dict], temperature: float = 0.3) -> dict:
"""发送对话请求，期望返回 JSON"""
支持的提供商（通过 base_url 切换）：
| 提供商 | Base URL |
|---|---|
| OpenAI | https://api.openai.com/v1 |
| DeepSeek | https://api.deepseek.com/v1 |
| 通义千问 | https://dashscope.aliyuncs.com/compatible-mode/v1 |
| Ollama (本地) | http://localhost:11434/v1 |
3.7 匹配服务（match_service.py）
class MatchService:
"""匹配关系管理"""
def create_match(self, story_id: str, person_id: str) -> Match:
"""创建匹配关系"""
def remove_match(self, story_id: str) -> bool:
"""取消匹配"""
def get_matched_persons(self) -> list[Person]:
"""获取已匹配的讲述者列表（用于导出）"""
3.8 DOCX 生成器（docx_generator.py）
def generate_person_doc(person: Person, stories: list[Story], output_path: str):
"""
为一位讲述者生成最终 DOCX 文档
结构：
┌─ 照片（居中）
├─ 个人信息（姓名、性别、年龄、城市、职业）
├─ 分隔线
└─ 故事正文（仅学员讲述内容，多个故事依次排列）
"""
四、错误处理
| 错误类型 | 处理方式 |
|---|---|
| 单个文字稿解析失败 | 记录错误，跳过，继续处理其他文件 |
| LLM API 调用失败 | 自动重试 3 次（指数退避），仍失败则标记该段为 error |
| LLM 返回格式异常 | 尝试修复 JSON，无法修复则丢弃该段结果 |
| DOCX 文件损坏 | 提示用户文件异常，跳过 |
| 提取任务中断（服务重启） | 进度丢失，用户需重新启动提取 |
五、数据库设计
5.1 ER 关系
Transcript 1───* Story *───0..1 Person
Config (单例)
5.2 表结构
transcripts 表
CREATE TABLE transcripts (
id TEXT PRIMARY KEY,
filename TEXT NOT NULL,
file_path TEXT NOT NULL,
line_count INTEGER DEFAULT 0,
file_size INTEGER DEFAULT 0,
status TEXT DEFAULT 'pending',
error_message TEXT,
uploaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
persons 表
CREATE TABLE persons (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
nickname TEXT,
filename TEXT NOT NULL,
file_path TEXT NOT NULL,
photo_path TEXT,
info TEXT,
uploaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
stories 表
CREATE TABLE stories (
id TEXT PRIMARY KEY,
title TEXT NOT NULL,
summary TEXT,
content TEXT,
speaker_nickname TEXT,
source_transcript_id TEXT NOT NULL,
source_lines TEXT,
duration_minutes REAL,
confidence REAL,
confidence_level TEXT,
person_id TEXT,
match_status TEXT DEFAULT 'pending',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (source_transcript_id) REFERENCES transcripts(id),
FOREIGN KEY (person_id) REFERENCES persons(id)
);
config 表
CREATE TABLE config (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
六、部署方案
6.1 Docker Compose（单容器）
version: '3.8'
services:
app:
build: .
ports:
- "80:8000"
volumes:
- ./data/uploads:/app/uploads
- ./data/exports:/app/exports
- ./data/db:/app/data
env_file: .env
6.2 FastAPI 托管静态文件
# main.py
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
app = FastAPI()
# API 路由（先注册）
app.include_router(files_router, prefix="/api/v1")
app.include_router(extraction_router, prefix="/api/v1")
app.include_router(stories_router, prefix="/api/v1")
app.include_router(matching_router, prefix="/api/v1")
app.include_router(export_router, prefix="/api/v1")
app.include_router(settings_router, prefix="/api/v1")
# 托管前端静态文件（放在最后，作为 fallback）
app.mount("/", StaticFiles(directory="frontend/dist", html=True), name="static")
6.3 环境变量（.env）
# LLM 配置
LLM_PROVIDER=openai
LLM_MODEL=gpt-4
LLM_BASE_URL=https://api.openai.com/v1
LLM_API_KEY=sk-xxx
# 提取参数
SEGMENT_MAX_LINES=500
STORY_MIN_LINES=50
CONFIDENCE_THRESHOLD=0.5
TEMPERATURE=0.3
# 应用
APP_HOST=0.0.0.0
APP_PORT=8000
UPLOAD_DIR=/app/uploads
EXPORT_DIR=/app/exports
6.4 启动流程
# 1. 配置环境变量
cp .env.example .env
vim .env
# 2. 构建前端
cd frontend && npm install && npm run build && cd ..
# 3. 一键启动（单容器）
docker-compose up -d
# 4. 访问
# http://<局域网IP>
6.5 依赖清单（requirements.txt）
fastapi>=0.110.0
uvicorn>=0.27.0
python-multipart>=0.0.9
sqlalchemy>=2.0
aiosqlite>=0.19.0
python-docx>=1.1.0
openai>=1.12.0
pydantic>=2.0
仅 8 个依赖，无 Redis、无 Celery。
七、扩展性设计
7.1 LLM 升级
- 所有 LLM 调用通过
LLMClient抽象层 - 切换模型只需修改配置（base_url + model + api_key）
7.2 文章搜索引擎（预留）
- stories 表已存储完整内容 + 结构化元数据
- 后续可接入 SQLite FTS5 全文搜索或 Meilisearch
7.3 多用户支持（预留）
- 当前为单用户模式（局域网共享）
- 后续可加入用户认证（JWT）
7.4 从极简架构升级
如果后续需要更强的任务管理能力，可以平滑升级：
ExtractionState→ Redisasyncio.create_task()→ Celery- 架构设计已预留接口，业务逻辑层不需要改动