|
1 | 1 | import re |
2 | | - |
3 | 2 | import yaml |
| 3 | +import abc |
| 4 | +from typing import Dict, Any, List |
4 | 5 |
|
5 | 6 | from biz.utils.log import logger |
6 | 7 | from biz.llm.factory import Factory |
7 | 8 |
|
8 | 9 |
|
9 | | -class CodeReviewer: |
10 | | - def __init__(self): |
| 10 | +class BaseReviewer(abc.ABC): |
| 11 | + """代码审查基类""" |
| 12 | + |
| 13 | + def __init__(self, prompt_key: str): |
11 | 14 | self.client = Factory().getClient() |
12 | | - self.prompts = self._load_prompts() |
| 15 | + self.prompts = self._load_prompts(prompt_key) |
13 | 16 |
|
14 | | - def _load_prompts(self) -> dict: |
| 17 | + def _load_prompts(self, prompt_key: str) -> Dict[str, Any]: |
15 | 18 | """加载提示词配置""" |
16 | 19 | prompt_templates_file = "conf/prompt_templates.yml" |
17 | | - with open(prompt_templates_file, "r") as file: |
18 | | - prompt_templates = yaml.safe_load(file) |
19 | | - system_prompt = prompt_templates['code_review_prompt']['system_prompt'] |
20 | | - user_prompt = prompt_templates['code_review_prompt']['user_prompt'] |
21 | | - |
22 | | - if not system_prompt or not user_prompt: |
23 | | - logger.warning(f"未找到提示词配置{prompt_templates_file}") |
24 | | - # 抛出异常 |
25 | | - raise Exception(f"未找到提示词配置{prompt_templates_file},或配置格式不正确") |
26 | | - |
27 | | - return { |
28 | | - "code_review": { |
29 | | - "system_message": { |
30 | | - "role": "system", |
31 | | - "content": system_prompt |
32 | | - }, |
33 | | - "user_message": { |
34 | | - "role": "user", |
35 | | - "content": user_prompt |
| 20 | + try: |
| 21 | + with open(prompt_templates_file, "r") as file: |
| 22 | + prompts = yaml.safe_load(file).get(prompt_key, {}) |
| 23 | + system_prompt = prompts.get("system_prompt") |
| 24 | + user_prompt = prompts.get("user_prompt") |
| 25 | + |
| 26 | + if not system_prompt or not user_prompt: |
| 27 | + raise ValueError(f"提示词配置 `{prompt_key}` 为空或格式不正确") |
| 28 | + |
| 29 | + return { |
| 30 | + "system_message": {"role": "system", "content": system_prompt}, |
| 31 | + "user_message": {"role": "user", "content": user_prompt}, |
36 | 32 | } |
37 | | - } |
38 | | - } |
| 33 | + except (FileNotFoundError, KeyError, yaml.YAMLError) as e: |
| 34 | + logger.error(f"加载提示词配置失败: {e}") |
| 35 | + raise Exception(f"提示词配置加载失败: {e}") |
| 36 | + |
| 37 | + def call_llm(self, messages: List[Dict[str, Any]]) -> str: |
| 38 | + """调用 LLM 进行代码审核""" |
| 39 | + logger.info(f"向 AI 发送代码 Review 请求, messages: {messages}") |
| 40 | + review_result = self.client.completions(messages=messages) |
| 41 | + logger.info(f"收到 AI 返回结果: {review_result}") |
| 42 | + return review_result |
| 43 | + |
| 44 | + @abc.abstractmethod |
| 45 | + def review_code(self, *args, **kwargs) -> str: |
| 46 | + """抽象方法，子类必须实现""" |
| 47 | + pass |
| 48 | + |
| 49 | + |
| 50 | +class CodeReviewer(BaseReviewer): |
| 51 | + """代码 Diff 级别的审查""" |
| 52 | + |
| 53 | + def __init__(self): |
| 54 | + super().__init__("code_review_prompt") |
39 | 55 |
|
40 | 56 | def review_code(self, diffs_text: str, commits_text: str = "") -> str: |
41 | | - """Review代码，并返回结果""" |
42 | | - prompts = self.prompts["code_review"] |
| 57 | + """Review 代码并返回结果""" |
43 | 58 | messages = [ |
44 | | - prompts["system_message"], |
| 59 | + self.prompts["system_message"], |
45 | 60 | { |
46 | 61 | "role": "user", |
47 | | - "content": prompts["user_message"]["content"].format( |
| 62 | + "content": self.prompts["user_message"]["content"].format( |
48 | 63 | diffs_text=diffs_text, |
49 | 64 | commits_text=commits_text |
50 | | - ) |
51 | | - } |
| 65 | + ), |
| 66 | + }, |
52 | 67 | ] |
53 | 68 | return self.call_llm(messages) |
54 | 69 |
|
55 | | - def call_llm(self, messages: list) -> str: |
56 | | - logger.info(f"向AI发送代码Review请求, message:{messages})") |
57 | | - review_result = self.client.completions( |
58 | | - messages=messages |
59 | | - ) |
60 | | - logger.info(f"收到AI返回结果:{review_result}") |
61 | | - return review_result |
62 | | - |
63 | 70 | @staticmethod |
64 | 71 | def parse_review_score(review_text: str) -> int: |
65 | | - """解析AI返回的Review结果，返回评分""" |
| 72 | + """解析 AI 返回的 Review 结果，返回评分""" |
66 | 73 | if not review_text: |
67 | | - return 0 # 如果review_text为空，返回 0 |
| 74 | + return 0 |
68 | 75 | match = re.search(r"总分[:：]\s*(\d+)分?", review_text) |
| 76 | + return int(match.group(1)) if match else 0 |
| 77 | + |
| 78 | + |
| 79 | +class CodeBaseReviewer(BaseReviewer): |
| 80 | + """代码库级别的审查""" |
| 81 | + |
| 82 | + def __init__(self): |
| 83 | + super().__init__("codebase_review_prompt") |
69 | 84 |
|
70 | | - if match: |
71 | | - return int(match.group(1)) # 提取数值部分并转换为整数 |
72 | | - return 0 # 如果未匹配到，返回 0 |
| 85 | + def review_code(self, language: str, directory_structure: str) -> str: |
| 86 | + """Review 代码库并返回结果""" |
| 87 | + messages = [ |
| 88 | + self.prompts["system_message"], |
| 89 | + { |
| 90 | + "role": "user", |
| 91 | + "content": self.prompts["user_message"]["content"].format( |
| 92 | + language=language, |
| 93 | + directory_structure=directory_structure, |
| 94 | + ), |
| 95 | + }, |
| 96 | + ] |
| 97 | + return self.call_llm(messages) |
0 commit comments