xiaxinhang/pr-agent
1
0
Fork 0
mirror of https://github.com/qodo-ai/pr-agent.git synced 2025-07-02 11:50:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

Apply PR review feedback: Code style and functionality improvements

Browse Source
This commit is contained in:
benedict.lee
2025-04-10 21:56:37 +09:00
parent c5165d917b
commit 9c06b6b266
4 changed files with 53 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
pr_agent/git_providers/git_provider.py
View File

@ -285,14 +285,8 @@ class GitProvider(ABC):
def get_comment_url(self, comment) -> str:
return ""
def get_review_comment_by_id(self, comment_id: int):
pass
def get_review_id_by_comment_id(self, comment_id: int):
pass
def get_review_thread_comments(self, comment_id: int):
def get_review_thread_comments(self, comment_id: int) -> list[dict]:
pass
#### labels operations ####

124
pr_agent/git_providers/github_provider.py
View File

@ -427,130 +427,44 @@ class GithubProvider(GitProvider):
self._publish_inline_comments_fallback_with_verification(comments)
except Exception as e:
get_logger().error(f"Failed to publish inline code comments fallback, error: {e}")
raise e
def get_review_comment_by_id(self, comment_id: int):
"""
Retrieves a review comment by its ID.
Args:
comment_id: Review comment ID
Returns:
Review comment object or None (if not found)
"""
try:
# Using PyGitHub library
# There's no direct way to get PR comment by ID, so we fetch all comments and filter
all_comments = list(self.pr.get_comments())
for comment in all_comments:
if comment.id == comment_id:
return comment
return None
except Exception as e:
get_logger().warning(f"Failed to get review comment {comment_id}, error: {e}")
return None
def get_review_id_by_comment_id(self, comment_id: int):
"""
Finds the review ID that a comment belongs to based on its comment ID.
Args:
comment_id: Review comment ID
Returns:
Review ID or None (if not found)
"""
try:
comment = self.get_review_comment_by_id(comment_id)
if comment:
return getattr(comment, 'pull_request_review_id', None)
return None
except Exception as e:
get_logger().warning(f"Failed to get review ID for comment {comment_id}, error: {e}")
return None
raise e
def get_review_thread_comments(self, comment_id: int):
def get_review_thread_comments(self, comment_id: int) -> list[dict]:
"""
Retrieves all comments in the thread that a specific comment belongs to.
Retrieves all comments in the same line as the given comment.
Args:
comment_id: Review comment ID
Returns:
List of comments in the same thread
List of comments on the same line
"""
try:
# Get comment information
comment = self.get_review_comment_by_id(comment_id)
# Get the original comment to find its location
comment = self.pr.get_comment(comment_id)
if not comment:
return []
# Extract file path and line number
file_path = comment.path
line_number = comment.raw_data["line"] if "line" in comment.raw_data else comment.raw_data.get("original_line")
# get all comments
# Get all comments
all_comments = list(self.pr.get_comments())
# Filter comments in the same thread
thread_comments = []
in_reply_to_map = {}
# Filter comments on the same line of the same file
thread_comments = [
c for c in all_comments
if c.path == file_path and (c.raw_data.get("line") == line_number or c.raw_data.get("original_line") == line_number)
]
# First build the in_reply_to relationship map
for c in all_comments:
in_reply_to_id = getattr(c, 'in_reply_to_id', None)
if in_reply_to_id:
in_reply_to_map[c.id] = in_reply_to_id
# Recursively find all ancestor comments (collect comment's ancestors)
def find_ancestors(cid):
ancestors = []
current = cid
while current in in_reply_to_map:
parent_id = in_reply_to_map[current]
ancestors.append(parent_id)
current = parent_id
return ancestors
# Recursively find all descendant comments (collect all replies to the comment)
def find_descendants(cid):
descendants = []
for c in all_comments:
if getattr(c, 'in_reply_to_id', None) == cid:
descendants.append(c.id)
descendants.extend(find_descendants(c.id))
return descendants
# Find all descendants of a specific ancestor (including sibling comments)
def find_all_descendants_of_ancestor(ancestor_id):
all_descendants = []
for c in all_comments:
if getattr(c, 'in_reply_to_id', None) == ancestor_id:
all_descendants.append(c.id)
all_descendants.extend(find_descendants(c.id))
return all_descendants
# Collect both ancestor and descendant IDs of the comment
ancestors = find_ancestors(comment_id)
descendants = find_descendants(comment_id)
# Create thread ID set (self, ancestors, descendants)
thread_ids = set([comment_id] + ancestors + descendants)
# For each ancestor, include all conversation branches (sibling comments with the same ancestor)
for ancestor_id in ancestors:
sibling_ids = find_all_descendants_of_ancestor(ancestor_id)
thread_ids.update(sibling_ids)
# Filter to only get comments belonging to the thread
for c in all_comments:
if c.id in thread_ids:
thread_comments.append(c)
# Sort chronologically (by creation date)
# Sort chronologically
thread_comments.sort(key=lambda c: c.created_at)
return thread_comments
except Exception as e:
get_logger().warning(f"Failed to get review thread comments for comment {comment_id}, error: {e}")
get_logger().warning(f"Failed to get review comments for comment {comment_id}, error: {e}")
return []
def _publish_inline_comments_fallback_with_verification(self, comments: list[dict]):

3
pr_agent/settings/pr_line_questions_prompts.toml
View File

@ -49,10 +49,9 @@ Previous discussion on this code:
======
{{ conversation_history|trim }}
======
Consider both the previous review comments from authors and reviewers, as well as any previous questions and answers about this code. The "Previous Question" and "Previous AI Answer" show earlier interactions about the same code. Use this context to provide a more informed and consistent answer.
Use this prior discussion context to provide a consistent and informed answer.
{%- endif %}
A question about the selected lines:
======
{{ question|trim }}

94
pr_agent/tools/pr_line_questions.py
View File

@ -16,7 +16,7 @@ from pr_agent.git_providers import get_git_provider
from pr_agent.git_providers.git_provider import get_main_pr_language
from pr_agent.log import get_logger
from pr_agent.servers.help import HelpMessage
from pr_agent.git_providers.github_provider import GithubProvider
class PR_LineQuestions:
def __init__(self, pr_url: str, args=None, ai_handler: partial[BaseAiHandler,] = LiteLLMAIHandler):
@ -43,9 +43,6 @@ class PR_LineQuestions:
get_settings().pr_line_questions_prompt.user)
self.patches_diff = None
self.prediction = None
# get settings for use conversation history
self.use_conversation_history = get_settings().pr_questions.use_conversation_history
def parse_args(self, args):
if args and len(args) > 0:
@ -62,7 +59,7 @@ class PR_LineQuestions:
# set conversation history if enabled
# currently only supports GitHub provider
if self.use_conversation_history and isinstance(self.git_provider, GithubProvider):
if get_settings().pr_questions.use_conversation_history and isinstance(self.git_provider, GithubProvider):
self._load_conversation_history()
self.patch_with_lines = ""
@ -104,70 +101,41 @@ class PR_LineQuestions:
def _load_conversation_history(self):
"""generate conversation history from the code review thread"""
# set conversation history to empty string
self.vars["conversation_history"] = ""
comment_id = get_settings().get('comment_id', '')
file_path = get_settings().get('file_name', '')
line_number = get_settings().get('line_end', '')
# early return if any required parameter is missing
if not all([comment_id, file_path, line_number]):
return
try:
comment_id = get_settings().get('comment_id', '')
file_path = get_settings().get('file_name', '')
line_number = get_settings().get('line_end', '')
# retrieve thread comments
thread_comments = self.git_provider.get_review_thread_comments(comment_id)
# return if any required parameter is missing
if not all([comment_id, file_path, line_number]):
return
# initialize conversation history
# generate conversation history
conversation_history = []
for comment in thread_comments:
body = getattr(comment, 'body', '')
# skip empty comments, current comment(will be added as a question at prompt)
if not body or not body.strip() or comment_id == comment.id:
continue
user = comment.user
author = user.login if hasattr(user, 'login') else 'Unknown'
conversation_history.append(f"{author}: {body}")
if hasattr(self.git_provider, 'get_review_thread_comments') and comment_id:
try:
# get review thread comments
thread_comments = self.git_provider.get_review_thread_comments(comment_id)
# current question id (this question is excluded from the context)
current_question_id = comment_id
# generate conversation history from the comments
for comment in thread_comments:
# skip empty comments
body = getattr(comment, 'body', '')
if not body or not body.strip():
continue
# except for current question
# except for current question
if current_question_id and comment.id == current_question_id:
# remove the AI command (/ask etc) from the beginning of the comment (optional)
clean_body = body
if clean_body.startswith('/'):
clean_body = clean_body.split('\n', 1)[-1] if '\n' in clean_body else ''
if not clean_body.strip():
continue
# author info
user = comment.user
author = user.login if hasattr(user, 'login') else 'Unknown'
# confirm if the author is the current user (AI vs user)
is_ai = 'bot' in author.lower() or '[bot]' in author.lower()
role = 'AI' if is_ai else 'User'
# append to the conversation history
conversation_history.append(f"{role} ({author}): {clean_body}")
# transform the conversation history to a string
if conversation_history:
self.vars["conversation_history"] = "\n\n".join(conversation_history)
get_logger().info(f"Loaded {len(conversation_history)} comments from the code review thread")
else:
self.vars["conversation_history"] = ""
except Exception as e:
get_logger().warning(f"Failed to get review thread comments: {e}")
self.vars["conversation_history"] = ""
# transform and save conversation history
if conversation_history:
self.vars["conversation_history"] = "\n\n".join(conversation_history)
get_logger().info(f"Loaded {len(conversation_history)} comments from the code review thread")
except Exception as e:
get_logger().error(f"Error loading conversation history: {e}")
self.vars["conversation_history"] = ""
get_logger().error(f"Error processing conversation history, error: {e}")
async def _get_prediction(self, model: str):
variables = copy.deepcopy(self.vars)