314 lines
12 KiB
Python
314 lines
12 KiB
Python
"""需求理解助手路由"""
|
||
import json
|
||
import os
|
||
import uuid
|
||
from typing import List
|
||
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Form
|
||
from fastapi.responses import StreamingResponse
|
||
from sqlalchemy.orm import Session
|
||
|
||
from database import get_db
|
||
from models.user import User
|
||
from models.conversation import Conversation, Message
|
||
from schemas.conversation import (
|
||
RequirementAnalyzeRequest, ConversationResponse,
|
||
ConversationDetail, MessageResponse,
|
||
)
|
||
from routers.auth import get_current_user
|
||
from services.ai_service import ai_service
|
||
from config import UPLOAD_DIR
|
||
|
||
router = APIRouter()
|
||
|
||
REQUIREMENT_SYSTEM_PROMPT = """# 角色定义
|
||
你同时具备两个身份：
|
||
1. **资深产品经理（10年+）**：擅长从模糊信息中提炼需求本质，精通用户故事、MECE拆解、优先级排序、验收标准制定
|
||
2. **高级全栈程序员（10年+）**：做过大量项目，对功能的复杂度、开发工作量有精准直觉，能一眼看出哪些需求是"看似简单实则巨坑"
|
||
|
||
你服务的对象是**程序员**，他们会把甲方发来的原始内容（口语化描述、语音转文字、截图、聊天记录、需求文档等）发给你。你需要站在「既懂业务又懂技术」的双重视角，将其转化为**清晰明确、可直接进入开发的结构化需求**。
|
||
|
||
> ⚠️ 本助手专注于**需求理解与分析**。技术选型、数据库设计、API设计、架构方案等请移步「架构选型AI助手」。
|
||
|
||
# 核心理念
|
||
- **需求层面**：甲方说的不一定是他真正想要的，你要透过表面描述挖掘真实诉求
|
||
- **程序员直觉**：每个功能你都要心里过一遍复杂度，标注哪些"看起来简单但实际很复杂"
|
||
- **落地导向**：不出空中楼阁式的分析，每条功能都要能对应到具体的开发任务
|
||
|
||
# 分析框架
|
||
收到用户输入后，按以下框架进行系统性分析：
|
||
|
||
## 第一步：需求还原（产品经理视角）
|
||
- 理解甲方的**核心意图**：到底想解决什么问题？服务什么业务场景？
|
||
- 识别**目标用户**是谁，使用频率、核心使用路径是什么
|
||
- 区分"真实需求"与"表面描述"，找出甲方没说但一定需要的**隐性需求**
|
||
- 判断产品定位：工具型/平台型/内容型？ToB/ToC？
|
||
|
||
## 第二步：功能拆解（遵循 MECE 原则）
|
||
将需求拆解为相互独立、完全穷尽的功能模块，每个功能需包含：
|
||
- 功能名称 + 具体说明
|
||
- 优先级（P0 核心必做 / P1 重要 / P2 锦上添花）
|
||
- 涉及的用户角色
|
||
- 复杂度预判（简单/中等/复杂）+ 复杂原因说明
|
||
|
||
## 第三步：用户故事与验收标准
|
||
将核心功能转写为标准用户故事：
|
||
> 作为【角色】，我希望【功能】，以便【价值/目的】
|
||
|
||
为 P0 功能补充验收标准（AC），采用 Given-When-Then 格式：
|
||
> 假设【前置条件】，当【用户操作】，那么【系统响应】
|
||
|
||
## 第四步：复杂度预警（程序员视角）
|
||
基于你做过大量项目的经验，标注：
|
||
- **隐藏复杂度**：哪些功能看似简单但实现起来有坑（如"支持实时聊天"看似一句话，实际涉及WebSocket、消息队列、已读回执等）
|
||
- **高风险功能**：涉及支付、权限、数据一致性等需要特别慎重的功能
|
||
- **工期杀手**：容易严重超出预期工时的功能，提前预警
|
||
- **工期粗估**：按模块给出大致工时范围（x-x天），帮助程序员评估排期
|
||
|
||
## 第五步：边界与待确认项
|
||
- 需求中**含糊不清**需要甲方确认的关键问题
|
||
- 容易遗漏的**边缘场景**（空状态、异常流、权限边界、数据量极值、并发操作）
|
||
- 甲方可能还没想到但**一定会追加**的需求（基于经验预判）
|
||
|
||
# 输出规范
|
||
严格使用以下 Markdown 结构输出：
|
||
|
||
---
|
||
|
||
## 📋 需求概述
|
||
> 用2-3句话概括：这是什么产品/功能，解决谁的什么问题，核心价值是什么。
|
||
> 产品定位：【工具型/平台型/内容型】 | 【ToB/ToC】
|
||
|
||
## 👥 用户角色
|
||
| 角色 | 说明 | 关键诉求 | 使用频率 |
|
||
|------|------|---------|---------|
|
||
|
||
## 🧩 功能清单
|
||
| 优先级 | 模块 | 功能项 | 功能说明 | 涉及角色 | 复杂度 |
|
||
|--------|------|--------|---------|---------|--------|
|
||
| P0 | xxx | xxx | xxx | xxx | 🔴复杂 - 原因 |
|
||
| P0 | xxx | xxx | xxx | xxx | 🟢简单 |
|
||
| P1 | xxx | xxx | xxx | xxx | 🟡中等 |
|
||
|
||
## 📖 核心用户故事与验收标准
|
||
### US-1: 【用户故事标题】
|
||
- **故事**：作为【角色】，我希望【功能】，以便【价值】
|
||
- **验收标准**：
|
||
- ✅ 假设【条件】，当【操作】，那么【预期结果】
|
||
- ✅ 假设【条件】，当【异常操作】，那么【兜底处理】
|
||
|
||
## ⚡ 复杂度预警（程序员必看）
|
||
### 🔴 隐藏复杂度
|
||
1. **【功能名】**：看似xxx，实际需要xxx，建议xxx
|
||
|
||
### ⏱️ 工期粗估
|
||
| 模块 | 工时范围 | 说明 |
|
||
|------|---------|------|
|
||
| 合计 | x-x天 | 基于1名全栈开发者，含开发+自测 |
|
||
|
||
### 🔮 甲方大概率会追加的需求
|
||
1. 【需求名】— 理由：基于经验，做了xxx之后甲方通常会要求xxx
|
||
|
||
## ❓ 待确认问题
|
||
> 以下问题会直接影响开发方案，建议优先与甲方确认：
|
||
1. **【问题】**：【为什么需要确认】→ 不确认的影响：【xxx】
|
||
|
||
## 💡 风险提示
|
||
- 【业务风险、边缘场景、容易遗漏的点等】
|
||
|
||
---
|
||
|
||
> 💡 **下一步建议**：需求确认后，可将功能清单发送至「架构选型AI助手」，获取技术选型、数据库设计、API接口设计等技术方案。
|
||
|
||
# 交互原则
|
||
1. **首次分析要全面**：即使信息不完整，也要基于已有信息给出最完整的分析，用待确认问题标注不确定的部分
|
||
2. **复杂度标注要诚实**：程序员最怕"这个很简单"，要如实标注复杂度和潜在的坑
|
||
3. **追问要有价值**：只追问影响开发方案的关键问题，不问显而易见的
|
||
4. **语言贴近程序员**：用开发者能直接理解的术语，避免纯商业话术
|
||
5. **持续迭代**：用户补充信息后，在之前分析的基础上更新完善，而非重头开始
|
||
6. **务实不空谈**：每条建议都基于实际经验，不说"建议优化用户体验"这类空话
|
||
7. **工时要诚实**：给出合理区间，标注不确定因素"""
|
||
|
||
|
||
@router.post("/upload-image")
|
||
async def upload_image(
|
||
file: UploadFile = File(...),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""上传图片"""
|
||
# 验证文件类型
|
||
allowed_types = ["image/jpeg", "image/png", "image/gif", "image/webp"]
|
||
if file.content_type not in allowed_types:
|
||
raise HTTPException(status_code=400, detail="不支持的文件类型")
|
||
|
||
# 生成唯一文件名
|
||
ext = file.filename.split(".")[-1] if "." in file.filename else "png"
|
||
filename = f"{uuid.uuid4().hex}.{ext}"
|
||
filepath = os.path.join(UPLOAD_DIR, filename)
|
||
|
||
# 保存文件
|
||
content = await file.read()
|
||
with open(filepath, "wb") as f:
|
||
f.write(content)
|
||
|
||
return {"url": f"/uploads/{filename}"}
|
||
|
||
|
||
@router.get("/conversations", response_model=List[ConversationResponse])
|
||
def get_conversations(
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""获取需求对话列表"""
|
||
conversations = (
|
||
db.query(Conversation)
|
||
.filter(Conversation.user_id == current_user.id, Conversation.type == "requirement")
|
||
.order_by(Conversation.updated_at.desc())
|
||
.all()
|
||
)
|
||
return [ConversationResponse.model_validate(c) for c in conversations]
|
||
|
||
|
||
@router.get("/conversations/{conversation_id}", response_model=ConversationDetail)
|
||
def get_conversation_detail(
|
||
conversation_id: int,
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""获取对话详情"""
|
||
conv = db.query(Conversation).filter(
|
||
Conversation.id == conversation_id,
|
||
Conversation.user_id == current_user.id,
|
||
).first()
|
||
if not conv:
|
||
raise HTTPException(status_code=404, detail="对话不存在")
|
||
|
||
messages = (
|
||
db.query(Message)
|
||
.filter(Message.conversation_id == conversation_id)
|
||
.order_by(Message.created_at.asc())
|
||
.all()
|
||
)
|
||
result = ConversationDetail.model_validate(conv)
|
||
result.messages = [MessageResponse.model_validate(m) for m in messages]
|
||
return result
|
||
|
||
|
||
@router.post("/analyze")
|
||
async def analyze_requirement(
|
||
request: RequirementAnalyzeRequest,
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""分析需求 - 流式输出"""
|
||
# 创建或获取对话
|
||
if request.conversation_id:
|
||
conv = db.query(Conversation).filter(
|
||
Conversation.id == request.conversation_id,
|
||
Conversation.user_id == current_user.id,
|
||
).first()
|
||
if not conv:
|
||
raise HTTPException(status_code=404, detail="对话不存在")
|
||
else:
|
||
conv = Conversation(
|
||
user_id=current_user.id,
|
||
title=request.content[:50] if request.content else "新需求分析",
|
||
type="requirement",
|
||
)
|
||
db.add(conv)
|
||
db.commit()
|
||
db.refresh(conv)
|
||
|
||
# 保存用户消息
|
||
user_msg = Message(
|
||
conversation_id=conv.id,
|
||
role="user",
|
||
content=request.content,
|
||
image_urls=json.dumps(request.image_urls) if request.image_urls else "",
|
||
)
|
||
db.add(user_msg)
|
||
db.commit()
|
||
|
||
# 构建历史消息
|
||
history_msgs = (
|
||
db.query(Message)
|
||
.filter(Message.conversation_id == conv.id)
|
||
.order_by(Message.created_at.asc())
|
||
.all()
|
||
)
|
||
|
||
messages = []
|
||
for msg in history_msgs:
|
||
if msg.role == "user":
|
||
content = msg.content
|
||
# 如果有图片，添加图片描述提示
|
||
if msg.image_urls:
|
||
try:
|
||
urls = json.loads(msg.image_urls)
|
||
if urls:
|
||
content += f"\n\n[用户上传了{len(urls)}张图片]"
|
||
except json.JSONDecodeError:
|
||
pass
|
||
messages.append({"role": "user", "content": content})
|
||
else:
|
||
messages.append({"role": "assistant", "content": msg.content})
|
||
|
||
# 确定任务类型
|
||
task_type = "multimodal" if request.image_urls else "reasoning"
|
||
|
||
# 流式调用AI
|
||
async def generate():
|
||
full_response = ""
|
||
try:
|
||
result = await ai_service.chat(
|
||
task_type=task_type,
|
||
messages=messages,
|
||
system_prompt=REQUIREMENT_SYSTEM_PROMPT,
|
||
stream=True,
|
||
model_config_id=request.model_config_id,
|
||
)
|
||
if isinstance(result, str):
|
||
# 非流式返回
|
||
full_response = result
|
||
yield f"data: {json.dumps({'content': result, 'done': False})}\n\n"
|
||
else:
|
||
async for chunk in result:
|
||
full_response += chunk
|
||
yield f"data: {json.dumps({'content': chunk, 'done': False})}\n\n"
|
||
except Exception as e:
|
||
error_msg = f"AI调用出错: {str(e)}"
|
||
full_response = error_msg
|
||
yield f"data: {json.dumps({'content': error_msg, 'done': False})}\n\n"
|
||
|
||
# 保存AI回复
|
||
ai_msg = Message(
|
||
conversation_id=conv.id,
|
||
role="assistant",
|
||
content=full_response,
|
||
)
|
||
db.add(ai_msg)
|
||
db.commit()
|
||
|
||
yield f"data: {json.dumps({'content': '', 'done': True, 'conversation_id': conv.id})}\n\n"
|
||
|
||
return StreamingResponse(generate(), media_type="text/event-stream")
|
||
|
||
|
||
@router.delete("/conversations/{conversation_id}")
|
||
def delete_conversation(
|
||
conversation_id: int,
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""删除对话"""
|
||
conv = db.query(Conversation).filter(
|
||
Conversation.id == conversation_id,
|
||
Conversation.user_id == current_user.id,
|
||
).first()
|
||
if not conv:
|
||
raise HTTPException(status_code=404, detail="对话不存在")
|
||
|
||
db.query(Message).filter(Message.conversation_id == conversation_id).delete()
|
||
db.delete(conv)
|
||
db.commit()
|
||
return {"message": "删除成功"}
|