Pythonでゼロから作るコーディングエージェント

はじめに

こんにちは。ナウキャストでデータエンジニアをしているTakumiです。

社内(Finatext HD内)の生成AIコンテストでMultiAgentを利用したシステムをスクラッチで構築しました。
具体的には、ユーザーがSlackでメッセージを送信し、コードの記述、レビュー、GitHubでのPR作成までEnd2Endでできるシステムです。
コンテストで構築したシステムの概要図は以下の通りです。

本記事では、複数のエージェントが協調して動作する本格的なコーディングAgent（Coodinator） に絞って、構築した概要を説明します。

今年、Finatextで開催された生成AIの概要は以下からご覧ください。(記事出来次第、追記予定)

過去のAIコンテスト出場記はこちら。
https://note.com/keiko_iwa/n/nfdee24704ebf

この記事でわかること

この記事では、LangChainとAzure OpenAIを使用してゼロからコーディングエージェントを構築する方法を詳しく解説します。

具体的には以下の内容を学ぶことができます。

• コーディングエージェントの実装方法：ProgrammerAgentとReviewerAgentを使ったMultiAgentシステムの構築方法
• Tool（Function）の設計と実装：エージェントが使用する機能の定義方法

この記事で紹介しているAgentについては以下のリポジトリで公開しています。
https://github.com/TakumiMukaiyama/coding_agent_from_scratch

実行環境

• Python: 3.12以上
• パッケージマネージャー: uv
• LLM: Azure OpenAI GPT-4.1
• フレームワーク: LangChain 0.3.24

LLMモデルの選択について

今回は、Azure OpenAI GPT-4.1を採用しました。当初は4o, 4o-mini, o3での実装を試みましたが、以下の理由からGPT-4.1を選択しました。

• トークン数の制限: 実際の使用において約22万トークンが必要となり、4oや4o-miniのトークン数上限（128K）を超過
• 複雑なコード生成: 複数ファイルを生成する必要があり、十分なコンテキストウインドウが必要

今回の生成対象がTerraformかつ、生成にあたって参照が必要なドキュメント数が多かったことからこの判断を行いました。
単純にコード生成を行うのであれば、Gemini、Claudeも含めた他モデルでも問題ないと思われます。

概要

コードの生成を行うエージェント(ProgrammerAgent)、コードのレビューを行うエージェント(ReviewerAgent)、それらを統合したマルチエージェント処理(AgentCoordinator) を構築しました。

処理フロー

具体的な処理フローは以下の通りです。

ProgrammerAgent

役割
ユーザーの指示に基づいてコードを生成する。

主な機能

• 指示内容の解析と理解
• 適切なプログラミング言語での実装
• ファイルの作成・編集・上書き
• 外部情報の検索と活用
• Gitブランチの作成と管理

ReviewerAgent

役割：生成されたコードの品質をレビューし、改善提案を行う。

主な機能：

• コード品質の評価（可読性、保守性、パフォーマンス）
• セキュリティ問題の検出
• ベストプラクティス準拠の確認
• 潜在的なバグの発見
• LGTM（Looks Good To Me）の判定

AgentCoordinator

役割：ProgrammerAgentとReviewerAgentの開発サイクルを管理。

主な機能：

• 開発サイクルの制御
• エージェント間の情報伝達
• 反復的改善プロセスの管理
• 最終成果物の統合

Tool（Function）とは

Tool（Function） は、エージェントが外部システムや環境と相互作用するための機能です。LangChainでは、これらをStructured Toolとして定義します。

Toolの基本構造

LangChainでToolを作成する際は、以下の要素を定義する必要があります。

1. 入力スキーマの定義

まず、Toolが受け取るパラメータの型と説明をPydanticモデルで定義します。

from pydantic import BaseModel, Field

class ToolInput(BaseModel):
    """Tool input schema definition"""
    parameter: str = Field(
        description="Parameter description for LLM to understand usage"
    )
    optional_param: int = Field(
        default=10,
        description="Optional parameter with default value"
    )

2. Tool実装の基本パターン

本プロジェクトでは、統一性を保つために以下のような基底クラスを使用しています。

from abc import ABC, abstractmethod
from typing import Type, Dict, Any
from langchain_core.tools import StructuredTool

class BaseFunction(ABC):
    """Base class for all function tools"""
    
    @classmethod
    def function_name(cls) -> str:
        """Return function name based on class name"""
        return cls.__name__.replace("Function", "").lower() + "_function"
    
    @staticmethod
    @abstractmethod
    def execute(*args, **kwargs) -> Dict[str, Any]:
        """Execute the function logic"""
        pass
    
    @classmethod
    @abstractmethod
    def to_tool(cls: Type["BaseFunction"]) -> StructuredTool:
        """Convert to LangChain StructuredTool"""
        pass

3. 具体的なTool実装例

class ReadFileInput(BaseModel):
    """Input schema for ReadFileFunction"""
    filepath: str = Field(description="Path to the file to read")

class ReadFileFunction(BaseFunction):
    """Function to read file contents"""

    @staticmethod
    def execute(filepath: str) -> Dict[str, str]:
        """Read the contents of a file
        
        Args:
            filepath (str): Path to the file to read
            
        Returns:
            Dict[str, str]: File contents or error message
        """
        try:
            if os.path.exists(filepath):
                with open(filepath, "r", encoding="utf-8") as f:
                    contents = f.read()
                return {
                    "filepath": filepath,
                    "file_contents": contents,
                }
            else:
                return {
                    "filepath": filepath,
                    "error": "File not found.",
                }
        except Exception as e:
            return {
                "filepath": filepath,
                "error": f"Error reading file: {str(e)}",
            }

    @classmethod
    def to_tool(cls: Type["ReadFileFunction"]) -> StructuredTool:
        """Convert to LangChain StructuredTool"""
        return StructuredTool.from_function(
            name=cls.function_name(),  # "readfile_function"
            description="Reads the specified file and returns its contents.",
            func=cls.execute,
            args_schema=ReadFileInput,
        )

実装編：Agentの構築方法

Tool（Function）の定義

一覧

実装したTool一覧は以下の通りです。

Tool名	機能	用途
GetFilesListFunction	ファイル一覧取得	プロジェクト構造の把握
ReadFileFunction	ファイル読み取り	既存コードの確認
MakeNewFileFunction	新規ファイル作成	コードファイルの生成
OverwriteFileFunction	ファイル上書き	既存ファイルの更新
CreateBranchFunction	Gitブランチ作成	作業ブランチの管理
ExecPytestTestFunction	単体テスト実施(pytest実行)	単体テストを実行する
GenerateDiffFunction	差分生成	変更内容の確認
ReviewCodeFunction	コードレビュー	品質チェック
RecordLgtmFunction	LGTM記録	レビュー承認の記録
GoogleSearchFunction	Web検索	技術情報の取得
OpenUrlFunction	URL開く	外部リソースの参照

具体的な定義

Toolの具体的な実装例を以下に示します。

ファイル読み取りTool

class ReadFileFunction(BaseFunction):
    """Function to read a file"""

    @staticmethod
    def execute(filepath: str) -> Dict[str, str]:
        """Read the contents of a file"""
        if os.path.exists(filepath):
            with open(filepath, "r", encoding="utf-8") as f:
                contents = f.read()
            return {
                "filepath": filepath,
                "file_contents": contents,
            }
        else:
            return {
                "filepath": filepath,
                "error": "File not found.",
            }

    @classmethod
    def to_tool(cls: Type["ReadFileFunction"]) -> StructuredTool:
        return StructuredTool.from_function(
            name=cls.function_name(),
            description="Reads the specified file and returns its contents.",
            func=cls.execute,
            args_schema=ReadFileInput,
        )

ファイル上書きTool

class OverwriteFileFunction(BaseFunction):
    """Function to overwrite a file"""

    @staticmethod
    def execute(filepath: str, new_text: str) -> Dict[str, str]:
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(new_text)

        return {"result": "success"}

    @classmethod
    def to_tool(cls: Type["OverwriteFileFunction"]) -> StructuredTool:
        return StructuredTool.from_function(
            name=cls.function_name(),
            description="Overwrites the specified file with new content.",
            func=cls.execute,
            args_schema=OverwriteFileInput,
        )

新規ファイル作成Tool

class MakeNewFileFunction(BaseFunction):
    """Function to create a new file"""

    @staticmethod
    def execute(filepath: str, file_contents: str) -> Dict[str, str]:
        directory = os.path.dirname(filepath)
        if not os.path.exists(directory):
            os.makedirs(directory, exist_ok=True)

        with open(filepath, "w", encoding="utf-8") as f:
            f.write(file_contents)

        return {"result": "success"}

    @classmethod
    def to_tool(cls: Type["MakeNewFileFunction"]) -> StructuredTool:
        return StructuredTool.from_function(
            name=cls.function_name(),
            description="Creates a new file and writes the specified content to it.",
            func=cls.execute,
            args_schema=MakeNewFileInput,
        )

コードレビューTool

class ReviewCodeFunction(BaseFunction):
    """Tool for performing code reviews."""

    @staticmethod
    def execute(diff: str, programmer_comment: str | None = None) -> dict[str, Any]:
        """Assumes review results are generated by LLM FunctionCalling.

        This only returns a stub response without raising exceptions.
        """
        return {
            "message": "Handled by LLM FunctionCalling. No real processing done here."
        }

    @classmethod
    def to_tool(cls: type["ReviewCodeFunction"]) -> StructuredTool:
        """Create the tool."""
        return StructuredTool.from_function(
            name=cls.function_name(),
            description="Reviews pull request code diffs, summarizes issues and improvements, and determines LGTM approval.",
            func=cls.execute,
            args_schema=ReviewerInput,
        )

Agentの構築

本システムでは以下の2つのエージェントを実装しています。

1. ProgrammerAgent：コード生成を担当
2. ReviewerAgent：コードレビューを担当

ProgrammerAgent

使用するTools

• GetFilesListFunction：プロジェクト構造の把握
• ReadFileFunction：既存コードの確認
• MakeNewFileFunction：新規ファイル作成
• OverwriteFileFunction：既存ファイルの更新
• GoogleSearchFunction：技術情報の検索
• CreateBranchFunction：作業ブランチの作成
• GenerateDiffFunction：変更差分の生成

実装

class ProgrammerAgent:
    """Code generation agent."""
    
    def __init__(self, default_project_root: str = "src/"):
        self.llm_client = AzureOpenAIClient()
        self.chat_llm = self.llm_client.initialize_chat()
        self.default_project_root = default_project_root
        
        self.tools = self._initialize_tools()
        self.agent_executor = self._initialize_executor(default_project_root)
    
    def _initialize_tools(self) -> list[BaseTool]:
        """Initialize tools for programmer agent."""
        return [
            GetFilesListFunction.to_tool(),
            ReadFileFunction.to_tool(),
            OverwriteFileFunction.to_tool(),
            MakeNewFileFunction.to_tool(),
            GoogleSearchFunction.to_tool(),
            OpenUrlFunction.to_tool(),
            CreateBranchFunction.to_tool(),
            GenerateDiffFunction.to_tool(),
        ]
    
    def _initialize_executor(self, project_root: str) -> AgentExecutor:
        """Initialize agent executor."""
        system_message = f"""
        You are a professional programmer.
        Your project root is: {project_root}
        
        Please follow these guidelines:
        1. Write clean, maintainable code
        2. Follow best practices for the target language
        3. Add appropriate comments and documentation
        4. Handle errors gracefully
        5. Create files in the specified project root
        """
        
        prompt = ChatPromptTemplate.from_messages([
            SystemMessage(content=system_message),
            MessagesPlaceholder(variable_name="chat_history", optional=True),
            HumanMessagePromptTemplate.from_template("{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ])
        
        agent = create_openai_tools_agent(
            llm=self.chat_llm,
            tools=self.tools,
            prompt=prompt,
        )
        
        return AgentExecutor(
            agent=agent, 
            tools=self.tools, 
            max_iterations=30, 
            verbose=True
        )
    
    def run(self, instruction: str, reviewer_comment: str = None) -> str:
        """Execute programmer agent.
        
        Args:
            instruction (str): Programming instruction
            reviewer_comment (str, optional): Feedback from reviewer
            
        Returns:
            str: Agent output
        """
        if reviewer_comment:
            input_text = f"{instruction}\n\n[Reviewer Feedback]:\n{reviewer_comment}"
        else:
            input_text = instruction
        
        result = self.agent_executor.invoke({"input": input_text})
        return result["output"]

ReviewerAgent

使用するTools

• ReviewCodeFunction：コードレビューの実行
• RecordLgtmFunction：承認状態の記録
• ExecPytestTestFunction：Pytestによる単体テストの実装

class ReviewerAgent:
    """Code review agent."""
    
    def __init__(self):
        self.llm_client = AzureOpenAIClient()
        self.chat_llm = self.llm_client.initialize_chat()
        self.tools = self._initialize_tools()
        self.agent_executor = self._initialize_executor()
    
    def _initialize_tools(self) -> list[BaseTool]:
        """Initialize tools for reviewer agent."""
        return [
            ReviewCodeFunction.to_tool(),
            RecordLgtmFunction.to_tool(),
            ExecPytestTestFunction.to_tool(),
        ]
    
    def _initialize_executor(self) -> AgentExecutor:
        """Initialize agent executor."""
        prompt = ChatPromptTemplate.from_messages([
            SystemMessage(content="""
            You are a professional code reviewer.
            
            Review code from these perspectives:
            - Code quality (readability, maintainability, performance)
            - Security issues
            - Best practice compliance
            - Potential bugs
            - Design issues
            
            Important: If the code has no issues and can be approved, 
            call the record_lgtm_function tool to record LGTM.
            """),
            MessagesPlaceholder(variable_name="chat_history", optional=True),
            HumanMessagePromptTemplate.from_template("{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ])
        
        agent = create_openai_tools_agent(
            llm=self.chat_llm,
            tools=self.tools,
            prompt=prompt,
        )
        
        return AgentExecutor(
            agent=agent, 
            tools=self.tools, 
            max_iterations=30, 
            verbose=True
        )
    
    def run(self, reviewer_input: ReviewerInput) -> ReviewerOutput:
        """Execute code review.
        
        Args:
            reviewer_input (ReviewerInput): Review input
            
        Returns:
            ReviewerOutput: Review result
        """
        # Reset LGTM status
        RecordLgtmFunction.reset_lgtm()
        
        input_text = f"""
        Please perform a code review.
        
        Diff:
        {reviewer_input.diff}
        
        Programmer comment: {reviewer_input.programmer_comment or "None"}
        """
        
        agent_result = self.agent_executor.invoke({"input": input_text})
        output_text = agent_result["output"]
        
        lgtm_flag = RecordLgtmFunction.lgtm()
        
        return ReviewerOutput(
            summary=str(output_text),
            suggestions=[],
            lgtm=lgtm_flag,
        )

Agentによる処理フローと実装

MultiAgent処理

class AgentCoordinator:
    """Coordinator for agent collaboration."""
    
    def development_cycle(
        self, 
        instruction: str, 
        max_iterations: int = 3,
        auto_create_branch: bool = True
    ) -> dict:
        """Execute development cycle."""
        
        # Branch creation
        if auto_create_branch:
            branch_name = self.generate_branch_name(instruction)
            self.create_working_branch(branch_name)
        
        reviewer_output = None
        
        # Development iterations
        for i in range(max_iterations):
            # Code generation
            programmer_output = self.run_programmer(
                instruction, 
                reviewer_comment=reviewer_output.summary if reviewer_output else None
            )
            
            # Code review
            reviewer_output = self.run_reviewer(
                programmer_comment=f"Iteration {i+1} completed"
            )
            
            # Check for approval
            if reviewer_output.lgtm:
                break
        
        return {
            "programmer_output": programmer_output,
            "reviewer_output": reviewer_output.summary,
            "branch_name": self.working_branch,
        }

実行例と動作確認

入力例

基本的な使用方法

uv run python src/main.py coordinator "Create a simple REST API using FastAPI"

詳細な指示の例

uv run python src/main.py coordinator "
Create a user management API with the following features:
1. User registration and authentication
2. CRUD operations for user profiles
3. JWT token-based authentication
4. Input validation using Pydantic
5. SQLite database integration
6. Comprehensive error handling
"

出力例

ProgrammerAgent 実行ログ

INFO - === Development cycle 1/3 ===
> Entering new AgentExecutor chain...
Invoking: `get_files_list_function` with `{'include_patterns': ['src/agent/rules/coding_rule.md']}`
{'files_list': ['src/agent/rules/coding_rule.md']}
Invoking: `read_file_function` with `{'filepath': 'src/agent/rules/coding_rule.md'}`
:
:
Invoking: `make_new_file_function` with `{'filepath': 'generated_code/database.py', 'file_contents': '・・・省略

ReviewerAgent 実行ログ

Invoking: `review_code_function` with `{'diff': 'diff --git a/generated_code/auth.py b/generated_code/auth.py\nnew file mode 100644\nindex 0000000..0031cd6\n--- /dev/null\n+++ b/generated_code/auth.py\n+"""\n+auth.py\n+Authentication utility functions and security dependencies.\n+"""\n+\n+from passlib.context import CryptContext\n+from datetime import datetime, timedelta\n+from jose import JWTError, jwt\n+fr・・・省略

Invoking: `record_lgtm_function` with `{}`
responded: ## Code Review

### Code Quality (Readability, Maintainability, Performance)

- The code is generally clean, modular, and well-commented with clear docstrings.
- File organization and structure follow common FastAPI and SQLAlchemy project conventions.
- Function and variable names are descriptive and readable.
- Docstrings use a consistent format.
- CRUD logic is separate from API endpoints, promoting maintainability.
- Relevant dependency injection is utilized for DB sessions.
- No redundant code observed.
- Performance is not likely to be a major concern for typical user management scenarios here.
：
省略

生成されるファイル例

generated_code
├── auth.py
├── crud.py
├── database.py
├── exceptions.py
├── main.py
├── models.py
└── schemas.py

実際に生成されたコード（main.py）

"""
main.py
Entrypoint for FastAPI application for User Management API.
"""
from fastapi import FastAPI, Depends, status, HTTPException
from sqlalchemy.orm import Session
from . import models, schemas, auth, crud, exceptions
from .database import engine, getDbSession
from fastapi.security import OAuth2PasswordRequestForm
from typing import List
import uvicorn

# Create all database tables
models.Base.metadata.create_all(bind=engine)

app = FastAPI(title="User Management API")

# Register global error handlers
exceptions.registerExceptionHandlers(app)

@app.post("/register", response_model=schemas.UserResponse, status_code=status.HTTP_201_CREATED)
def registerUser(userCreate: schemas.UserCreate, db: Session = Depends(getDbSession)):
    """
    API endpoint for user registration.
    """
    user = crud.createUser(db, userCreate)
    return user

@app.post("/login", response_model=schemas.Token)
def loginUser(formData: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(getDbSession)):
    """
    User authentication and JWT issuance.
    """
    user = crud.authenticateUser(db, formData.username, formData.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect username or password.")
    accessToken = auth.createAccessToken(data={"sub": user.username})
    return schemas.Token(accessToken=accessToken)

@app.get("/users/me", response_model=schemas.UserResponse)
def getMyProfile(currentUser: models.User = Depends(auth.getCurrentUser)):
    """
    Return user profile for requester.
    """
    return currentUser

@app.patch("/users/me", response_model=schemas.UserResponse)
def updateMyProfile(userUpdate: schemas.UserBase, db: Session = Depends(getDbSession), currentUser: models.User = Depends(auth.getCurrentUser)):
    """
    Update profile info (email, fullName) for requester.
    """
    updatedUser = crud.updateUser(db=db, user=currentUser, userUpdate=userUpdate)
    return updatedUser

@app.delete("/users/me", status_code=status.HTTP_204_NO_CONTENT)
def deleteMyProfile(db: Session = Depends(getDbSession), currentUser: models.User = Depends(auth.getCurrentUser)):
    """
    Delete user profile for requester.
    """
    crud.deleteUser(db, currentUser)
    return None

@app.get("/users", response_model=List[schemas.UserResponse])
def getAllUsers(db: Session = Depends(getDbSession), currentUser: models.User = Depends(auth.getCurrentUser)):
    """
    Admin/debug endpoint: retrieve all users.
    """
    return crud.listUsers(db)

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

まとめ

本記事では、LangChainとAzure OpenAI GPT-4.1を使用してゼロからコーディングエージェントを構築する方法を詳しく解説しました。実際にゼロから作成してみて、Cursorなどの既存ツールの完成度の高さを改めて実感しました。

何か質問だったり、指摘等ありましたらお気軽にコメントいただけますと幸いです。

今後実現したいこと

• Coding AgentのAPIサーバー化：Web APIとして提供し、他のシステムとの連携を容易にする
• ドキュメントインデックス化：ドキュメントに対してインデックスを作成し、読み取りコストを削減する
• 他のLLMClient実装：GPT-4.1以外も実装した上で、抽象化したLLMClient Classを実装する

参考文献・リンク集

公式ドキュメント

• LangChain Documentation
• OpenAI API Reference
• Azure OpenAI Service

フレームワーク・ライブラリ

• uv - Python Package Manager
• Pydantic - Data Validation
• Typer - CLI Framework
• GitPython - Git Integration