見出し画像

AI Agent Github issue resolver : LLMとGitHubの統合による自動コード生成エージェント

はじめに

近年のLLMの飛躍的なコーディング能力の向上、特にOpenAIのo1モデルの登場により、LLM(大規模言語モデル)のコーディング能力は飛躍的に向上しました。ソフトウェア開発業務の中でLLMに頼る場面はけして少なくないでしょう。近い将来、コードを書く業務自体はLLMに委ねられる可能性も高いでしょう。一方で、LLMにコードを作成を依頼するインターフェースは未発達で、効果的にコード作成の依頼ができていないと思います。例えば、修正箇所をチャットで一行ずつ指示し、生成されたコードをコピー&ペーストする必要があるなど、煩雑な作業を強いられるケースも多いのではないでしょうか。そこで、GitHubと統合することで、仕様や作業内を記載したIssueから直接コードを生成し、プルリクエストまで自動化するAIエージェントを開発しました。本記事では、このエージェントの技術的な特徴や実装方法、そして今後の展望について詳しく解説します。


技術的な特徴

1. カスタムGitHubツール群の開発

このプロジェクトでは、以下のような一連のGitHub操作ツールをLangChainのカスタムツールとして実装しました:

Issue取得ツール(GitHubIssueRetrieverTool)
ブランチ作成ツール(GitHubBranchCreatorTool)
ファイル取得ツール(GitHubFileRetrieverTool)
ファイル更新ツール(GitHubFileUpdaterTool)
プルリクエスト作成ツール(GitHubPullRequestCreatorTool)
Issueクローズツール(GitHubIssueCloserTool)

これらのツールをLangGraphフレームワークと統合し、LLMがGitHubを直接操作できる環境を構築しました。

ツールの実装例

GitHubIssueRetrieverTool

  • GitHubのIssue一覧を取得する基本的なツール

  • GitHub APIを使用してリポジトリの全オープンIssueを取得

class GitHubIssueRetrieverInput(GitHubRepoInfo):
    pass

class GitHubIssueRetrieverTool(BaseTool):
    name: str = "github_issue_retriever"
    description: str = "指定されたGitHubリポジトリからオープンなIssueを取得します。"
    args_schema: Type[BaseModel] = GitHubIssueRetrieverInput

    def _run(
        self, tool_input: Optional[Dict[str, Any]] = None, run_manager: Optional = None, **kwargs
    ) -> str:
        if tool_input is None:
            tool_input = kwargs
        owner = tool_input.get('owner', OWNER)
        repo = tool_input.get('repo', REPO)
        access_token = tool_input.get('access_token', ACCESS_TOKEN)
        headers = {"Authorization": f"token {access_token}"}
        url = f"https://api.github.com/repos/{owner}/{repo}/issues"
        response = requests.get(url, headers=headers)
        if response.status_code == 200:
            issues = response.json()
            return json.dumps(issues)
        else:
            return f"Failed to retrieve issues: {response.status_code} {response.text}"

    async def _arun(self, *args, **kwargs):
        raise NotImplementedError("github_issue_retriever only supports sync")

GitHubBranchCreatorTool

  • 新しいブランチを作成するためのツール

  • 元になるブランチ(デフォルトはmain)のSHAを取得

  • そのSHAを基に新しいブランチを作成

  • ブランチ名とソースブランチを指定可能

class GitHubBranchCreatorInput(GitHubRepoInfo):
    new_branch_name: str = Field(description="作成する新しいブランチの名前")
    source_branch_name: str = Field(default="main", description="新しいブランチの元になる既存のブランチ名")

class GitHubBranchCreatorTool(BaseTool):
    name: str = "github_branch_creator"
    description: str = "GitHubリポジトリに新しいブランチを作成します。"
    args_schema: Type[BaseModel] = GitHubBranchCreatorInput

    def _run(
        self, tool_input: Optional[Dict[str, Any]] = None, run_manager: Optional = None, **kwargs
    ) -> str:
        if tool_input is None:
            tool_input = kwargs
        owner = tool_input.get('owner', OWNER)
        repo = tool_input.get('repo', REPO)
        new_branch_name = tool_input['new_branch_name']
        source_branch_name = tool_input.get('source_branch_name', 'main')
        access_token = tool_input.get('access_token', ACCESS_TOKEN)
        headers = {
            "Authorization": f"token {access_token}",
            "Content-Type": "application/json",
        }

        # 元になるブランチの最新コミットのSHAを取得
        url = f"https://api.github.com/repos/{owner}/{repo}/git/ref/heads/{source_branch_name}"
        get_response = requests.get(url, headers=headers)
        if get_response.status_code == 200:
            source_branch_info = get_response.json()
            sha = source_branch_info['object']['sha']
        else:
            return f"Failed to get source branch SHA: {get_response.status_code} {get_response.text}"

        # 新しいブランチを作成
        url = f"https://api.github.com/repos/{owner}/{repo}/git/refs"
        data = {
            "ref": f"refs/heads/{new_branch_name}",
            "sha": sha
        }
        post_response = requests.post(url, headers=headers, json=data)
        if post_response.status_code == 201:
            return f"Branch '{new_branch_name}' created successfully."
        else:
            return f"Failed to create branch: {post_response.status_code} {post_response.text}"

    async def _arun(self, *args, **kwargs):
        raise NotImplementedError("github_branch_creator only supports sync")

GitHubFileUpdaterTool

  • ファイルの作成・更新を行うツール

  • mainブランチへの直接変更を防止する安全機能付き

  • 既存ファイルの場合はSHAを取得して更新

  • 新規ファイルの場合は新たに作成

  • コンテンツはBase64エンコードして送信

class GitHubFileUpdaterInput(GitHubRepoInfo):
    file_path: str = Field(description="リポジトリ内のファイルのパス")
    content: str = Field(description="ファイルの新しい内容")
    commit_message: str = Field(description="コミットメッセージ")
    branch: str = Field(description="更新するブランチ名")

class GitHubFileUpdaterTool(BaseTool):
    name: str = "github_file_updater"
    description: str = "GitHubリポジトリ内のファイルを新しい内容で更新または作成します。"
    args_schema: Type[BaseModel] = GitHubFileUpdaterInput

    def _run(
        self, tool_input: Optional[Dict[str, Any]] = None, run_manager: Optional = None, **kwargs
    ) -> str:
        if tool_input is None:
            tool_input = kwargs
        owner = tool_input.get('owner', OWNER)
        repo = tool_input.get('repo', REPO)
        file_path = tool_input['file_path']
        content = tool_input['content']
        commit_message = tool_input['commit_message']
        access_token = tool_input.get('access_token', ACCESS_TOKEN)
        branch = tool_input.get('branch', 'main')

        if branch == 'main':
            return "Error: Direct modifications to main branch are not allowed. Please create a new branch first."

        headers = {
            "Authorization": f"token {access_token}",
            "Content-Type": "application/json",
        }

        # ファイルのSHAを取得(ブランチを指定)
        url = f"https://api.github.com/repos/{owner}/{repo}/contents/{file_path}?ref={branch}"
        get_response = requests.get(url, headers=headers)
        if get_response.status_code == 200:
            file_info = get_response.json()
            sha = file_info['sha']
        elif get_response.status_code == 404:
            sha = None  # 新規作成
        else:
            return f"Failed to get file SHA: {get_response.status_code} {get_response.text}"

        new_content = base64.b64encode(content.encode('utf-8')).decode('utf-8')
        data = {
            "message": commit_message,
            "content": new_content,
            "branch": branch,
        }
        if sha:
            data["sha"] = sha  # 既存ファイルを更新

        # ファイルの作成または更新
        url = f"https://api.github.com/repos/{owner}/{repo}/contents/{file_path}"
        put_response = requests.put(url, headers=headers, json=data)
        if put_response.status_code in [200, 201]:
            return "File updated successfully."
        else:
            return f"Failed to update file: {put_response.status_code} {put_response.text}"

    async def _arun(self, *args, **kwargs):
        raise NotImplementedError("github_file_updater only supports sync")

2. ツールとLLMの統合

LLMとカスタムツールを統合することで、自然言語の指示から直接GitHub操作を行えるようにしています。

# LLMの設定
llm = ChatOpenAI(model_name="gpt-4o")

# ツールのリストを作成
tools = [
    GitHubIssueRetrieverTool(),
    GitHubFileRetrieverTool(),
    GitHubFileUpdaterTool(),
    GitHubIssueCloserTool(),
    GitHubBranchCreatorTool(),
    GitHubPullRequestCreatorTool(),
]

# LLMにツールをバインド
llm_with_tools = llm.bind_tools(tools)

3. エージェントのワークフロー構築

langgraphを使用して、エージェントの対話フローを定義しています。
エージェントの基本的な処理手順とmainブランチへの直接編集の禁止などの注意事項は、システムメッセージで定義しています。

# システムプロンプトの定義
system_prompt = """
あなたはGitHubリポジトリの管理を行うエージェントです。以下の基本的な流れに従ってタスクを実行してください:

1. オープンなIssueを読む
2. 新しい作業用ブランチを作成
3. 新しいブランチにコードの作成または更新
4. コミットしてプッシュ
5. プルリクエストを作成

注意事項:
- mainブランチへの直接の変更は絶対に行わないでください
- 必ず新しいブランチを作成してから作業を開始してください
- ファイル更新時は必ずブランチを指定してください
"""

# ノードの定義
def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# ツールノード
tool_node = ToolNode(tools=tools)

# グラフの構築
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)
graph_builder.add_conditional_edges("chatbot", tools_condition)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

4. エージェントの実行

ユーザーの入力に対してエージェントが応答し、ツールを適切に呼び出してGitHub操作を行います。

# エージェントの実行関数
def run_agent(user_input: str, thread_id: str = "default"):
    config = {"configurable": {"thread_id": thread_id}}
    messages = [SystemMessage(content=system_prompt), HumanMessage(content=user_input)]
    events = graph.stream({"messages": messages}, config, stream_mode="values")
    for event in events:
        if "messages" in event:
            last_message = event["messages"][-1]
            print("Assistant:", last_message.content)
    if isinstance(last_message, AIMessage):
        messages.append(last_message)

# エージェントとの対話
if __name__ == "__main__":
    while True:
        user_input = input("User: ")
        if user_input.lower() in ["exit", "quit"]:
            break
        run_agent(user_input)

5. 完全なコード(2024/11/17追加)

実行例:Hello Worldプログラムの自動生成

実際のIssueを入力として与えたところ、エージェントは以下の一連の処理を自動的に実行しました。

1. Issueの要件を理解

  • 要件

    • コンソールに”Hello World”と出力

    • 適切なコメントの追加

    •  コーディング規約への準拠

    •  基本的なエラーハンドリングの実装

エージェントに作業を依頼するために作成したissue

2. 適切なブランチを作成

Branch 'feature/hello-world-program' created successfully.

3. コードを実装してプルリクエストを作成

コードの生成

def main():
    """
    メイン関数
    """
    try:
        print("Hello World")
    except Exception as e:
        print(f"エラーが発生しました: {e}")

if __name__ == "__main__":
    main()
  • プルリクエスト の自動作成

    • タイトル: Add Hello World Program

エージェントが作成したプルリクエスト
エージェントがissueをもとに作成したコード

実際の実行ログ

User: issueを参照してプルリクを作成して
Assistant: issueを参照してプルリクを作成して
Assistant: 
Assistant: [{"url": "https://api.github.com/repos/pome223/github-issue-resolver/issues/2", "repository_url": "https://api.github.com/xxxxxxxxxxxxx..
Assistant: 
Assistant: Branch 'feature/hello-world-program' created successfully.
Assistant: 
Assistant: File updated successfully.
Assistant: 
Assistant: Pull request #4 created successfully: https://github.com/pome223/github-issue-resolver/pull/4
Assistant: プルリクエストを作成しました。以下のリンクから詳細を確認できます: [プルリクエスト #4](https://github.com/pome223/github-issue-resolver/pull/4)

まとめ

このAIエージェントは、LLMの高度なコーディング能力とGitHubの開発プラットフォームを統合することで、ソフトウェア開発の新しい可能性を示しています。正確な仕様記述さえあれば、誰でも高品質なコード開発を依頼できる未来が現実のものとなりつつあります。

今後、 レビュープロセスの自動化や誤ったブランチに変更を加えたり、望ましくない操作を防止するためのガードレール(ルール) 機能などさらなる機能拡張により、開発プロセス全体の効率化と品質向上をはかりたいと考えています。



いいなと思ったら応援しよう!