239 lines
8.3 KiB
Python
239 lines
8.3 KiB
Python
# core/requirement_analyzer.py - 需求分解 & 函数签名生成
|
||
import re
|
||
from typing import List, Optional
|
||
|
||
import config
|
||
from core.llm_client import LLMClient
|
||
from database.models import FunctionalRequirement
|
||
|
||
|
||
class RequirementAnalyzer:
|
||
"""
|
||
使用 LLM 将原始需求分解为功能需求列表，并生成函数接口签名。
|
||
支持注入知识库上下文以提升分解质量。
|
||
"""
|
||
|
||
def __init__(self, llm_client: Optional[LLMClient] = None):
|
||
"""
|
||
初始化需求分析器
|
||
|
||
Args:
|
||
llm_client: LLM 客户端实例，为 None 时自动创建
|
||
"""
|
||
self.llm = llm_client or LLMClient()
|
||
|
||
# ══════════════════════════════════════════════════
|
||
# 需求分解
|
||
# ══════════════════════════════════════════════════
|
||
|
||
def decompose(
|
||
self,
|
||
raw_requirement: str,
|
||
project_id: int,
|
||
raw_req_id: int,
|
||
knowledge: str = "",
|
||
) -> List[FunctionalRequirement]:
|
||
"""
|
||
将原始需求分解为功能需求列表
|
||
|
||
Args:
|
||
raw_requirement: 原始需求文本
|
||
project_id: 所属项目 ID
|
||
raw_req_id: 原始需求记录 ID
|
||
knowledge: 知识库文本（可选）
|
||
|
||
Returns:
|
||
FunctionalRequirement 对象列表（未持久化，id=None）
|
||
|
||
Raises:
|
||
ValueError: LLM 返回格式不合法
|
||
json.JSONDecodeError: JSON 解析失败
|
||
"""
|
||
knowledge_section = self._build_knowledge_section(knowledge)
|
||
prompt = config.DECOMPOSE_PROMPT_TEMPLATE.format(
|
||
knowledge_section=knowledge_section,
|
||
raw_requirement=raw_requirement,
|
||
)
|
||
|
||
result = self.llm.chat_json(
|
||
system_prompt="你是一位资深软件架构师，擅长需求分析与系统设计。",
|
||
user_prompt=prompt,
|
||
)
|
||
|
||
items = result.get("functional_requirements", [])
|
||
if not items:
|
||
raise ValueError("LLM 未返回任何功能需求，请检查原始需求描述")
|
||
|
||
requirements = []
|
||
for item in items:
|
||
req = FunctionalRequirement(
|
||
project_id=project_id,
|
||
raw_req_id=raw_req_id,
|
||
index_no=int(item.get("index", len(requirements) + 1)),
|
||
title=item.get("title", "未命名功能"),
|
||
description=item.get("description", ""),
|
||
function_name=self._sanitize_function_name(
|
||
item.get("function_name", f"func_{len(requirements)+1}")
|
||
),
|
||
priority=item.get("priority", "medium"),
|
||
)
|
||
requirements.append(req)
|
||
|
||
return requirements
|
||
|
||
# ══════════════════════════════════════════════════
|
||
# 函数签名生成（新增）
|
||
# ══════════════════════════════════════════════════
|
||
|
||
def build_function_signature(
|
||
self,
|
||
func_req: FunctionalRequirement,
|
||
knowledge: str = "",
|
||
) -> dict:
|
||
"""
|
||
为单个功能需求生成函数接口签名 JSON
|
||
|
||
Args:
|
||
func_req: 功能需求对象（需含有效 id）
|
||
knowledge: 知识库文本（可选）
|
||
|
||
Returns:
|
||
符合接口规范的 dict，包含 name/requirement_id/description/
|
||
type/parameters/return 字段
|
||
|
||
Raises:
|
||
json.JSONDecodeError: LLM 返回非合法 JSON
|
||
"""
|
||
requirement_id = self._format_requirement_id(func_req.index_no)
|
||
knowledge_section = self._build_knowledge_section(knowledge)
|
||
|
||
prompt = config.FUNC_SIGNATURE_PROMPT_TEMPLATE.format(
|
||
knowledge_section=knowledge_section,
|
||
requirement_id=requirement_id,
|
||
title=func_req.title,
|
||
function_name=func_req.function_name,
|
||
description=func_req.description,
|
||
)
|
||
|
||
signature = self.llm.chat_json(
|
||
system_prompt=(
|
||
"你是一位资深软件架构师，专注于 API 接口设计。"
|
||
"只输出合法 JSON，不添加任何说明文字。"
|
||
),
|
||
user_prompt=prompt,
|
||
)
|
||
|
||
# 确保关键字段存在，做兜底处理
|
||
signature.setdefault("name", func_req.function_name)
|
||
signature.setdefault("requirement_id", requirement_id)
|
||
signature.setdefault("description", func_req.description)
|
||
signature.setdefault("type", "function")
|
||
signature.setdefault("parameters", {})
|
||
signature.setdefault("return", {"type": "void", "description": ""})
|
||
|
||
return signature
|
||
|
||
def build_function_signatures_batch(
|
||
self,
|
||
func_reqs: List[FunctionalRequirement],
|
||
knowledge: str = "",
|
||
on_progress=None,
|
||
) -> List[dict]:
|
||
"""
|
||
批量为功能需求列表生成函数接口签名
|
||
|
||
Args:
|
||
func_reqs: 功能需求列表
|
||
knowledge: 知识库文本（可选）
|
||
on_progress: 进度回调 fn(index, total, func_req, signature, error)
|
||
|
||
Returns:
|
||
签名 dict 列表，顺序与 func_reqs 一致；
|
||
生成失败的条目使用降级结构填充，不中断整体流程
|
||
"""
|
||
results = []
|
||
total = len(func_reqs)
|
||
|
||
for i, req in enumerate(func_reqs):
|
||
try:
|
||
sig = self.build_function_signature(req, knowledge)
|
||
results.append(sig)
|
||
if on_progress:
|
||
on_progress(i + 1, total, req, sig, None)
|
||
except Exception as e:
|
||
# 降级：用基础信息填充，保证 JSON 完整性
|
||
fallback = self._build_fallback_signature(req)
|
||
results.append(fallback)
|
||
if on_progress:
|
||
on_progress(i + 1, total, req, fallback, e)
|
||
|
||
return results
|
||
|
||
# ══════════════════════════════════════════════════
|
||
# 私有工具方法
|
||
# ══════════════════════════════════════════════════
|
||
|
||
@staticmethod
|
||
def _build_knowledge_section(knowledge: str) -> str:
|
||
"""构建知识库 Prompt 段落"""
|
||
if not knowledge or not knowledge.strip():
|
||
return ""
|
||
return f"""## 参考知识库
|
||
{knowledge}
|
||
|
||
---
|
||
"""
|
||
|
||
@staticmethod
|
||
def _sanitize_function_name(name: str) -> str:
|
||
"""
|
||
清理函数名，确保符合 snake_case 规范
|
||
|
||
Args:
|
||
name: 原始函数名
|
||
|
||
Returns:
|
||
合法的 snake_case 函数名
|
||
"""
|
||
name = re.sub(r"[^a-zA-Z0-9_]", "_", name).lower()
|
||
name = re.sub(r"_+", "_", name).strip("_")
|
||
if name and name[0].isdigit():
|
||
name = "func_" + name
|
||
return name or "unnamed_function"
|
||
|
||
@staticmethod
|
||
def _format_requirement_id(index_no: int) -> str:
|
||
"""
|
||
将序号格式化为需求编号字符串
|
||
|
||
Args:
|
||
index_no: 功能需求序号（从 1 开始）
|
||
|
||
Returns:
|
||
格式化编号，如 'REQ.01'、'REQ.12'
|
||
"""
|
||
return f"REQ.{index_no:02d}"
|
||
|
||
@staticmethod
|
||
def _build_fallback_signature(func_req: FunctionalRequirement) -> dict:
|
||
"""
|
||
构建降级签名（LLM 调用失败时使用）
|
||
|
||
Args:
|
||
func_req: 功能需求对象
|
||
|
||
Returns:
|
||
包含基础信息的签名 dict
|
||
"""
|
||
return {
|
||
"name": func_req.function_name,
|
||
"requirement_id": f"REQ.{func_req.index_no:02d}",
|
||
"description": func_req.description,
|
||
"type": "function",
|
||
"parameters": {},
|
||
"return": {
|
||
"type": "void",
|
||
"description": "TODO: define return value"
|
||
},
|
||
"_note": "Auto-generated fallback due to LLM error"
|
||
} |