从零构建图书创作与二次文创 AI Agent 系统：技术分享

一、项目背景

在 AIGC 时代，如何让大模型成为作家的“智能助手”，而不仅仅是一个聊天机器人？本文介绍一个基于 DeepSeek API 的图书创作 + 二次文创 AI Agent 系统，实现从世界观搭建、人物设定、大纲生成到章节续写的全链路创作能力，同时支持基于原著的同人番外、短剧脚本等衍生内容开发。

---

二、技术选型

层级	技术	说明
后端	Python 3 + Flask	轻量 Web 框架，提供 API 和页面渲染
前端	原生 HTML/CSS/JS + marked.js	单页面应用，Markdown 实时渲染
AI	DeepSeek v4-flash	通过 OpenAI 兼容协议调用
通信	fetch + SSE (Server-Sent Events)	流式生成，实时展示
存储	JSON 文件	模板、历史记录持久化，零依赖
部署	Docker + docker-compose	一键容器化部署

---

三、系统架构

┌─────────────┐     HTTP/SSE      ┌──────────────┐    Proxy     ┌─────────────────┐
│   Browser   │ ◄──────────────►  │  Flask App   │ ◄─────────►  │ DeepSeek API    │
│  (HTML+JS)  │                   │  (app.py)    │              │ (v4-flash)      │
└─────────────┘                   └──────┬───────┘              └─────────────────┘
                                         │
                                         ▼
                                  ┌──────────────┐
                                  │ JSON Storage │
                                  │ data/        │
                                  └──────────────┘

核心设计原则：

1. API Key 不暴露：所有大模型请求经后端代理，前端只调内部接口
2. 流式输出：长文本生成采用 SSE，用户可实时看到生成过程
3. 上下文保持：每次请求携带完整对话历史，保证内容连贯
4. 数据持久化：JSON 文件存储，重启不丢失

---

四、核心模块详解

1. Prompt 工程：动态构建系统提示词

app.py 中的 build_system_prompt() 函数将用户选择的参数拼接为结构化 Prompt：

def build_system_prompt(params):
    genre = params.get("genre", "")
    style = params.get("style", "")
    perspective = params.get("perspective", "")
    stage = params.get("stage", "full")
    custom = params.get("custom", "")

    # 体裁角色定义
    genre_roles = {
        "短文": "你是一位优秀的短篇作家...",
        "长篇小说": "你是一位长篇小说作家...",
        "故事脚本": "你是一位编剧...",
    }

    # 文风描述
    style_descs = {
        "情感细腻": "文字风格要情感细腻...",
        "爽文风格": "节奏明快，爽感十足...",
        "悬疑紧张": "营造紧张悬疑的氛围...",
    }

    parts = []
    parts.append(genre_roles.get(genre, "你是一位专业的创作助手。"))
    
    if style in style_descs:
        parts.append(style_descs[style])
    
    if perspective == "第一人称":
        parts.append('使用第一人称叙事，以"我"的视角展开故事。')
    
    # 创作阶段指令
    stage_instructions = {
        "worldbuilding": "你现在处于【世界观搭建】阶段...",
        "characters": "你现在处于【人物设定】阶段...",
        "outline": "你现在处于【大纲生成】阶段...",
        "continue": "你现在处于【章节续写】阶段...",
        "polish": "你现在处于【内容润色】阶段...",
    }
    if stage in stage_instructions:
        parts.append(stage_instructions[stage])
    
    if custom:
        parts.append(f"额外要求：{custom}")

    return "\n".join(parts)

关键点：

• 通过字典映射将用户下拉框选择转换为专业指令
• 支持"创作阶段"切换，同一主题在不同阶段获得不同指导
• 自定义指令追加到末尾，灵活扩展

2. 流式生成：SSE 实时推送

传统 API 调用需要等待完整响应，对于长文本体验极差。本系统采用 Server-Sent Events 实现流式输出：

@app.route('/api/generate/stream', methods=['POST'])
def generate_stream():
    data = request.json
    system_prompt = build_system_prompt(data.get('params', {}))
    user_input = data.get('user_input', '')
    
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_input}
    ]

    def event_stream():
        full_response = ""
        token_count = 0
        try:
            resp = make_llm_request(messages, stream=True, temperature=0.8)
            for line in resp.iter_lines(decode_unicode=True):
                if not line or not line.startswith("data: "):
                    continue
                data_str = line[6:].strip()
                if data_str == "[DONE]":
                    break
                try:
                    chunk = json.loads(data_str)
                    content = chunk["choices"][0].get("delta", {}).get("content", "")
                    if content:
                        full_response += content
                        token_count += 1
                        yield f"data: {json.dumps({'content': content, 'count': token_count})}\n\n"
                except:
                    continue
            
            # 保存历史
            save_history(project_id, system_prompt, user_input, full_response, token_count)
            yield f"data: {json.dumps({'done': True, 'project_id': project_id})}\n\n"
        except Exception as e:
            yield f"data: {json.dumps({'error': str(e)})}\n\n"

    return Response(
        stream_with_context(event_stream()),
        mimetype="text/event-stream",
        headers={"Cache-Control": "no-cache"}
    )

前端接收逻辑：

function callSSE(url, body, onChunk, onDone, onError) {
    abortController = new AbortController();
    fetch(url, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify(body),
        signal: abortController.signal
    }).then(response => {
        const reader = response.body.getReader();
        const decoder = new TextDecoder();
        let buffer = '';

        function read() {
            reader.read().then(({ done, value }) => {
                if (done) { onDone && onDone(); return; }
                buffer += decoder.decode(value, { stream: true });
                const lines = buffer.split('\n');
                buffer = lines.pop() || '';
                
                for (const line of lines) {
                    if (line.startsWith('data: ')) {
                        try {
                            const data = JSON.parse(line.slice(6));
                            if (data.error) { onError(data.error); return; }
                            if (data.done) { onDone(data); return; }
                            onChunk(data);  // 逐字追加显示
                        } catch (e) {}
                    }
                }
                read();
            });
        }
        read();
    });
}

3. 模板系统：内置 + 自定义

模板存储在 data/templates_default.json 和 data/templates_custom.json，结构如下：

{
  "小说": {
    "都市爽文": {
      "system": "你是一位网文大神级作家。请创作一部都市爽文...",
      "user_template": "题材：{theme}\n文风：{style}\n",
      "params": {"文风": "爽文风格", "篇幅": "3000字开篇"}
    }
  }
}

套用流程：

1. 前端点击"套用"按钮
2. 自动填充右侧参数面板（文风、篇幅等）
3. 将 user_template 填入输入框，用户只需替换 {theme} 占位符
4. 点击生成即可

保存自定义模板：

@app.route('/api/templates/save', methods=['POST'])
def save_template():
    data = request.json
    name = data.get('name')
    category = data.get('category', '自定义')
    config = data.get('config')
    
    with open(CUSTOM_TEMPLATE_FILE, "r", encoding="utf-8") as f:
        customs = json.load(f)
    if category not in customs:
        customs[category] = {}
    customs[category][name] = config
    
    with open(CUSTOM_TEMPLATE_FILE, "w", encoding="utf-8") as f:
        json.dump(customs, f, ensure_ascii=False, indent=2)
    
    return jsonify({"status": "ok"})

4. 二次文创：元素提取 + 定向生成

步骤 1：提取原文核心元素

@app.route('/api/derivative/extract', methods=['POST'])
def extract_elements():
    text = request.json.get('text', '')
    
    messages = [
        {"role": "system", "content": """你是一个专业的文本分析专家。请提取以下字段：
- characters: 人物列表
- scenes: 场景列表
- plot_points: 关键剧情点
- quotes: 经典台词
- worldview: 世界观要素
仅返回JSON，不要额外解释。"""},
        {"role": "user", "content": f"原文：\n{text[:8000]}"}
    ]
    
    resp = make_llm_request(messages, stream=False, temperature=0.3)
    result = resp.json()["choices"][0]["message"]["content"]
    
    # 清理可能的 markdown 代码块标记
    cleaned = re.sub(r'^```\w*\n?', '', result.strip())
    cleaned = re.sub(r'\n?```$', '', cleaned)
    elements = json.loads(cleaned)
    
    return jsonify({"elements": elements})

步骤 2：基于方向生成衍生内容

DERIVATIVE_DIRECTIONS = {
    "同人番外": "请基于原文的人物和世界观，创作一篇同人番外故事...",
    "剧情改写": "请基于原文的故事框架，进行剧情改写...",
    "结局重写": "请为原文重新设计一个结局...",
    "短剧脚本": "请将原文改编为短剧脚本格式...",
    "有声书台词": "请将原文改编为有声书台词版本...",
}

@app.route('/api/derivative/generate/stream', methods=['POST'])
def derivative_generate_stream():
    direction = data.get('direction', '同人番外')
    direction_prompt = DERIVATIVE_DIRECTIONS.get(direction)
    
    system_prompt = f"你是一位专业的文创开发专家。{direction_prompt}"
    if custom_instruction:
        system_prompt += f"\n额外要求：{custom_instruction}"
    
    user_content = f"原文内容：\n{original_text[:6000]}"
    if elements:
        user_content += f"\n\n已提取的核心元素：\n{json.dumps(elements)}"
    
    # ... 后续流式生成逻辑同上

5. 速率限制与 Token 统计

简易速率限制（防滥用）：

_rate_records = {}  # {ip: [timestamps]}
RATE_LIMIT = 30       # 每分钟最多 30 次
RATE_WINDOW = 60      # 窗口 60 秒

def check_rate_limit(client_ip):
    now = time.time()
    records = _rate_records.get(client_ip, [])
    records = [t for t in records if now - t < RATE_WINDOW]
    if len(records) >= RATE_LIMIT:
        return False
    records.append(now)
    _rate_records[client_ip] = records
    return True

在每个 API 入口调用：

client_ip = request.remote_addr
if not check_rate_limit(client_ip):
    return jsonify({"error": "请求过于频繁，请稍后再试"}), 429

Token 统计：

_token_stats = {"total_input": 0, "total_output": 0, "requests": 0}

# 每次生成完成后累加
_token_stats["total_output"] += token_count
_token_stats["requests"] += 1

@app.route('/api/stats')
def get_stats():
    return jsonify(_token_stats)

前端左下角实时展示今日消耗。

---

五、前端交互设计

1. 三栏布局

┌──────────┬──────────────────────────┬──────────────┐
│ 左侧导航  │     中央编辑区            │  右侧参数面板  │
│          │                          │              │
│ ✏ 创作   │  [输入框]                 │  创作主题     │
│ 📋 模板  │  [生成结果 Markdown]      │  体裁下拉     │
│ 🔄 文创  │                          │  文风下拉     │
│ 📁 历史  │                          │  篇幅下拉     │
│          │                          │  自定义指令   │
│          │                          │  [保存模板]   │
└──────────┴──────────────────────────┴──────────────┘

2. 创作阶段导航

用户在"完整创作 / 世界观搭建 / 人物设定 / 大纲生成 / 章节续写 / 内容润色"之间切换，后端根据当前阶段调整 System Prompt。

3. 选中段落定向改写

• 用户在生成结果中选中一段文字
• 自动弹出"改写"按钮
• 输入指令（如"让描写更生动"）
• 后端调用 /api/generate/rewrite，返回改写后内容并替换原文

---

六、Docker 部署

Dockerfile：

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
RUN mkdir -p data/history data/user_templates uploads
EXPOSE 5000
CMD ["python", "app.py"]

docker-compose.yml：

version: '3.8'
services:
  book-ai-agent:
    build: .
    container_name: book-ai-agent
    ports:
      - "5000:5000"
    volumes:
      - ./data:/app/data
      - ./uploads:/app/uploads
    environment:
      - TZ=Asia/Shanghai
    restart: unless-stopped

启动命令：

docker-compose up -d --build

访问 http://localhost:5000 即可使用。

---

七、关键技术总结

技术点	实现方式	价值
Prompt 动态拼接	参数字典映射 + 阶段指令	让非技术人员也能生成高质量 Prompt
SSE 流式输出	Flask stream_with_context + 前端 ReadableStream	长文本生成体验提升 10 倍
模板系统	JSON 文件存储 + 前端卡片展示	降低使用门槛，沉淀最佳实践
二次文创	两阶段处理（提取→生成）	结构化理解原文，衍生更精准
速率限制	基于 IP 的时间窗口计数	防止 API 滥用，控制成本
Docker 化	多阶段构建 + Volume 挂载	一键部署，数据持久化

---

八、源码地址

本项目完整代码已开源，包含：

• app.py：Flask 后端（730 行）
• templates/index.html：前端单页面（988 行）
• Dockerfile + docker-compose.yml：容器化配置
• requirements.txt：Python 依赖

GitHub:

---

九、后续优化方向

1. 向量数据库集成：引入 Chroma/Pinecone，实现长文档语义检索
2. 多模型路由：根据任务复杂度自动选择 deepseek-v4-flash / deepseek-chat
3. 协作功能：支持多人同时编辑同一项目，实时同步
4. 移动端适配：响应式布局，支持手机端创作
5. 插件系统：允许开发者扩展新的文创方向或模板类型

---

结语： 本系统展示了如何将大模型 API、Prompt 工程、Web 开发有机结合，打造一个真正可用的 AI 辅助创作工具。希望对你有所启发！