Neo-ZQYY/docs/specs/ai-prompt-refinement/tasks.md

实施计划：NS2 AI Prompt 细化

概述

按照设计文档，将实施拆分为：共享数据获取层 → 应用 3/4/5/6/7 Prompt 拼接 → 应用 1 页面上下文 → 前端筛选参数 → 集成联调。每个任务增量构建，确保无孤立代码。属性测试（Hypothesis）和单元测试作为可选子任务紧跟实现步骤。

后端使用 Python（FastAPI + asyncio），测试使用 Hypothesis + pytest。

任务

• 1. 创建共享数据获取层基础结构

  • 1.1 创建 apps/backend/app/ai/data_fetchers/ 模块骨架
    • 创建 __init__.py，导出 fetch_member_consumption_data、fetch_assistant_info、fetch_service_history、fetch_member_notes、build_page_text
    • 创建 member_data.py、assistant_data.py、page_context.py 空文件
    • 需求: 1.1, 2.1, 8.1

• 2. 实现客户消费数据获取（member_data.py）

  • 2.1 实现 fetch_member_consumption_data(site_id, member_id, months=3)

    • 使用 get_etl_readonly_connection(site_id) + SET LOCAL app.current_site_id 获取 FDW 连接
    • 从 v_dwd_settlement_head + v_dwd_table_fee_log 获取台桌结账（settle_type IN (1,3)）
    • 从 v_dwd_store_goods_sale 获取商城订单
    • 从 v_dim_member_card_account 获取会员卡明细
    • 从 v_dws_member_consumption_summary + v_dws_member_visit_detail 获取汇总和到店数据
    • 从 v_dim_member（scd2_is_current=1）获取会员昵称
    • 消费记录限制 100 条，按 settle_date DESC 排序，超出标注 truncated
    • 金额使用 items_sum 口径，拆分 assistant_pd_money / assistant_cx_money
    • 5 秒查询超时，超时抛出 TimeoutError
    • 需求: 1.1, 1.2, 1.3, 1.4, 1.5, 1.6, 1.7, 1.8, 1.9, 1.10, 11.1, 11.2, 11.3, 11.4, 11.5, 11.7

  • 2.2 实现 fetch_member_notes(site_id, member_id, limit=50)

    • 使用 get_connection() 查询 biz.notes
    • 按 created_at DESC 排序，最多 limit 条
    • 单条备注内容截断 500 字符，超出附加"…（已截断）"
    • 需求: 4.3, 4.4, 6.3, 6.4, 14.4

  • 2.3 编写属性测试：数据获取返回结构完整性（P1）

    • Property 1: 数据获取函数返回结构完整性
    • Mock 数据库返回随机行，验证 fetch_member_consumption_data() 返回字典包含所有必需键
    • 验证每条消费记录包含 settle_date、settle_type、items_sum 等 10 个字段
    • 测试文件: tests/test_data_fetchers/test_member_data_props.py
    • 验证: 需求 1.1, 1.3, 1.5, 2.1, 2.2, 2.4

  • 2.4 编写属性测试：正向交易过滤（P2）+ 金额口径（P3）+ 记录限制与排序（P4）

    • Property 2: 消费记录仅包含正向交易 — 生成混合 settle_type 记录，验证输出仅含 1/3
    • Property 3: 金额口径使用 items_sum — 验证输出包含 items_sum 且不含 consume_money
    • Property 4: 消费记录数量限制与排序 — 生成 0-200 条记录，验证输出 ≤100 且 settle_date 降序
    • 测试文件: tests/test_data_fetchers/test_member_data_props.py
    • 验证: 需求 1.2, 1.4, 1.9, 3.5, 11.3, 11.4, 14.3

  • 2.5 编写属性测试：备注截断（P6）+ 备注数量限制（P17）

    • Property 6: 备注截断不变量 — 生成 0-2000 字符字符串，验证截断后 ≤500，超长以"…（已截断）"结尾
    • Property 17: 备注数量限制 — 生成 0-100 条备注，验证输出 ≤50 且 created_at 降序
    • 测试文件: tests/test_data_fetchers/test_member_data_props.py
    • 验证: 需求 4.4, 6.4, 14.4

  • 2.6 编写单元测试：member_data 边界情况

    • 客户无消费记录时返回空数组和默认值
    • 客户无备注时返回空数组
    • FDW 查询超时时抛出 TimeoutError
    • 会员不存在时 member_nickname 为空字符串
    • 测试文件: tests/test_data_fetchers/test_data_fetchers_unit.py
    • 需求: 1.10, 3.4, 6.6, 12.1, 12.2

• 3. 实现助教数据获取（assistant_data.py）

  • 3.1 实现 fetch_assistant_info(site_id, assistant_id)

    • 从 v_dim_assistant 获取花名、级别、入职日期
    • 从 v_dws_assistant_salary_calc 获取本月客户数、绩效档位
    • 计算 tenure_months（入职日期到当前日期）
    • 助教不存在时抛出 ValueError("assistant not found")
    • 需求: 2.1, 2.5, 2.6, 11.1, 11.2

  • 3.2 实现 fetch_service_history(site_id, assistant_id, member_id, months=3)

    • 从 v_dwd_assistant_service_log 获取服务记录，WHERE is_trash = false
    • 从 v_dws_member_assistant_relation_index 获取关系指数
    • 从 v_dws_member_assistant_intimacy 获取亲密度
    • 按 service_date DESC 排序
    • 需求: 2.2, 2.3, 2.4, 11.1, 11.2

  • 3.3 编写属性测试：废单排除（P5）

    • Property 5: 废单记录排除
    • 生成含 is_trash 标记的服务记录，验证输出不含 is_trash=true 的记录
    • 测试文件: tests/test_data_fetchers/test_assistant_data_props.py
    • 验证: 需求 2.3

  • 3.4 编写单元测试：assistant_data 边界情况

    • 助教不存在时抛出 ValueError
    • 无服务历史时返回空列表
    • FDW 查询超时时抛出 TimeoutError
    • 测试文件: tests/test_data_fetchers/test_data_fetchers_unit.py
    • 需求: 2.6, 4.5, 12.1

• 4. 检查点 — 数据获取层完成

  • 确保所有测试通过，ask the user if questions arise.

• 5. 实现应用 3 Prompt 拼接（app3_clue.py）

  • 5.1 改造 app3_clue.py 的 build_prompt() 为 async def

    • 新增 site_id 参数，调用 fetch_member_consumption_data() 获取真实消费数据
    • 组装 Prompt JSON：current_time、member_nickname、main_data、reference
    • reference 从 ai_cache 获取 app6_clues + 最近 2 套 app8_history
    • 空数据时标注"该客户暂无消费记录"
    • run() 中调用 build_prompt() 需 await
    • 需求: 3.1, 3.2, 3.3, 3.4, 3.5, 12.1, 12.2, 12.3

  • 5.2 编写属性测试：App3 Prompt 结构完整性（P7 局部）

    • Property 7（App3 部分）: build_prompt 返回结构完整性
    • Mock 数据获取返回随机数据，验证 Prompt JSON 包含 current_time、member_nickname、main_data、reference 顶层键
    • 测试文件: tests/test_ai_apps/test_build_prompt_props.py
    • 验证: 需求 3.1, 3.2

• 6. 实现应用 4 Prompt 拼接（app4_analysis.py）

  • 6.1 改造 app4_analysis.py 的 build_prompt() 为 async def

    • 使用 asyncio.gather 并发调用 fetch_assistant_info() + fetch_service_history() + fetch_member_consumption_data()
    • 调用 fetch_member_notes() 获取客户备注
    • 组装 Prompt JSON：current_time、assistant_info、service_history、task_assignment_basis、customer_data（含 system_data + notes）、reference
    • reference 从 ai_cache 获取 app8_current + 最近 2 套 app8_history
    • 部分数据获取失败时使用 return_exceptions=True 降级处理
    • 需求: 4.1, 4.2, 4.3, 4.4, 4.5, 4.6, 11.6, 12.1, 12.4

  • 6.2 编写属性测试：App4 Prompt 结构完整性（P7 局部）

    • Property 7（App4 部分）: build_prompt 返回结构完整性
    • Mock 数据获取返回随机数据，验证 Prompt JSON 包含 current_time、assistant_info、service_history、customer_data、reference 顶层键
    • 测试文件: tests/test_ai_apps/test_build_prompt_props.py
    • 验证: 需求 4.1, 4.2

• 7. 实现应用 5 Prompt 拼接（app5_tactics.py）

  • 7.1 改造 app5_tactics.py 的 build_prompt() 为 async def

    • 复用应用 4 的数据获取逻辑（fetch_assistant_info + fetch_service_history + fetch_member_consumption_data）
    • 从 context["app4_result"] 获取 task_suggestion，缺失时设为空对象
    • 组装 Prompt JSON：同 App4 结构 + task_suggestion
    • 需求: 5.1, 5.2, 5.3, 5.4

  • 7.2 编写属性测试：App5 task_suggestion 传递（P8）

    • Property 8: App5 task_suggestion 传递
    • 生成随机 app4_result（含空/缺失），验证 task_suggestion 字段正确传递或设为空对象
    • 测试文件: tests/test_ai_apps/test_build_prompt_props.py
    • 验证: 需求 5.3, 5.4

• 8. 实现应用 6 Prompt 拼接（app6_note.py）

  • 8.1 改造 app6_note.py 的 build_prompt() 为 async def
    • 调用 fetch_member_consumption_data() 获取消费数据
    • 调用 fetch_member_notes() 获取全部备注作为 all_notes
    • 组装 Prompt JSON：current_time、current_note、reference（含 member_nickname、consumption_data、all_notes、app3_clues、app8_history）
    • 空备注时 all_notes 设为空数组
    • 需求: 6.1, 6.2, 6.3, 6.4, 6.5, 6.6, 12.1, 12.2

• 9. 实现应用 7 Prompt 拼接（app7_customer.py）

  • 9.1 改造 app7_customer.py 的 build_prompt() 为 async def

    • 调用 fetch_member_consumption_data() 获取客观数据
    • 调用 fetch_member_notes() 获取备注作为 subjective_data.notes
    • 每条备注标注"【来源：{recorded_by}，请甄别信息真实性】"
    • 组装 Prompt JSON：current_time、member_id、member_nickname、objective_data、subjective_data、reference
    • 空备注时标注"该客户暂无主观备注信息"
    • 需求: 7.1, 7.2, 7.3, 7.4, 7.5, 7.6, 12.1, 12.2

  • 9.2 编写属性测试：App7 主观信息标注（P9）

    • Property 9: App7 主观信息来源标注
    • 生成随机备注（含 recorded_by），验证 Prompt 中每条备注附带"【来源：{recorded_by}，请甄别信息真实性】"
    • 测试文件: tests/test_ai_apps/test_build_prompt_props.py
    • 验证: 需求 7.4

• 10. 检查点 — 应用 3-7 Prompt 拼接完成

  • 确保所有测试通过，ask the user if questions arise.

• 11. 实现错误降级与 Token 预算控制

  • 11.1 在各应用 build_prompt() 中实现统一错误降级模式

    • 使用 asyncio.gather(return_exceptions=True) 捕获部分失败
    • 失败部分使用默认空值（空数组/空对象），附带提示文本
    • 确保 Prompt JSON 不含 None（Python）/ null（JSON），使用 default=str 处理 datetime/Decimal
    • 需求: 12.1, 12.2, 12.3, 12.4

  • 11.2 实现 Prompt 字符数限制

    • 应用 3/4/5/6/7 的 system message content ≤ 8000 字符
    • 消费记录超 100 条时截断并标注"仅展示最近 100 条，共 {total} 条"
    • 备注超 50 条时截断并标注"仅展示最近 50 条备注"
    • 单个文本字段不超过 1000 字符
    • 需求: 14.1, 14.3, 14.4, 14.5

  • 11.3 编写属性测试：错误降级（P14）+ Token 预算（P15）

    • Property 14: 错误降级产生合法 Prompt — 随机让 0-N 个数据获取函数失败，验证 build_prompt 返回合法 JSON，不含 null
    • Property 15: 应用 3-7 Prompt Token 预算 — 生成大量数据，验证 system message ≤ 8000 字符
    • 测试文件: tests/test_ai_apps/test_build_prompt_props.py
    • 验证: 需求 12.1, 12.2, 12.4, 14.1

• 12. 实现页面上下文文本化（page_context.py）

  • 12.1 实现 build_page_text(source_page, context_id, site_id, filters) 框架

    • 根据 source_page 路由到对应内部函数（_text_task_detail()、_text_customer_detail() 等）
    • 输出结构化中文描述文本（分段标题 + 缩进），非 JSON
    • 输出限制 2000 字符，超出截断并标注
    • 未识别的 source_page 返回空字符串
    • 数据获取失败返回"页面上下文获取失败，请直接描述您的问题"
    • 需求: 8.1, 8.2, 8.8, 8.9, 8.11

  • 12.2 实现详情类页面文本化函数

    • _text_task_detail(task_id, site_id) — 从 biz.coach_tasks + 会员信息 + 备注 + ai_cache 获取数据
    • _text_customer_detail(member_id, site_id) — 复用 fetch_member_consumption_data() + 维客线索
    • _text_coach_detail(assistant_id, site_id) — 复用 fetch_assistant_info() + 任务统计 + 备注
    • _text_customer_service_records(member_id, site_id) — 服务记录列表
    • _text_task_list(task_id, site_id) — 任务摘要 + 客户-助教关系
    • 不传入 member_phone 等敏感字段
    • 需求: 8.3, 8.4, 8.5, 8.12

  • 12.3 实现看板类与其他页面文本化函数

    • _text_board_finance(site_id, filters) — 财务 DWS 汇总，默认"本月"
    • _text_board_customer(site_id, filters) — 客户排名 top 列表
    • _text_board_coach(site_id, filters) — 助教排名
    • _text_performance(site_id, filters) — v_dws_assistant_salary_calc 绩效数据
    • _text_my_profile(site_id) — 用户信息 + 助教绑定
    • 看板未传筛选参数时使用默认值
    • 需求: 8.6, 8.7, 8.10

  • 12.4 编写属性测试：页面上下文长度（P10）+ 类型覆盖（P11）+ 敏感字段排除（P12）

    • Property 10: 页面上下文输出长度约束 — 生成大量数据，验证 build_page_text() 输出 ≤ 2000 字符
    • Property 11: 页面上下文覆盖所有页面类型 — 枚举 10 种类型，验证不抛异常且返回非空字符串
    • Property 12: 页面上下文不泄露敏感字段 — 生成含手机号的数据，验证输出不含 member_phone
    • 测试文件: tests/test_data_fetchers/test_page_context_props.py
    • 验证: 需求 8.8, 8.1, 8.2, 8.12

  • 12.5 编写单元测试：page_context 边界情况

    • 未识别 contextType 时返回空字符串
    • 看板无筛选参数时使用默认值
    • 数据获取失败时返回降级文本
    • task-detail 页面完整流程验证
    • 测试文件: tests/test_data_fetchers/test_data_fetchers_unit.py
    • 需求: 8.10, 8.11, 9.5

• 13. 实现应用 1 页面上下文集成（app1_chat.py）

  • 13.1 改造 app1_chat.py 的 _build_page_context() 为 async def

    • 调用 page_context.build_page_text(source_page, context_id, site_id, filters) 获取页面上下文文本
    • source_page 映射到 contextType，contextId 传入 context_id
    • 看板类页面从请求参数提取筛选参数传入 filters
    • contextType 为空或未识别时跳过页面上下文注入
    • 将返回文本作为 system prompt 中的 page_context 字段注入
    • 确保 biz_params.user_prompt_params（User_ID、Role、Nickname）注入后仍正确存在
    • system prompt 总字符数控制在 4000 以内
    • 需求: 9.1, 9.2, 9.3, 9.4, 9.5, 9.6, 14.2

  • 13.2 编写属性测试：biz_params 不变量（P13）+ App1 Token 预算（P16）

    • Property 13: biz_params 注入后不变量 — 生成随机用户信息，验证 user_prompt_params 包含 User_ID、Role、Nickname 且值一致
    • Property 16: 应用 1 System Prompt Token 预算 — 生成大量页面上下文，验证 system prompt ≤ 4000 字符
    • 测试文件: tests/test_ai_apps/test_app1_props.py
    • 验证: 需求 9.6, 13.2, 14.2

  • 13.3 编写单元测试：App1 页面上下文集成

    • contextType 为空时跳过页面上下文
    • task-detail 页面上下文注入后 system prompt 结构正确
    • 看板筛选参数正确传递到 build_page_text()
    • 页面上下文获取失败时使用降级文本
    • 测试文件: tests/test_ai_apps/test_ai_apps_unit.py
    • 需求: 9.1, 9.4, 9.5, 12.5

• 14. 检查点 — 页面上下文与应用 1 集成完成

  • 确保所有测试通过，ask the user if questions arise.

• 15. 前端看板筛选参数传递（小程序端）

  • 15.1 修改看板页面跳转 chat 的参数传递

    • board-finance 页面跳转时传入 timeDimension + areaFilter
    • board-customer 页面跳转时传入 dimension + typeFilter
    • board-coach 页面跳转时传入 dimension + projectFilter + timeDimension
    • 将筛选参数编码为 chat 页面 URL 查询参数
    • 需求: 10.1, 10.2, 10.3, 10.4

  • 15.2 后端从请求中提取看板筛选参数

    • 在 chat 消息接口中解析 URL 查询参数中的筛选参数
    • 将筛选参数传入 build_page_text() 的 filters 参数
    • 需求: 10.4, 9.4

• 16. 集成连线与最终验证

  • 16.1 确保 dispatcher 调用链中 build_prompt() 的 await 正确

    • 消费事件链：App3 → App8 → App7 + App4 → App5，所有 build_prompt 调用加 await
    • 备注事件链：App6 → App8，build_prompt 调用加 await
    • 验证 run() 方法中 build_prompt() 调用已改为 await
    • 需求: 3.1, 4.1, 5.1, 6.1, 7.1, 11.6

  • 16.2 编写单元测试：各应用完整流程

    • App3 完整流程：Mock 数据 → build_prompt → 验证 JSON 结构和内容
    • App4 并发数据获取：验证 asyncio.gather 调用
    • App5 task_suggestion 传递：验证 context["app4_result"] 正确传入
    • App7 主观信息标注：验证备注标注格式
    • RLS 隔离：验证 get_etl_readonly_connection 被调用且传入正确 site_id
    • 测试文件: tests/test_ai_apps/test_ai_apps_unit.py
    • 需求: 3.1, 4.1, 5.3, 7.4, 11.2, 13.1

• 17. 最终检查点 — 全部完成

  • 确保所有测试通过，ask the user if questions arise.

备注

• 标记 * 的任务为可选，可跳过以加速 MVP
• 每个任务引用具体需求编号，确保可追溯
• 属性测试验证设计文档中的 17 个正确性属性（P1-P17）
• 检查点确保增量验证
• 测试文件组织：tests/test_data_fetchers/（数据获取层）、tests/test_ai_apps/（应用层）