实现计划：P5 AI 集成层（miniapp-ai-integration）

概述

基于 P5-A 阶段设计，在 apps/backend/app/ai/ 新建 AI 模块，实现百炼 API 封装、SSE 对话、事件调度、缓存服务、8 个 AI 应用（其中 App2/App8 含完整 Prompt，App3/4/5/6/7 仅骨架）。每个任务增量构建，最终通过路由和事件调度器串联所有组件。

任务

1. 数据库表结构与基础模块搭建
- 1.1 创建 DDL 迁移脚本，在 biz schema 下建表 ai_conversations、ai_messages、ai_cache
  - 按设计文档中的 DDL 创建三张表，包含所有字段、CHECK 约束、索引
  - DDL 文件放置于 db/zqyy_app/migrations/ 目录，日期前缀命名
  - 需求: 1.1, 1.2, 1.3, 1.4, 1.5
- 1.2 创建 AI 模块目录结构和 Pydantic Schema
  - 创建 apps/backend/app/ai/ 目录及 __init__.py
  - 创建 apps/backend/app/ai/apps/ 子目录及 __init__.py
  - 创建 apps/backend/app/ai/prompts/ 子目录及 __init__.py
  - 在 apps/backend/app/ai/schemas.py 中定义所有 Pydantic 模型：
    - ChatStreamRequest（message, source_page, page_context, screen_content）
    - SSEEvent（type, content, conversation_id, tokens_used, message）
    - CacheTypeEnum（7 个枚举值）
    - ClueItem（category, summary, detail, emoji）
    - ConsolidatedClueItem（含 providers）
    - App2InsightItem、App4Result、App5TacticsItem、App6Result、App7Result
    - App2Result、App3Result、App8Result
  - 需求: 4.4, 5.2, 5.3, 6.3, 7.3, 8.2, 8.3, 9.2, 10.4, 10.5
- 1.3 编写属性测试：AI 应用输出 JSON 结构验证
  - Property 8: AI 应用输出 JSON 结构验证
  - 使用 hypothesis 生成随机 JSON，验证各应用 Pydantic 模型的解析和校验
  - 验证 App3 category ∈ {客户基础, 消费习惯, 玩法偏好}，App6/8 category ∈ 6 个枚举值
  - 测试文件：tests/test_p5_ai_integration_properties.py
  - 验证: 需求 4.4, 5.2, 5.3, 5.4, 6.3, 7.3, 8.2, 8.3, 8.4, 9.2, 10.4, 10.5
2. 百炼 API 统一封装层（BailianClient）
- 2.1 实现 BailianClient 核心逻辑
  - 文件：apps/backend/app/ai/bailian_client.py
  - 使用 openai Python SDK，base_url 指向百炼端点
  - 实现 chat_stream（流式，AsyncGenerator[str, None]）
  - 实现 chat_json（非流式，返回 tuple[dict, int]）
  - 实现 _inject_current_time（首条消息注入 current_time）
  - 实现 _call_with_retry（指数退避，最多 3 次，1s→2s→4s）
  - 定义异常类：BailianApiError、BailianJsonParseError、BailianAuthError
  - 环境变量：BAILIAN_API_KEY、BAILIAN_BASE_URL、BAILIAN_MODEL
  - 需求: 2.1, 2.2, 2.3, 2.4, 2.5, 2.6
- 2.2 编写属性测试：双模式调用一致性
  - Property 1: BailianClient 双模式调用一致性
  - Mock 百炼 API，验证 chat_stream 返回非空 chunk 序列，chat_json 返回可解析 JSON + 正整数 tokens_used
  - 测试文件：apps/backend/tests/test_ai_bailian.py
  - 验证: 需求 2.1, 2.3, 2.4
- 2.3 编写属性测试：指数退避重试策略
  - Property 2: 指数退避重试策略
  - Mock 可控失败次数的 API，验证重试间隔为 base_interval × 2^(n-1)，超过 max_retries 抛出 BailianApiError
  - 测试文件：apps/backend/tests/test_ai_bailian.py
  - 验证: 需求 2.2
- 2.4 编写属性测试：JSON 解析失败错误处理
  - Property 3: JSON 解析失败错误处理
  - Mock 返回非法 JSON 字符串，验证 chat_json 抛出 BailianJsonParseError
  - 测试文件：apps/backend/tests/test_ai_bailian.py
  - 验证: 需求 2.5
- 2.5 编写属性测试：current_time 注入不变量
  - Property 4: current_time 注入不变量
  - 纯函数测试，随机消息列表，验证首条消息注入 current_time（ISO 格式精确到秒），其余消息不变
  - 测试文件：apps/backend/tests/test_ai_bailian.py
  - 验证: 需求 2.6
3. 对话记录持久化服务（ConversationService）
- 3.1 实现 ConversationService
  - 文件：apps/backend/app/ai/conversation_service.py
  - create_conversation：创建 ai_conversations 记录，系统调用时 user_id='system'
  - add_message：写入 ai_messages 记录（role, content, tokens_used）
  - get_conversations：按 user_id + site_id 查询，created_at DESC，分页（page_size=20）
  - get_messages：按 conversation_id 查询所有消息
  - 需求: 3.2, 3.4, 3.5, 3.7, 13.1, 13.2, 13.3
- 3.2 编写属性测试：AI 调用记录持久化 round-trip
  - Property 5: AI 调用记录持久化 round-trip
  - 使用 test_zqyy_app 数据库，随机 app_id 和消息内容，验证写入后查询一致
  - 测试文件：apps/backend/tests/test_ai_conversation.py
  - 验证: 需求 3.2, 3.4, 3.5, 13.1, 13.2, 13.3
- 3.3 编写属性测试：历史对话列表排序与分页
  - Property 6: 历史对话列表排序与分页
  - 使用 test_zqyy_app 数据库，随机时间戳创建对话，验证返回严格降序且每页 ≤ page_size
  - 测试文件：apps/backend/tests/test_ai_conversation.py
  - 验证: 需求 3.7
4. AI 缓存读写服务（AICacheService）
- 4.1 实现 AICacheService
  - 文件：apps/backend/app/ai/cache_service.py
  - get_latest：按 (cache_type, site_id, target_id) 查询最新记录
  - get_history：查询历史记录（created_at DESC，默认 limit=2），用于 Prompt reference
  - write_cache：写入缓存记录，写入后异步清理超限记录
  - _cleanup_excess：保留最近 500 条，删除最旧的
  - 需求: 12.1, 12.2, 12.3, 12.4, 12.5
- 4.2 编写属性测试：缓存写入 round-trip
  - Property 7: 缓存写入 round-trip
  - 使用 test_zqyy_app 数据库，随机 cache_type、target_id、result_json，验证写入后查询一致
  - 测试文件：apps/backend/tests/test_ai_cache.py
  - 验证: 需求 4.7, 5.6, 6.6, 7.5, 8.6, 9.5, 10.10
- 4.3 编写属性测试：缓存查询 site_id 隔离
  - Property 13: 缓存查询 site_id 隔离
  - 使用 test_zqyy_app 数据库，写入 site_id=A 的记录，以 site_id=B 查询应返回空
  - 测试文件：apps/backend/tests/test_ai_cache.py
  - 验证: 需求 12.1, 12.5
- 4.4 编写属性测试：缓存保留上限
  - Property 14: 缓存保留上限
  - 使用 test_zqyy_app 数据库，批量写入 >500 条记录，验证清理后 ≤ 500
  - 测试文件：apps/backend/tests/test_ai_cache.py
  - 验证: 需求 12.3
5. 检查点 - 基础服务验证
- 确保所有测试通过，ask the user if questions arise.
- 验证 BailianClient、ConversationService、AICacheService 三个核心服务可独立工作
6. 应用 1 通用对话 SSE 端点
- 6.1 实现 App1 Chat 核心逻辑
  - 文件：apps/backend/app/ai/apps/app1_chat.py
  - 每次进入 chat 页面新建 ai_conversations 记录（不复用）
  - 首条消息注入页面上下文（source_page、page_context、screen_content）
  - 用户消息立即写入 ai_messages（role=user）
  - 流式返回完成后写入完整 assistant 回复（含 tokens_used）
  - 通过 biz_params.user_prompt_params 传入 User_ID、Role、Nickname
  - 上下文注入框架留接口（页面文本化工具 P5-B 实现）
  - 需求: 3.1, 3.2, 3.3, 3.4, 3.5, 3.6, 3.8
- 6.2 实现 SSE 路由端点
  - 文件：apps/backend/app/routers/xcx_ai_chat.py
  - POST /api/ai/chat/stream：SSE 协议推送，Content-Type: text/event-stream
  - SSE 事件格式：chunk / done / error
  - GET /api/ai/conversations：历史对话列表（分页，每页 20 条）
  - GET /api/ai/conversations/{conversation_id}/messages：对话消息列表
  - JWT 认证，从 token 提取 user_id、site_id、nickname、role
  - 注册路由到 FastAPI app
  - 需求: 3.1, 3.7
- 6.3 编写单元测试：SSE 端点
  - 验证 SSE Content-Type 和事件格式（chunk/done/error）
  - 验证未认证返回 401、空消息返回 422
  - 测试文件：apps/backend/tests/test_ai_chat.py
  - 需求: 3.1
7. 应用 2 财务洞察（完整 Prompt）
- 7.1 实现 App2 Finance Prompt 模板
  - 文件：apps/backend/app/ai/prompts/app2_finance_prompt.py
  - 完整 Prompt 包含：当期和上期收入结构（table_fee=table_charge_money、assistant_pd=assistant_pd_money、assistant_cx=assistant_cx_money、goods=goods_money、recharge=充值 pay_amount settle_type=5）
  - 包含储值资产、费用汇总、平台结算数据
  - 使用 items_sum 口径，禁止 consume_money
  - 需求: 4.5, 4.6
- 7.2 实现 App2 Finance 核心逻辑
  - 文件：apps/backend/app/ai/apps/app2_finance.py
  - 8 个时间维度独立调用（this_month, last_month, this_week, last_week, last_3_months, this_quarter, last_quarter, last_6_months）
  - 营业日分界点 08:00（BUSINESS_DAY_START_HOUR 环境变量）
  - 每次调用结果写入 ai_cache（cache_type=app2_finance，target_id=时间维度编码）
  - 每次调用创建 ai_conversations + ai_messages 记录
  - 返回结构化 JSON（insights 数组：seq + title + body）
  - 需求: 4.1, 4.2, 4.3, 4.4, 4.7
- 7.3 编写单元测试：App2 时间维度计算
  - 验证 8 个时间维度编码的计算逻辑（营业日分界点 08:00）
  - 验证 Prompt 使用 items_sum 口径字段映射
  - 测试文件：apps/backend/tests/test_ai_app2.py
  - 需求: 4.3, 4.6
8. 应用 3/4/5/6/7 骨架实现
- 8.1 实现 App3 Clue 骨架
  - 文件：apps/backend/app/ai/apps/app3_clue.py
  - run 函数：构建 Prompt → 调用百炼 → 写入 conversation + cache
  - build_prompt：留接口，返回占位 Prompt，标注待细化字段（consumption_records 等待 P9-T1）
  - 线索 category 限定 3 个枚举值，providers 标记为"系统"
  - 使用 items_sum 口径
  - Prompt reference 包含 App6 线索 + 最近 2 套 App8 历史（附 generated_at）
  - 结果写入 ai_cache（cache_type=app3_clue，target_id=member_id）
  - 需求: 5.1, 5.2, 5.3, 5.4, 5.5, 5.6, 5.7, 5.8
- 8.2 实现 App4 Analysis 骨架
  - 文件：apps/backend/app/ai/apps/app4_analysis.py
  - build_prompt：留接口（service_history、assistant_info 待 P6-T4）
  - Prompt reference 包含 App8 最新 + 最近 2 套历史（附 generated_at）
  - 缓存不存在时 reference 传空对象，标注"暂无历史线索"
  - 结果写入 ai_cache（cache_type=app4_analysis，target_id={assistant_id}_{member_id}）
  - 需求: 6.1, 6.2, 6.3, 6.4, 6.5, 6.6, 6.7
- 8.3 实现 App5 Tactics 骨架
  - 文件：apps/backend/app/ai/apps/app5_tactics.py
  - 接收 App4 完整返回结果作为 Prompt 中的 task_suggestion 字段
  - build_prompt：留接口（service_history、assistant_info 随 App4 同步在 P6-T4）
  - Prompt reference 包含最近 2 套 App8 历史（附 generated_at）
  - 结果写入 ai_cache（cache_type=app5_tactics，target_id={assistant_id}_{member_id}）
  - 需求: 7.1, 7.2, 7.3, 7.4, 7.5, 7.6
- 8.4 实现 App6 Note 骨架
  - 文件：apps/backend/app/ai/apps/app6_note.py
  - build_prompt：留接口（consumption_data 待 P9-T1）
  - 返回 score（1-10）+ clues 数组，category 限定 6 个枚举值
  - 线索提供者标记为当前备注提供人
  - 评分规则：6 分为标准分
  - Prompt reference 包含 App3 线索 + 最近 2 套 App8 历史（附 generated_at）
  - 结果写入 ai_cache（cache_type=app6_note_analysis，target_id=member_id），score 存入 ai_cache.score
  - 需求: 8.1, 8.2, 8.3, 8.4, 8.5, 8.6, 8.7, 8.8
- 8.5 实现 App7 Customer 骨架
  - 文件：apps/backend/app/ai/apps/app7_customer.py
  - build_prompt：留接口（objective_data 待 P9-T1）
  - 使用 items_sum 口径
  - 对主观信息标注【来源：XXX，请甄别信息真实性】
  - Prompt reference 包含最新 + 最近 2 套 App8 历史（附 generated_at）
  - 结果写入 ai_cache（cache_type=app7_customer_analysis，target_id=member_id）
  - 需求: 9.1, 9.2, 9.3, 9.4, 9.5, 9.6, 9.7
- 8.6 编写属性测试：Prompt reference 历史注入
  - Property 9: Prompt reference 历史注入
  - Mock 缓存数据，验证各应用 build_prompt 的 reference 字段包含正确的缓存结果和 generated_at 时间戳
  - 缓存不存在时 reference 为空对象
  - 测试文件：apps/backend/tests/test_ai_apps_prompt.py
  - 验证: 需求 5.8, 6.4, 6.5, 7.2, 7.4, 8.8, 9.7
9. 应用 8 维客线索整理（完整 Prompt）+ ClueWriter
- 9.1 实现 App8 Consolidation Prompt 模板
  - 文件：apps/backend/app/ai/prompts/app8_consolidation_prompt.py
  - 完整 Prompt：接收 App3 和 App6 全部线索（附 generated_at），整合去重
  - 分类标签限定 6 个枚举值（与 member_retention_clue CHECK 约束一致）
  - 合并相似线索（多提供者逗号分隔），其余原文返回，最小改动原则
  - 需求: 10.3, 10.4, 10.5, 10.6
- 9.2 实现 ClueWriter 全量替换逻辑
  - 集成在 apps/backend/app/ai/apps/app8_consolidation.py
  - DELETE source IN ('ai_consumption', 'ai_note') → INSERT 新线索（事务）
  - 字段映射：emoji+summary 拼接、providers→recorded_by_name、source 判断逻辑
  - recorded_by_assistant_id 填 NULL
  - 人工线索（source='manual'）不受影响
  - 需求: 10.7, 10.8, 10.9
- 9.3 实现 App8 Consolidation 核心逻辑
  - 文件：apps/backend/app/ai/apps/app8_consolidation.py
  - run 函数：构建 Prompt → 调用百炼 → 写入 conversation + cache + member_retention_clue
  - 结果同时写入 ai_cache（cache_type=app8_clue_consolidated，target_id=member_id）
  - 需求: 10.1, 10.2, 10.10
- 9.4 编写属性测试：ClueWriter 全量替换不变量
  - Property 12: ClueWriter 全量替换不变量
  - 使用 test_zqyy_app 数据库，随机线索列表 + 预置人工线索
  - 验证：AI 线索数量 = new_clues 数量、人工线索不变、recorded_by_assistant_id=NULL、summary=emoji+空格+原始 summary
  - 测试文件：apps/backend/tests/test_ai_clue_writer.py
  - 验证: 需求 10.7, 10.8, 10.9
10. 检查点 - 应用层验证
- 确保所有测试通过，ask the user if questions arise.
- 验证 8 个应用的 run 函数可独立调用（Mock 百炼 API）
11. 事件调度与调用链编排（AIDispatcher） ✅
- 11.1 实现 AIDispatcher 核心逻辑
  - 文件：apps/backend/app/ai/dispatcher.py
  - handle_consumption_event：App3 → App8 → App7（+ App4 → App5 如有助教）
  - handle_note_event：App6 → App8
  - handle_task_assign_event：App4 → App5（读已有 App8 缓存）
  - _run_chain：串行执行调用链，某步失败记录日志后继续
  - 容错：失败应用记录错误日志 + 写入失败 conversation，后续应用使用已有缓存
  - 整条链后台异步执行，不阻塞业务请求
  - 需求: 11.1, 11.2, 11.3, 11.4, 11.5, 11.6, 11.7
- 11.2 集成事件触发点
  - 在 trigger_scheduler.fire_event() 中注册 AI 事件处理器
  - 消费事件（consumption_settled）→ ai_dispatcher.handle_consumption_event
  - 备注事件（note_created）→ ai_dispatcher.handle_note_event
  - 任务分配事件（task_assigned）→ ai_dispatcher.handle_task_assign_event
  - 需求: 5.1, 6.1, 6.2, 7.1, 8.1, 9.1, 11.1, 11.2, 11.3, 11.4
- 11.3 编写属性测试：事件调用链顺序正确性
  - Property 10: 事件调用链顺序正确性
  - Mock 所有应用，记录调用序列，验证四种事件链的严格顺序
  - 测试文件：apps/backend/tests/test_ai_dispatcher.py
  - 验证: 需求 11.1, 11.2, 11.3, 11.4, 11.5, 11.6
- 11.4 编写属性测试：调用链容错不变量
  - Property 11: 调用链容错不变量
  - Mock 随机应用失败，验证后续应用继续执行且失败应用有错误日志
  - 测试文件：apps/backend/tests/test_ai_dispatcher.py
  - 验证: 需求 11.7
12. 缓存查询路由与环境配置
- 12.1 实现缓存查询路由
  - 文件：apps/backend/app/routers/xcx_ai_cache.py
  - GET /api/ai/cache/{cache_type}?target_id=xxx：查询最新缓存
  - JWT 认证，site_id 从 token 提取强制过滤
  - 注册路由到 FastAPI app
  - 需求: 12.1, 12.2, 12.5
- 12.2 新增环境变量配置
  - 在 .env.template 中添加 BAILIAN_API_KEY、BAILIAN_BASE_URL、BAILIAN_MODEL、BUSINESS_DAY_START_HOUR
  - 在后端配置加载逻辑中读取这些变量，缺失时报错
  - 需求: 2.1, 14.1, 14.2
13. 百炼技术方案确认文档 ✅
- 13.1 输出百炼技术方案确认文档
  - 文件：docs/reports/bailian-technical-solution.md
  - 确认流式返回方案（OpenAI 兼容 SSE）
  - 确认 JSON 输出模式（response_format + System Prompt 约束）
  - 确认 SDK 选择（openai Python SDK + base_url 指向百炼）
  - 作为 BailianClient 实现的依据
  - 需求: 14.1, 14.2, 14.3
14. 最终检查点 - 全量验证 ✅
- 全部 9 个测试文件、62 个测试用例通过（2026-03-09）
- 验证所有路由注册正确、事件触发点集成完毕、环境变量配置完整

备注

标记 * 的任务为可选，可跳过以加速 MVP 交付
每个任务引用具体需求编号以确保可追溯
属性测试验证设计文档中定义的 14 个正确性属性
使用 test_zqyy_app 测试库执行数据库相关测试，禁止连接正式库
App3/4/5/6/7 的 Prompt 细化将在 P5-B 阶段（P6/P9 对应任务）中完成

18 KiB Raw Permalink Blame History Unescape Escape

实现计划：P5 AI 集成层（miniapp-ai-integration）

概述

任务

备注

18 KiB

Raw Permalink Blame History