RAG/utils/code_prompt_manager.py

726 lines
28 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
代码Prompt管理模块
用于生成和管理代码相关问答的专属Prompt模板
支持基于代码意图分类的动态Prompt选择和生成
"""
import json
import re
import sys
import os
from typing import Dict, Any, Optional, List, Tuple
from enum import Enum
from loguru import logger
# 添加项目根目录到 Python 模块搜索路径
current_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.dirname(current_dir)
if project_root not in sys.path:
sys.path.insert(0, project_root)
from utils.query_processor import CodeIntentCategory, PromptTemplateType
from utils.prompt import (
INTENT_DETECTION_TEMPLATE,
CODE_EXPLANATION_TEMPLATE,
CODE_DEBUGGING_TEMPLATE,
CODE_GENERATION_TEMPLATE,
ALGORITHM_EXPLANATION_TEMPLATE,
CODE_OPTIMIZATION_TEMPLATE,
GENERAL_QA_TEMPLATE,
)
class CodePromptManager:
"""代码Prompt管理类"""
def __init__(self):
"""
初始化代码Prompt管理器
"""
self._templates = self._load_templates()
self._context_cache: Dict[str, List[Dict[str, str]]] = {}
logger.info("代码Prompt管理器初始化完成")
def _load_templates(self) -> Dict[PromptTemplateType, str]:
"""
加载Prompt模板
Returns:
Dict[PromptTemplateType, str]: Prompt模板字典
"""
return {
PromptTemplateType.CODE_EXPLANATION: CODE_EXPLANATION_TEMPLATE,
PromptTemplateType.CODE_DEBUGGING: CODE_DEBUGGING_TEMPLATE,
PromptTemplateType.CODE_GENERATION: CODE_GENERATION_TEMPLATE,
PromptTemplateType.ALGORITHM_EXPLANATION: ALGORITHM_EXPLANATION_TEMPLATE,
PromptTemplateType.CODE_OPTIMIZATION: CODE_OPTIMIZATION_TEMPLATE,
PromptTemplateType.GENERAL_QA: GENERAL_QA_TEMPLATE
}
def _map_intent_to_prompt_type(self, intent_category: str) -> PromptTemplateType:
"""
将代码意图分类映射到Prompt类型
Args:
intent_category: 代码意图分类(字符串)
Returns:
PromptTemplateType: 对应的Prompt类型
"""
mapping = {
# 代码解释与逻辑类 -> CODE_EXPLANATION
"logic_explanation": PromptTemplateType.CODE_EXPLANATION,
"entity_introduction": PromptTemplateType.CODE_EXPLANATION,
"code_structure": PromptTemplateType.CODE_EXPLANATION,
# 代码生成与实现类 -> CODE_GENERATION
"code_generation": PromptTemplateType.CODE_GENERATION,
"boilerplate_implementation": PromptTemplateType.CODE_GENERATION,
# 调试、优化与理论类
"error_debugging": PromptTemplateType.CODE_DEBUGGING,
"code_optimization": PromptTemplateType.CODE_OPTIMIZATION,
"algorithm_theory": PromptTemplateType.ALGORITHM_EXPLANATION,
# 非代码问题 -> GENERAL_QA
"general_technical": PromptTemplateType.GENERAL_QA,
"non_technical": PromptTemplateType.GENERAL_QA,
"unknown": PromptTemplateType.GENERAL_QA
}
return mapping.get(intent_category, PromptTemplateType.GENERAL_QA)
def _build_conversation_history(self, history: Optional[Any]) -> str:
"""
构建对话历史字符串
Args:
history: 对话历史,可以是字符串或字典列表
Returns:
str: 格式化的对话历史
"""
if not history:
return ""
# 如果是字符串,直接返回
if isinstance(history, str):
return history
# 如果是字典列表,格式化为字符串
if isinstance(history, list):
history_str = []
for item in history:
if isinstance(item, dict):
role = item.get('role', 'user')
content = item.get('content', '')
if role == 'user':
history_str.append(f"用户: {content}")
else:
history_str.append(f"助手: {content}")
return "\n".join(history_str)
# 其他类型,转换为字符串
return str(history)
def _extract_code_from_context(self, code_context: str) -> str:
"""
从上下文中提取代码
Args:
code_context: 代码上下文
Returns:
str: 提取的代码
"""
if not code_context:
return ""
# 尝试提取代码块
code_blocks = re.findall(r'```[\w]*\n[\s\S]*?```', code_context)
if code_blocks:
# 提取所有代码块并合并
extracted_code = []
for block in code_blocks:
# 提取语言标记
lang_match = re.match(r'```([\w]*)\n', block)
language = lang_match.group(1) if lang_match else ""
# 去除代码块标记
code = re.sub(r'```[\w]*\n|```', '', block)
code = code.strip()
if code:
if language:
extracted_code.append(f"语言: {language}\n{code}")
else:
extracted_code.append(code)
return "\n\n".join(extracted_code)
# 如果没有代码块标记,尝试提取看起来像代码的部分
# 查找连续的多行代码(以缩进或常见代码关键字开头)
lines = code_context.split('\n')
code_lines = []
in_code = False
for line in lines:
# 检查是否是代码行
line_stripped = line.strip()
if (line_stripped and
(line.startswith(' ') or line.startswith('\t') or # 缩进
line_stripped.startswith('def ') or line_stripped.startswith('class ') or # Python关键字
line_stripped.startswith('import ') or line_stripped.startswith('from ') or # 导入
line_stripped.startswith('if ') or line_stripped.startswith('for ') or # 控制流
line_stripped.startswith('while ') or line_stripped.startswith('try ') or
line_stripped.startswith('except ') or line_stripped.startswith('finally ') or
line_stripped.startswith('return ') or line_stripped.startswith('print(') or
line_stripped.startswith('// ') or line_stripped.startswith('# ') or # 注释
line_stripped.endswith(';') or # 分号结尾如Java、C++等)
line_stripped.startswith('{') or line_stripped.startswith('}') or # 大括号
re.match(r'^[\w_]+\s*=\s*', line_stripped) or # 变量赋值
re.match(r'^[\w_]+\s*\(.*\)\s*\{{?', line_stripped))): # 函数定义
code_lines.append(line)
in_code = True
elif in_code and line.strip() == '':
# 保留代码中的空行
code_lines.append(line)
elif in_code and len(code_lines) > 3:
# 如果已经收集了多行代码,并且遇到非代码行,停止收集
break
else:
# 非代码行,重置
code_lines = []
in_code = False
if len(code_lines) > 3:
return "\n".join(code_lines)
return code_context
def _format_code_block(self, code: str, language: str = "") -> str:
"""
格式化代码块,提高显示质量
Args:
code: 代码内容
language: 代码语言
Returns:
str: 格式化的代码块
"""
if not code:
return ""
# 添加语言标记
lang_tag = language if language else ""
# 确保代码块格式正确
formatted_code = f"```{lang_tag}\n{code}\n```"
return formatted_code
def _build_enhanced_context(self, retrieved_results: List[Dict[str, Any]], intent_result: Optional[Dict[str, Any]]) -> str:
"""
根据意图和检索结果构建增强的上下文
Args:
retrieved_results: 检索结果列表,每个元素包含 id、text 和 metadata
intent_result: 意图识别结果
Returns:
str: 增强的上下文
"""
if not retrieved_results:
return "未找到相关参考信息"
context_parts = []
intent = intent_result.get('intent', '') if intent_result else ''
for result in retrieved_results:
metadata = result.get('metadata', {})
text = result.get('text', '')
result_id = result.get('id', 1)
# 提取所有 metadata 字段
func_id = metadata.get('func_id', '')
func_name = metadata.get('func_name', '')
class_name = metadata.get('class_name', 'None')
file_path = metadata.get('file_path', '')
lang = metadata.get('lang', '')
params = metadata.get('params', 0)
return_type = metadata.get('return_type', 'None')
docstring = metadata.get('docstring', '')
start_line = metadata.get('start_line', '')
end_line = metadata.get('end_line', '')
repo_id = metadata.get('repo_id', '')
branch = metadata.get('branch', '')
func_body = metadata.get('func_body', '')
# 根据意图构建不同的上下文
if intent == "code_understanding":
# 代码理解意图,强调语言、函数名、类名、参数、返回类型和函数体
context_part = f"【参考信息{result_id}】这是由{lang}实现的函数{func_name}"
if class_name and class_name != "None":
context_part += f",属于{class_name}"
context_part += f",它接收{params}个参数,返回类型为{return_type}"
if docstring:
context_part += f"。函数说明:{docstring}"
context_part += f"\n文件路径:{file_path},位置:{start_line}-{end_line}\n"
context_part += f"具体实现:\n{func_body}\n"
context_part += f"仓库:{repo_id},分支:{branch}\n"
context_part += f"原始文本:\n{text}"
elif intent == "code_modification":
# 代码修改意图,强调文件路径、位置和函数体
context_part = f"【参考信息{result_id}】需要修改的代码位于文件:{file_path},位置:{start_line}-{end_line}"
context_part += f"\n函数名:{func_name}"
if class_name and class_name != "None":
context_part += f"{class_name}类)"
context_part += f",由{lang}实现\n"
context_part += f"函数签名:接收{params}个参数,返回类型为{return_type}\n"
if docstring:
context_part += f"函数说明:{docstring}\n"
context_part += f"具体实现:\n{func_body}\n"
context_part += f"仓库:{repo_id},分支:{branch}\n"
context_part += f"原始文本:\n{text}"
elif intent == "functionality_question":
# 功能询问意图,强调函数名、文档、参数和返回类型
context_part = f"【参考信息{result_id}】函数{func_name}"
if class_name and class_name != "None":
context_part += f"{class_name}类)"
context_part += f"的功能说明:\n{docstring}\n"
context_part += f"{lang}实现,接收{params}个参数,返回类型为{return_type}\n"
context_part += f"文件路径:{file_path},位置:{start_line}-{end_line}\n"
context_part += f"具体实现:\n{func_body}\n"
context_part += f"仓库:{repo_id},分支:{branch}\n"
context_part += f"原始文本:\n{text}"
else:
# 其他意图,综合所有信息
context_part = f"【参考信息{result_id}】(来源:{file_path}"
context_part += f"\n函数:{func_name}"
if class_name and class_name != "None":
context_part += f"{class_name}类)"
context_part += f",语言:{lang}\n"
context_part += f"参数:{params}个,返回类型:{return_type}\n"
if docstring:
context_part += f"说明:{docstring}\n"
context_part += f"位置:{start_line}-{end_line}\n"
context_part += f"仓库:{repo_id},分支:{branch}\n"
context_part += f"实现:\n{func_body}\n"
context_part += f"原始文本:\n{text}"
context_parts.append(context_part)
return "\n\n".join(context_parts)
def generate_prompt(
self,
user_query: str,
intent_category: CodeIntentCategory,
code_context: Optional[str] = None,
conversation_history: Optional[List[Dict[str, str]]] = None,
error_message: Optional[str] = None,
target_language: Optional[str] = None,
user_requirement: Optional[str] = None
) -> str:
"""
生成代码专用Prompt
Args:
user_query: 用户问题
intent_category: 代码意图分类
code_context: 代码上下文
conversation_history: 对话历史
error_message: 错误信息仅Bug修复场景
target_language: 目标编程语言(仅代码生成场景)
user_requirement: 用户需求(仅代码生成场景)
Returns:
str: 生成的Prompt
"""
try:
logger.info(f"生成代码Prompt意图分类: {intent_category}")
# 映射意图到Prompt类型
prompt_type = self._map_intent_to_prompt_type(intent_category)
logger.info(f"选择Prompt类型: {prompt_type}")
# 获取对应模板
template = self._templates.get(prompt_type)
if not template:
logger.warning(f"未找到对应Prompt模板: {prompt_type}")
template = self._templates[PromptTemplateType.GENERAL_QA]
# 准备参数
params = {
"user_query": user_query,
"code_context": code_context or "",
"conversation_history": self._build_conversation_history(conversation_history),
"error_message": error_message or "",
"target_language": target_language or "根据上下文判断",
"user_requirement": user_requirement or user_query,
"algorithm_code": self._extract_code_from_context(code_context) if code_context else ""
}
# 填充模板
prompt = template
for key, value in params.items():
placeholder = f"{{{key}}}"
prompt = prompt.replace(placeholder, value)
logger.info(f"Prompt生成完成长度: {len(prompt)}字符")
return prompt
except Exception as e:
logger.error(f"生成Prompt失败: {e}")
# 返回通用模板
return self._templates[PromptTemplateType.GENERAL_QA].format(
user_query=user_query,
code_context=code_context or "",
conversation_history=self._build_conversation_history(conversation_history),
error_message="",
target_language="根据上下文判断",
user_requirement=user_query,
algorithm_code=""
)
def generate_dynamic_prompt(
self,
user_query: str,
intent_result: Optional[Dict[str, Any]] = None,
code_context: Optional[str] = None,
conversation_history: Optional[List[Dict[str, str]]] = None,
**kwargs
) -> str:
"""
生成动态Prompt基于意图识别结果
Args:
user_query: 用户问题
intent_result: 意图识别结果
code_context: 代码上下文
conversation_history: 对话历史
**kwargs: 其他参数
Returns:
str: 生成的动态Prompt
"""
try:
if intent_result:
# 从意图结果中提取分类
category_str = intent_result.get('category', 'unknown')
# 直接使用category_str因为CodeIntentCategory是一个普通的类不是枚举类型
intent_category = category_str
else:
# 默认使用通用分类
intent_category = CodeIntentCategory.UNKNOWN
# 提取其他参数
error_message = kwargs.get('error_message')
target_language = kwargs.get('target_language')
user_requirement = kwargs.get('user_requirement')
retrieved_results = kwargs.get('retrieved_results', [])
# 根据意图和 retrieved_results 构建增强的上下文
enhanced_context = code_context
if retrieved_results:
enhanced_context = self._build_enhanced_context(retrieved_results, intent_result)
# 生成Prompt
return self.generate_prompt(
user_query=user_query,
intent_category=intent_category,
code_context=enhanced_context,
conversation_history=conversation_history,
error_message=error_message,
target_language=target_language,
user_requirement=user_requirement
)
except Exception as e:
logger.error(f"生成动态Prompt失败: {e}")
# 返回通用Prompt
return self._templates[PromptTemplateType.GENERAL_QA].format(
user_query=user_query,
code_context=code_context or "",
conversation_history=self._build_conversation_history(conversation_history),
error_message="",
target_language="根据上下文判断",
user_requirement=user_query,
algorithm_code=""
)
def optimize_prompt(
self,
prompt: str,
max_length: int = 4000,
preserve_structure: bool = True
) -> str:
"""
优化Prompt长度
Args:
prompt: 原始Prompt
max_length: 最大长度
preserve_structure: 是否保留结构
Returns:
str: 优化后的Prompt
"""
if len(prompt) <= max_length:
return prompt
logger.warning(f"Prompt过长 ({len(prompt)} > {max_length}),需要优化")
if preserve_structure:
# 保留结构,只优化内容部分
# 1. 保留角色设定和核心指令
# 2. 精简分析要求
# 3. 缩短代码上下文
# 提取角色设定和核心指令
role_match = re.search(r'# 角色设定[\s\S]*?# 核心指令[\s\S]*?\n', prompt)
if role_match:
role_section = role_match.group(0)
else:
role_section = ""
# 提取分析要求
req_match = re.search(r'# 分析要求[\s\S]*?(?=# |$)', prompt)
if req_match:
req_section = req_match.group(0)
# 精简分析要求
req_lines = req_section.split('\n')
# 只保留前3条要求
req_section = '\n'.join(req_lines[:4]) # 保留标题和前3条
else:
req_section = ""
# 提取其他部分
rest_match = re.search(r'# (代码上下文|错误信息|对话历史|用户问题|输出格式)[\s\S]*$', prompt)
if rest_match:
rest_section = rest_match.group(0)
# 缩短代码上下文
if '# 代码上下文' in rest_section:
code_match = re.search(r'# 代码上下文[\s\S]*?(?=# |$)', rest_section)
if code_match:
code_section = code_match.group(0)
# 只保留前500个字符
if len(code_section) > 600:
code_lines = code_section.split('\n')
if len(code_lines) > 3:
# 保留标题和前几行
code_section = '\n'.join(code_lines[:2]) + '\n...\n(代码已截断)'
rest_section = rest_section.replace(code_match.group(0), code_section)
else:
rest_section = ""
optimized = role_section + '\n' + req_section + '\n' + rest_section
if len(optimized) > max_length:
# 进一步缩短
optimized = optimized[:max_length - 3] + '...'
else:
# 直接截断
optimized = prompt[:max_length - 3] + '...'
logger.info(f"Prompt优化完成长度: {len(optimized)}字符")
return optimized
def save_prompt_template(
self,
template_type: PromptTemplateType,
template_content: str,
description: Optional[str] = None
) -> bool:
"""
保存自定义Prompt模板
Args:
template_type: Prompt类型
template_content: 模板内容
description: 模板描述
Returns:
bool: 保存是否成功
"""
try:
# 这里可以扩展为持久化存储
# 目前只是在内存中更新
self._templates[template_type] = template_content
logger.info(f"保存Prompt模板成功: {template_type.value}")
return True
except Exception as e:
logger.error(f"保存Prompt模板失败: {e}")
return False
def get_prompt_template(self, template_type: PromptTemplateType) -> Optional[str]:
"""
获取Prompt模板
Args:
template_type: Prompt类型
Returns:
Optional[str]: 模板内容
"""
return self._templates.get(template_type)
def list_available_templates(self) -> List[Dict[str, Any]]:
"""
列出可用的Prompt模板
Returns:
List[Dict[str, Any]]: 模板列表
"""
templates = []
for template_type, content in self._templates.items():
templates.append({
"type": template_type.value,
"name": template_type.name,
"length": len(content),
"sample": content[:100] + "..." if len(content) > 100 else content
})
return templates
# 全局Prompt管理器实例
_prompt_manager = None
def get_prompt_manager() -> CodePromptManager:
"""
获取全局Prompt管理器实例
Returns:
CodePromptManager: Prompt管理器实例
"""
global _prompt_manager
if _prompt_manager is None:
_prompt_manager = CodePromptManager()
return _prompt_manager
def generate_code_prompt(
user_query: str,
intent_category: CodeIntentCategory,
code_context: Optional[str] = None,
conversation_history: Optional[List[Dict[str, str]]] = None,
**kwargs
) -> str:
"""
生成代码专用Prompt
Args:
user_query: 用户问题
intent_category: 代码意图分类
code_context: 代码上下文
conversation_history: 对话历史
**kwargs: 其他参数
Returns:
str: 生成的Prompt
"""
manager = get_prompt_manager()
return manager.generate_prompt(
user_query=user_query,
intent_category=intent_category,
code_context=code_context,
conversation_history=conversation_history,
**kwargs
)
def generate_dynamic_code_prompt(
user_query: str,
intent_result: Optional[Dict[str, Any]] = None,
code_context: Optional[str] = None,
conversation_history: Optional[List[Dict[str, str]]] = None,
**kwargs
) -> str:
"""
生成动态代码Prompt
Args:
user_query: 用户问题
intent_result: 意图识别结果
code_context: 代码上下文
conversation_history: 对话历史
**kwargs: 其他参数
Returns:
str: 生成的动态Prompt
"""
manager = get_prompt_manager()
return manager.generate_dynamic_prompt(
user_query=user_query,
intent_result=intent_result,
code_context=code_context,
conversation_history=conversation_history,
**kwargs
)
if __name__ == "__main__":
"""测试代码"""
import asyncio
from utils.code_intent import CodeIntentDetector
async def test_prompt_generation():
"""测试Prompt生成"""
print("=" * 80)
print("测试代码Prompt生成")
print("=" * 80)
# 初始化管理器
manager = CodePromptManager()
detector = CodeIntentDetector()
# 测试用例
test_cases = [
{
"query": "这个函数是做什么的?如何使用它?",
"code": "def calculate_factorial(n):\n if n <= 1:\n return 1\n return n * calculate_factorial(n-1)",
"category": CodeIntentCategory.ENTITY_INTRODUCTION
},
{
"query": "为什么会报语法错误?",
"code": "for i in range(10)\n print(i)",
"error": "SyntaxError: invalid syntax",
"category": CodeIntentCategory.ERROR_DEBUGGING
},
{
"query": "如何实现快速排序算法?",
"category": CodeIntentCategory.CODE_GENERATION
},
{
"query": "如何优化这段代码的性能?",
"code": "def slow_function():\n result = []\n for i in range(100000):\n result.append(i * 2)\n return result",
"category": CodeIntentCategory.CODE_OPTIMIZATION
}
]
for i, test_case in enumerate(test_cases):
print(f"\n测试用例 {i+1}: {test_case['query']}")
print("-" * 60)
# 生成Prompt
prompt = manager.generate_prompt(
user_query=test_case['query'],
intent_category=test_case['category'],
code_context=test_case.get('code'),
error_message=test_case.get('error')
)
# 打印结果
print(f"Prompt类型: {manager._map_intent_to_prompt_type(test_case['category']).value}")
print(f"Prompt长度: {len(prompt)}字符")
print("\nPrompt内容:")
print(prompt[:300] + "..." if len(prompt) > 300 else prompt)
print("-" * 60)
print("\n" + "=" * 80)
print("测试完成")
print("=" * 80)
asyncio.run(test_prompt_generation())