[ PR Agent ] CLI 방식 동작구조

PR Agent 를 로컬에서 실행해보았다. 

CLI 방식으로 실행해보았고, 동작구조를 살펴보기 위해 pr_agent/cli.py 파일을 들여다 보았다. 

 

동작구조 도식화

---

0. cli.py 파일 트리거

cli.py 파일이 실행되면, run 메소드가 실행된다.

if __name__ == '__main__':
    run()

 

1. CLI argument parser

run 메소드가 실행되면, 가장 먼저 set_parser 즉, CLI로 입력된 커맨드라인 인자값을 파싱하는 메소드(=set_parser)가 실행된다.

def run(inargs=None, args=None):
    parser = set_parser()

 

set_parser() 메소드

PR-Agent를 실행할 때 어떤 인자를 받을지 정의하는 역할을 한다.

def set_parser():
    parser = argparse.ArgumentParser(description='AI based pull request analyzer', usage="...")
    parser.add_argument('--version', action='version', version=f'pr-agent {get_version()}')
    parser.add_argument('--pr_url', type=str, help='The URL of the PR to review', default=None)
    parser.add_argument('--issue_url', type=str, help='The URL of the Issue to review', default=None)
    parser.add_argument('command', type=str, help='The', choices=commands, default='review')
    parser.add_argument('rest', nargs=argparse.REMAINDER, default=[])
    return parser

• PR-Agent 버전 확인 : parser.add_argument('--version', action='version', version=f'pr-agent {get_version()}') 
• 분석할 PR URL : parser.add_argument('--pr_url', type=str, help='The URL of the PR to review', default=None)
• 분석할 Issue URL : parser.add_argument('--issue_url', type=str, help='The URL of the Issue to review', default=None)
• 실행할 커맨드를 선택 ( review, describe, improve 등 ) : parser.add_argument('command', type=str, help='The', choices=commands, default='review')
• 나머지 추가 인자를 모두 문자열 리스트 형태로 수집 ( ask "What does this PR do?" ): parser.add_argument('rest', nargs=argparse.REMAINDER, default=[])

 

2. 입력인자 유효성 체크 및 분기처리

 

인자값 없을 때, 보완처리

if not args:
        args = parser.parse_args(inargs)

• 인자가 비어 있다면, inargs를 파싱해서 args에 넣는다.
• 즉, 사용자가 명령어를 입력하지 않고 실행했을 경우, inargs로 받은 실제 커맨드라인 인자를 사용하도록 보완하는 처리다.

---

--pr_url이나 --issue_url 둘 다 입력되지 않았을 경우

 

 

if not args.pr_url and not args.issue_url:
    parser.print_help()
    return

 

• CLI 사용법과 명령어 목록이 포함된 도움말 메시지를 출력하고,
• 프로그램 실행을 중단(return)한다.

 

3. 비동기 루틴 async def inner()

3.1 pr / issue 분기처리

if args.issue_url:
    result = await asyncio.create_task(
        PRAgent().handle_request(args.issue_url, [command] + args.rest)
    )
else:
    result = await asyncio.create_task(
        PRAgent().handle_request(args.pr_url, [command] + args.rest)
    )

 

• PRAgent().handle_request(...)가 핵심 처리 함수로, 내부적으로 LLM을 호출하거나 코드 리뷰를 실행하는 기능이 여기에 포함된다.
• command는 review, improve, ask 등 사용자가 CLI에 입력한 명령어이며, args.rest는 추가 인자를 리스트로 전달한다.
• issue_url이 있으면 해당 이슈를, 없으면 PR을 대상으로 한다.
• asyncio.create_task(...)로 작업을 비동기 태스크로 실행하며, await로 그 결과를 기다린다.

📌 pr_agent.py/handle_request 메소드를 잠시 보자면, 

# pr_agent.py

class PRAgent:
    def __init__(self, ai_handler: partial[BaseAiHandler,] = LiteLLMAIHandler):
        self.ai_handler = ai_handler  # will be initialized in run_action

    async def handle_request(self, pr_url, request, notify=None) -> bool:
        # First, apply repo specific settings if exists
        apply_repo_settings(pr_url)

        # Then, apply user specific settings if exists
        if isinstance(request, str):
            request = request.replace("'", "\\'")
            lexer = shlex.shlex(request, posix=True)
            lexer.whitespace_split = True
            action, *args = list(lexer)
        else:
            action, *args = request

        # validate args
        is_valid, arg = CliArgs.validate_user_args(args)
        if not is_valid:
            get_logger().error(
                f"CLI argument for param '{arg}' is forbidden. Use instead a configuration file."
            )
            return False

        # Update settings from args
        args = update_settings_from_args(args)

        # Append the response language in the extra instructions
        response_language = get_settings().config.get('response_language', 'en-us')
        if response_language.lower() != 'en-us':
            get_logger().info(f'User has set the response language to: {response_language}')
            for key in get_settings():
                setting = get_settings().get(key)
                if str(type(setting)) == "<class 'dynaconf.utils.boxing.DynaBox'>":
                    if hasattr(setting, 'extra_instructions'):
                        current_extra_instructions = setting.extra_instructions
                        if current_extra_instructions:
                            setting.extra_instructions = current_extra_instructions+ f"\n======\n\nIn addition, Your response MUST be written in the language corresponding to local code: {response_language}. This is crucial."
                        else:
                            setting.extra_instructions = f"Your response MUST be written in the language corresponding to locale code: '{response_language}'. This is crucial."

        action = action.lstrip("/").lower()
        if action not in command2class:
            get_logger().warning(f"Unknown command: {action}")
            return False
        with get_logger().contextualize(command=action, pr_url=pr_url):
            get_logger().info("PR-Agent request handler started", analytics=True)
            if action == "answer":
                if notify:
                    notify()
                await PRReviewer(pr_url, is_answer=True, args=args, ai_handler=self.ai_handler).run()
            elif action == "auto_review":
                await PRReviewer(pr_url, is_auto=True, args=args, ai_handler=self.ai_handler).run()
            elif action in command2class:
                if notify:
                    notify()

                await command2class[action](pr_url, ai_handler=self.ai_handler, args=args).run()
            else:
                return False
            return True

 

• PR URL 기반으로 레포지토리 및 사용자 설정을 적용한다.
• 입력받은 request 문자열을 파싱해 실행할 액션과 인자를 분리한다.
• 금지된 인자가 포함되어 있는지 확인하고, 있으면 에러 로그 출력 후 종료한다. ( return False ) 
• 설정 파일에서 응답 언어(response_language)를 가져와, 영어가 아닐 경우 AI에게 언어 지시를 추가로 부여함.
• 인식 가능한 명령어이면 command2class를 통해 해당 클래스의 run() 메서드를 실행하고, 성공 여부를 True 또는 False로 반환함.

 

---

3.2 비동기 콜백처리 

if get_settings().litellm.get("enable_callbacks", False):
    get_logger().debug("Waiting for event queue to complete")
    await asyncio.wait([
        task for task in asyncio.all_tasks()
        if task is not asyncio.current_task()
    ])

• enable_callbacks 설정이 켜져 있다면, PR-Agent는 LLM 호출에 따른 Task Queue를 기다리게 된다. 예를 들어 웹훅 전송, 결과 로깅, 후처리 작업 등이 있을 경우 그 태스크들이 완료될 때까지 기다리는 처리다.
• 현재 실행 중인 작업은 제외하고, 나머지 모든 태스크를 기다린다. ( if task is not asyncio.current_task() ~ )

---

3.3 결과반환

return result

• 위에서 수행한 PRAgent().handle_request()의 결과를 반환한다.
• 이 결과는 CLI에서 사용자에게 보여질 수도 있고, 후속 로직에 사용될 수도 있다.