PR-Agent `algo` 폴더 리팩토링을 통한 오픈소스 기여하기

Tasker_Jang·2025년 5월 12일
agent오픈소스인공지능
1

오늘은 PR-Agent 오픈소스 프로젝트의 algo 폴더를 분석하고 이를 리팩토링하여 기여할 수 있는 방법에 대해 알아보겠습니다.

PR-Agent의 algo 폴더와 관련된 주요 이슈들은 다음과 같습니다:

오류 처리 관련 이슈

이슈 #1053: Error Handling Improvements

  • 현재 오류 처리가 일관되지 않고 예외가 제대로 전파되지 않는 문제를 다룸
  • 치명적인(Fatal) 오류와 무시할 수 있는(Ignore) 오류, 경고(Warning) 수준의 오류를 구분하는 체계 필요
  • 이슈 링크: https://github.com/qodo-ai/pr-agent/issues/1053

모델 호환성 관련 이슈

이슈 #1589: Integration broken for non-Tier 3 OpenAI accounts due to o3-mini requirement

  • Tier 3 OpenAI 계정이 없는 사용자들이 o3-mini 모델 요구사항 때문에 통합이 실패하는 문제
  • 대체 모델 메커니즘이 제대로 작동하지 않음
  • 이슈 링크: https://github.com/qodo-ai/pr-agent/issues/1589

이슈 #1461: Add Support for Mistral API

인증 관련 이슈

이슈 #1422: Using authenticated Ollama instance

이슈 #1557: Need help setting headers to azure open ai

이러한 이슈들은 algo 폴더의 AI 핸들러, 토큰 처리, 오류 처리 등의 개선이 필요함을 보여줍니다. 특히 오류 처리 체계화(#1053), 모델 호환성 개선(#1589, #1461), 그리고 인증 메커니즘 강화(#1422, #1557)가 가장 중점적으로 다뤄야 할 영역으로 보입니다.

algo 폴더 구조와 역할

algo 폴더는 PR-Agent의 핵심 알고리즘을 담고 있는 중요한 디렉토리입니다. 코드베이스를 분석한 결과, 이 폴더는 다음과 같은 주요 파일들로 구성되어 있습니다:

  1. ai_handlers/ - AI 모델과의 상호작용을 처리하는 핸들러들

    • base_ai_handler.py - 기본 추상 클래스
    • langchain_ai_handler.py - Langchain 기반 핸들러
    • litellm_ai_handler.py - LiteLLM 기반 핸들러
    • openai_ai_handler.py - OpenAI 직접 연동 핸들러

  2. git_patch_processing.py - Git 패치 처리 알고리즘

  3. pr_processing.py - PR 처리 로직

  4. token_handler.py - 토큰 관리 및 계산

  5. file_filter.py - 파일 필터링 로직

  6. language_handler.py - 언어 처리

  7. utils.py - 각종 유틸리티 함수

  8. types.py - 타입 정의

  9. cli_args.py - CLI 인자 처리

리팩토링을 위해 중점적으로 살펴볼 파일은 git_patch_processing.py, pr_processing.py, token_handler.py, 그리고 ai_handlers 디렉토리의 파일들입니다.

개선이 필요한 영역과 해결 방안

코드베이스와 오픈 이슈를 분석한 결과, 다음과 같은 개선 포인트를 발견했습니다:

1. 일관성 있는 오류 처리 체계 구축

현재 문제점:

  • 코드베이스 전반에 걸쳐 예외 처리 방식이 일관되지 않음
  • 일부는 예외를 무시하고, 일부는 상위로 전파하지 않음
  • 서비스로 실행 시 PR 검토 실행이 제대로 작동하는지 모니터링하기 어려움

개선 방안:

# 기존 코드 (pr_processing.py)
try:
    # 일부 로직
except Exception as e:
    get_logger().warning(f"Failed to extend patch: {e}", 
        artifact={"traceback": traceback.format_exc()})
    return patch_str  # 조용히 원본 패치 반환

# 개선된 코드 예시
class ErrorLevel(Enum):
    IGNORE = 0
    WARNING = 1
    FATAL = 2

def handle_error(e: Exception, message: str, level: ErrorLevel, context=None):
    context = context or {}
    if level == ErrorLevel.IGNORE:
        get_logger().debug(f"{message}: {e}", artifact={"context": context})
    elif level == ErrorLevel.WARNING:
        get_logger().warning(f"{message}: {e}", 
            artifact={"context": context, "traceback": traceback.format_exc()})
    elif level == ErrorLevel.FATAL:
        get_logger().error(f"{message}: {e}", 
            artifact={"context": context, "traceback": traceback.format_exc()})
        raise PRAgentException(f"{message}: {e}") from e

try:
    # 일부 로직
except Exception as e:
    handle_error(e, "Failed to extend patch", ErrorLevel.WARNING, {"filename": filename})
    return patch_str

이러한 방식으로 오류 처리 로직을 일관되게 적용하면, 서비스 운영 시 문제 발생 지점을 명확히 식별할 수 있고, 실패 원인을 쉽게 추적할 수 있습니다.

2. 모델 호환성 및 대체 로직 강화

현재 문제점:

  • 일부 모델(예: o3-mini)은 특정 API 등급이 필요하여 사용자에게 제약이 있음
  • 대체 모델 메커니즘이 제대로 작동하지 않는 경우가 있음
  • Mistral API 등 추가 모델 지원 요청이 있음

개선 방안:

# 기존 코드의 일부 (token_handler.py)
def _get_all_models(model_type: ModelType = ModelType.REGULAR) -> List[str]:
    if model_type == ModelType.WEAK:
        model = get_model('model_weak')
    elif model_type == ModelType.REASONING:
        model = get_model('model_reasoning')
    elif model_type == ModelType.REGULAR:
        model = get_settings().config.model
    else:
        model = get_settings().config.model
    fallback_models = get_settings().config.fallback_models
    if not isinstance(fallback_models, list):
        fallback_models = [m.strip() for m in fallback_models.split(",")]
    all_models = [model] + fallback_models
    return all_models

# 개선된 코드 예시
def is_model_available(model_name: str) -> bool:
    """주어진 모델이 현재 환경에서 사용 가능한지 확인합니다."""
    # OpenAI 모델인 경우
    if 'gpt' in model_name.lower() or 'o' in model_name.lower():
        # OpenAI API를 사용하여 모델 가용성 확인
        try:
            # API 요청을 통해 모델 확인
            return True
        except Exception:
            return False
    # 다른 모델 제공자에 대한 검사
    return True  # 기본값

def get_available_models(model_type: ModelType = ModelType.REGULAR) -> List[str]:
    """사용 가능한 모델 목록을 반환합니다. 주 모델이 사용 불가능한 경우 자동으로 대체 모델로 전환합니다."""
    primary_model = get_primary_model(model_type)
    fallback_models = get_fallback_models()
    
    all_models = [primary_model] + fallback_models
    available_models = [model for model in all_models if is_model_available(model)]
    
    if not available_models:
        get_logger().warning(f"No available models found for {model_type}. Using default safe model.")
        # 항상 사용 가능한 기본 모델 반환
        return ["gpt-3.5-turbo"]
    
    return available_models

이렇게 모델 가용성을 확인하고 실제로 사용 가능한 모델만 반환하도록 개선하면, 사용자 환경에 따른 오류를 줄일 수 있습니다.

3. 인증 메커니즘 개선

현재 문제점:

  • Ollama 인증된 인스턴스 사용 시 헤더 기반 인증 지원 필요
  • Azure OpenAI에 커스텀 헤더 설정 시 문제 발생

개선 방안:

# 기존 코드 (litellm_ai_handler.py의 일부)
def __init__(self):
    self._api_base = get_settings().get("ollama.api_base", None)
    # ...

# 개선된 코드 예시
def __init__(self):
    self._api_base = get_settings().get("ollama.api_base", None)
    self._headers = self._get_auth_headers()
    # ...

def _get_auth_headers(self) -> Dict[str, str]:
    """서비스별 인증 헤더를 생성합니다."""
    headers = {}
    
    # Ollama 인증 헤더
    ollama_api_key = get_settings().get("ollama.api_key", None)
    if ollama_api_key:
        headers["X-API-Key"] = ollama_api_key
    
    # Azure OpenAI 커스텀 헤더
    if get_settings().get("openai.api_type") == "azure":
        azure_headers_str = get_settings().get("openai.default_headers", "{}")
        try:
            azure_headers = json.loads(azure_headers_str)
            headers.update(azure_headers)
        except json.JSONDecodeError as e:
            get_logger().warning(f"Failed to parse Azure headers: {e}")
    
    return headers

async def chat_completion(self, model: str, system: str, user: str, temperature: float = 0.2, img_path: str = None):
    # 헤더 적용
    extra_params = {"headers": self._headers}
    # ...

이러한 접근 방식으로 다양한 인증 메커니즘을 지원할 수 있습니다. 특히 Ollama나 Azure OpenAI와 같은 다양한 서비스에 대한 인증 헤더를 유연하게 처리할 수 있습니다.

4. 코드 구조 개선 및 유지보수성 향상

현재 문제점:

  • pr_processing.py와 git_patch_processing.py가 각각 26K, 22K 라인으로 너무 큼
  • 단일 책임 원칙이 잘 지켜지지 않음
  • 복잡한 중첩 조건문이 많음

개선 방안:

먼저 큰 파일을 기능별로 분리합니다:

  1. git_patch_processing.py를 다음과 같이 분리:

    • patch_extension.py - 패치 확장 로직
    • patch_hunk_handler.py - 헝크 처리 로직
    • patch_line_numbers.py - 라인 번호 처리
    • patch_utils.py - 패치 관련 유틸리티

  2. pr_processing.py를 다음과 같이 분리:

    • pr_diff_generator.py - PR 차이점 생성 로직
    • pr_compression.py - PR 압축 전략
    • pr_model_handler.py - 모델 관련 로직
    • pr_metadata.py - 메타데이터 처리

복잡한 함수를 더 작은 단위로 분리하는 예시:

# 기존 코드 (pr_processing.py)
def generate_full_patch(convert_hunks_to_line_numbers, file_dict, max_tokens_model, remaining_files_list_prev, token_handler):
    total_tokens = token_handler.prompt_tokens # initial tokens
    patches = []
    remaining_files_list_new = []
    files_in_patch_list = []
    for filename, data in file_dict.items():
        if filename not in remaining_files_list_prev:
            continue

        patch = data['patch']
        new_patch_tokens = data['tokens']
        edit_type = data['edit_type']

        # Hard Stop, no more tokens
        if total_tokens > max_tokens_model - OUTPUT_BUFFER_TOKENS_HARD_THRESHOLD:
            get_logger().warning(f"File was fully skipped, no more tokens: {filename}.")
            continue

        # 많은 다른 조건과 로직...
    return total_tokens, patches, remaining_files_list_new, files_in_patch_list

# 개선된 코드 예시
def should_skip_file(filename, remaining_files_list):
    """파일을 건너뛰어야 하는지 확인합니다."""
    return filename not in remaining_files_list

def is_token_limit_exceeded(total_tokens, max_tokens_model):
    """토큰 제한이 초과되었는지 확인합니다."""
    return total_tokens > max_tokens_model - OUTPUT_BUFFER_TOKENS_HARD_THRESHOLD

def is_patch_too_large(total_tokens, patch_tokens, max_tokens_model):
    """패치가 너무 큰지 확인합니다."""
    return total_tokens + patch_tokens > max_tokens_model - OUTPUT_BUFFER_TOKENS_SOFT_THRESHOLD

def format_patch(filename, patch, convert_hunks_to_line_numbers):
    """패치를 포맷팅합니다."""
    if not convert_hunks_to_line_numbers:
        return f"\n\n## File: '{filename.strip()}'\n\n{patch.strip()}\n"
    else:
        return "\n\n" + patch.strip()

def generate_full_patch(convert_hunks_to_line_numbers, file_dict, max_tokens_model, remaining_files_list_prev, token_handler):
    total_tokens = token_handler.prompt_tokens
    patches = []
    remaining_files_list_new = []
    files_in_patch_list = []
    
    for filename, data in file_dict.items():
        if should_skip_file(filename, remaining_files_list_prev):
            continue

        patch = data['patch']
        new_patch_tokens = data['tokens']
        
        if is_token_limit_exceeded(total_tokens, max_tokens_model):
            get_logger().warning(f"File was fully skipped, no more tokens: {filename}.")
            continue

        if is_patch_too_large(total_tokens, new_patch_tokens, max_tokens_model):
            remaining_files_list_new.append(filename)
            continue

        if patch:
            patch_final = format_patch(filename, patch, convert_hunks_to_line_numbers)
            patches.append(patch_final)
            total_tokens += token_handler.count_tokens(patch_final)
            files_in_patch_list.append(filename)
            
    return total_tokens, patches, remaining_files_list_new, files_in_patch_list

이러한 리팩토링으로 각 함수의 책임이 명확해지고, 코드의 가독성과 유지보수성이 크게 향상됩니다.

5. 토큰 처리 개선

현재 문제점:

  • 다양한 모델의 토큰 계산 방식이 복잡함
  • 큰 패치에 대한 토큰 계산이 비효율적일 수 있음

개선 방안:

# 기존 코드 (token_handler.py)
def count_tokens(self, patch: str, force_accurate=False) -> int:
    encoder_estimate = len(self.encoder.encode(patch, disallowed_special=()))

    if not force_accurate:
        return encoder_estimate

    model = get_settings().config.model.lower()
    if 'claude' in model and get_settings(use_context=False).get('anthropic.key'):
        return self.calc_claude_tokens(patch)

    return self.estimate_token_count_for_non_anth_claude_models(model, encoder_estimate)

# 개선된 코드 예시
@functools.lru_cache(maxsize=128)
def _count_tokens_cached(self, text: str) -> int:
    """토큰 수를 계산하고 결과를 캐시합니다."""
    return len(self.encoder.encode(text, disallowed_special=()))

def count_tokens(self, patch: str, force_accurate=False) -> int:
    """패치의 토큰 수를 계산합니다."""
    # 매우 큰 텍스트는 청크로 나눠 계산
    if len(patch) > 50000:  # 매우 큰 패치
        return self._count_tokens_for_large_text(patch, force_accurate)
    
    # 일반적인 크기의 텍스트
    encoder_estimate = self._count_tokens_cached(patch)
    
    if not force_accurate:
        return encoder_estimate
    
    # 모델별 정확한 토큰 계산
    return self._get_accurate_token_count_by_model(patch, encoder_estimate)

def _count_tokens_for_large_text(self, text: str, force_accurate=False) -> int:
    """대용량 텍스트의 토큰 수를 청크 단위로 계산합니다."""
    # 텍스트를 청크로 나누기
    chunk_size = 10000  # 문자 기준
    chunks = [text[i:i+chunk_size] for i in range(0, len(text), chunk_size)]
    
    # 각 청크의 토큰 수 계산
    total_tokens = 0
    for chunk in chunks:
        total_tokens += self._count_tokens_cached(chunk)
    
    # 필요시 모델별 정확한 계산 적용
    if force_accurate:
        adjustment_factor = self._get_model_adjustment_factor()
        total_tokens = int(total_tokens * adjustment_factor)
    
    return total_tokens

def _get_model_adjustment_factor(self) -> float:
    """모델별 토큰 조정 계수를 반환합니다."""
    model = get_settings().config.model.lower()
    
    # 모델별 조정 계수
    adjustment_factors = {
        'gpt-4': 1.0,
        'gpt-3.5-turbo': 1.0,
        # 다른 모델들...
    }
    
    # 기본 조정 계수
    default_factor = 1 + get_settings().get('config.model_token_count_estimate_factor', 0)
    
    return adjustment_factors.get(model, default_factor)

이러한 개선으로 토큰 계산의 효율성과 정확성을 높일 수 있습니다. 특히 큰 파일을 처리할 때 성능 향상이 기대됩니다.

6. 파일 인코딩 처리 개선

현재 문제점:

  • 파일 인코딩 처리가 여러 곳에 분산되어 있음
  • 일부 인코딩 처리 실패 시 예외 처리가 미흡할 수 있음

개선 방안:

# 기존 코드 (git_patch_processing.py)
def decode_if_bytes(original_file_str):
    if isinstance(original_file_str, (bytes, bytearray)):
        try:
            return original_file_str.decode('utf-8')
        except UnicodeDecodeError:
            encodings_to_try = ['iso-8859-1', 'latin-1', 'ascii', 'utf-16']
            for encoding in encodings_to_try:
                try:
                    return original_file_str.decode(encoding)
                except UnicodeDecodeError:
                    continue
            return ""
    return original_file_str

# 개선된 코드 예시
class EncodingHelper:
    """파일 인코딩을 처리하는 헬퍼 클래스"""
    
    # 자주 사용되는 인코딩 목록
    COMMON_ENCODINGS = ['utf-8', 'iso-8859-1', 'latin-1', 'utf-16', 'cp1252', 'ascii']
    
    @staticmethod
    def decode_bytes(content: Union[bytes, bytearray, str], 
                   filename: Optional[str] = None) -> Tuple[str, Optional[str]]:
        """바이트 데이터를 문자열로, 사용된 인코딩을 반환합니다."""
        if not isinstance(content, (bytes, bytearray)):
            return content, None
        
        # 파일 확장자에 따른 우선 인코딩
        preferred_encoding = EncodingHelper._get_preferred_encoding(filename)
        encodings_to_try = [preferred_encoding] + [e for e in EncodingHelper.COMMON_ENCODINGS 
                                                if e != preferred_encoding]
        
        # 인코딩 시도
        for encoding in encodings_to_try:
            try:
                decoded = content.decode(encoding)
                return decoded, encoding
            except (UnicodeDecodeError, LookupError):
                continue
                
        # 마지막 수단으로 오류 무시 모드 시도
        try:
            decoded = content.decode('utf-8', errors='ignore')
            get_logger().warning(f"Decoded with utf-8 ignoring errors for {filename}")
            return decoded, 'utf-8-ignore'
        except Exception as e:
            get_logger().error(f"Failed to decode content for {filename}: {e}")
            return "", None
    
    @staticmethod
    def _get_preferred_encoding(filename: Optional[str]) -> str:
        """파일 유형에 따른 선호 인코딩을 반환합니다."""
        if not filename:
            return 'utf-8'
            
        # 확장자별 일반적인 인코딩
        if filename.endswith(('.js', '.py', '.java', '.kt', '.ts', '.html', '.css', '.json')):
            return 'utf-8'
        if filename.endswith(('.bat', '.cmd')):
            return 'cp1252'
            
        return 'utf-8'  # 기본 인코딩

이 개선된 EncodingHelper 클래스는 인코딩 처리를 중앙화하고, 파일 유형에 따른 최적의 인코딩을 선택합니다. 또한 인코딩 결과를 더 자세히 로깅하여 문제 해결에 도움을 줍니다.

리팩토링 진행 계획

위의 개선 사항을 기반으로 다음과 같은 단계로 리팩토링을 진행할 수 있습니다:

  1. 1단계: 오류 처리 체계 구축

    • 오류 수준 정의 (IGNORE, WARNING, FATAL)
    • 일관된 오류 처리 함수 구현
    • 주요 모듈에 적용

  2. 2단계: 모듈 분리

    • 큰 파일들을 기능별로 작은 모듈로 분리
    • 각 모듈의 인터페이스 정의
    • 단위 테스트 추가

  3. 3단계: 모델 호환성 개선

    • 모델 가용성 확인 메커니즘 추가
    • 대체 모델 로직 강화
    • Mistral API 지원 추가

  4. 4단계: 인증 메커니즘 강화

    • 헤더 기반 인증 지원
    • Azure OpenAI 커스텀 헤더 지원
    • Ollama 인증 지원

  5. 5단계: 성능 최적화

    • 토큰 계산 최적화
    • 캐싱 메커니즘 추가
    • 대용량 파일 처리 개선

결론

PR-Agent의 algo 폴더는 프로젝트의 핵심 알고리즘을 담고 있으며, 여러 개선 포인트가 있습니다. 오류 처리 체계 구축, 코드 구조 개선, 모델 호환성 강화, 인증 메커니즘 개선, 성능 최적화 등의 작업을 통해 코드의 품질과 유지보수성을 크게 향상시킬 수 있습니다.

profile
Tasker_Jang
ML Engineer 🧠 | AI 모델 개발과 최적화 경험을 기록하며 성장하는 개발자 🚀 The light that burns twice as bright burns half as long ✨
이전 포스트

Market Analysis Agent

다음 포스트

Alpha Vantage API에서 미국 주식 분석기 만들기 9)

0개의 댓글