はじめに
こんにちは。ナウキャストでデータエンジニアをしているTakumiです。
社内(Finatext HD内)の生成AIコンテストでMultiAgentを利用したシステムをスクラッチで構築しました。
具体的には、ユーザーがSlackでメッセージを送信し、コードの記述、レビュー、GitHubでのPR作成までEnd2Endでできるシステムです。
コンテストで構築したシステムの概要図は以下の通りです。
本記事では、複数のエージェントが協調して動作する本格的なコーディングAgent（Coodinator） に絞って、構築した概要を説明します。
この記事でわかること
この記事では、LangChainとAzure OpenAIを使用してゼロからコーディングエージェントを構築する方法を詳しく解説します。
具体的には以下の内容を学ぶことができます。
- コーディングエージェントの実装方法：ProgrammerAgentとReviewerAgentを使ったMultiAgentシステムの構築方法
- Tool（Function）の設計と実装：エージェントが使用する機能の定義方法
この記事で紹介しているAgentについては以下のリポジトリで公開しています。
実行環境
- 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つのエージェントを実装しています。
- ProgrammerAgent：コード生成を担当
- 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を実装する
Discussion