はじめまして
ふっきーです。
理想の開発環境を目指して、LangGraphを使ってコーディングエージェントを構築していきたいと思います。ざっくり以下のような流れにしようと思っています。
1. 開発プロジェクトの分析
2. 影響範囲の調査・変更対象ファイルのリストアップ
3. 設計＆設計文書の作成
4. 設計文書のレビュー（人間）
5. 実装
実際の開発プロセスもざっくりこんなもんかなと思っています。
今日はその記念すべき1回目で、プロジェクトの分析エージェントを作っていきます。
プロジェクト分析エージェントの役割
プロジェクト分析エージェントは、開発対象のプロジェクトがどのようなプロジェクトか、どのファイルになにが書かれているのかを分析して、1つのドキュメントにまとめます。
この分析結果を参照して、次の工程の影響範囲の調査をしていく感じになります。
構成
プロジェクト分析エージェントは以下のような構成です。
- decideノード：分析するか判定する
- file_alalysisノード：1つのファイルの役割とクラスやメソッド、依存関係を抽出する
- summaryノード：ファイルごとの分析結果を1つのファイルにまとめる
decideノード
プロジェクト分析を実施するかどうかを判定するノードです。
プロジェクト内のソースコード等のファイルすべてをスキャンするので、毎回確定で分析していたら、財布が焼けます。なので、Skipできたらします。
しかし！
"分析したことがなければ分析" / "すでに分析されていれば、分析しない"というがばがば判定です。
変更のあったファイル部分だけ、分析しなおして更新するとうの修正が今後必要そうですが、後回しにします。
ノードの実装
def decide_node(state: State):
"""プロジェクトの分析状態を判断し、分析対象ファイルを特定するノード"""
project_path = state["project_path"]
analysis_file = os.path.join(project_path, "project_analysis.md")
if os.path.exists(analysis_file):
return {
"agent_response": "Already analyzed."
}
# 分析対象のファイルパスを取得
file_paths = git_ls_files()
# current_phaseの更新は行わず、必要なデータのみ返す
return {
"file_paths": file_paths,
"analysis_results": []
}
ファイルリストアップ関数
def git_ls_files() -> List[str]:
"""Lists all files tracked by Git, excluding hidden files and directories."""
try:
result = subprocess.run(
['git', 'ls-files'],
capture_output=True,
text=True,
check=True
)
all_files = result.stdout.strip().split('\n')
visible_files = [
f for f in all_files
if not f.startswith('.')
and not any(part.startswith('.') for part in f.split('/'))
]
return visible_files
except subprocess.CalledProcessError as e:
raise RuntimeError(f"Git command failed: {e}")
except Exception as e:
raise RuntimeError(f"Error listing files: {e}")
そうです。スキャン対象はGitで構成管理されているファイルのみです。
Gitで構成管理していないプロジェクト？？知らんなぁ
file_alalysisノード
ファイルを分析するノードです。
分析はエージェントにまかせます。ファイルから抽出する情報は、
- ファイルの概要
- クラスとその説明
- 関数とその説明
- 依存しているライブラリ
です。
FileAnalysisAgent関係のコード
自作の基底クラスを使っていて抽象化されているのですが、雰囲気はつかめると思います。
class CodeComponent(BaseModel):
name: str = Field(description="クラスまたは関数の名前")
description: str = Field(description="コンポーネントの簡単な説明")
class FileAnalysis(BaseModel):
file_path: str = Field(description="分析対象のファイルパス")
overview: str = Field(description="ファイル全体の説明")
classes: List[CodeComponent] = Field(description="定義されているクラスのリスト")
functions: List[CodeComponent] = Field(description="定義されている関数のリスト")
dependencies: List[str] = Field(description="依存しているモジュールやパッケージのリスト")
class FileAnalysisAgent(BaseLLM):
name: str = "file_analysis"
desc: str = "Analyzes a source code file structure and its components"
def __init__(self, llm):
system_prompt = """
You are a source code analysis expert.
Your role is to analyze the content of specified files and understand their main components and dependencies.
Please extract the following information:
1. Overview description of the file
2. List of defined classes with a concise one-line description for each class
3. List of defined functions with a concise one-line description for each function
4. List of modules or packages the file depends on
5. Explanation of necessary specifications
Important notes:
- Output in English
- Extract all classes and functions without omission
- Describe each class and function concisely in one line
- Extract dependencies accurately from import statements
- Provide a concise and accurate overview
"""
super().__init__(
llm=llm,
system_prompt=system_prompt,
output_structure= FileAnalysis
)
def analyze(self, file_path: str, content: str) -> FileAnalysis:
"""
Analyzes the content of the specified file.
Args:
file_path: Path to the file being analyzed
content: Content of the file
Returns:
FileAnalysis: Structured analysis results
"""
query = f"""
Please analyze the following file:
File path: {file_path}
File content:
{content}
Please provide a concise one-line description of each class and function.
"""
return self._invoke(query)
def format_analysis(self, analysis: FileAnalysis) -> str:
"""
Converts structured analysis results to string format.
Args:
analysis: Analysis results
Returns:
str: Formatted analysis results as a string
"""
classes_str = "\n - ".join(
[""] + [f"{comp.name}: {comp.description}" for comp in analysis.classes]
) if analysis.classes else " None"
functions_str = "\n - ".join(
[""] + [f"{comp.name}: {comp.description}" for comp in analysis.functions]
) if analysis.functions else " None"
dependencies_str = "\n - ".join(
[""] + analysis.dependencies
) if analysis.dependencies else " None"
return f"""=== {analysis.file_path} ===
{analysis.overview}
== Source Code Structure ==
- Classes:{classes_str}
- Functions:{functions_str}
== Dependencies ==
This file depends on the following modules/packages:{dependencies_str}
"""
情報が落ちないようにoutput_strucureを定義しています。また、出力を1つの構造化されたテキストに直すためのメソッドformat_analysisも用意しています。トークン数を抑えるため、英語の出力です。
ノードの実装
def analyze_file(state: State) -> Dict[str, Any]:
"""単一ファイルの分析を行うノード"""
file_path = state["file_path"]
agent = FileAnalysisAgent(llm=OPENAI_o3_MINI)
try:
# テキストファイルとして読み込みを試みる
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
result = agent.analyze(file_path, content)
formatted_result = agent.format_analysis(result)
# current_phaseの更新は行わず、analysis_resultsのみ返す
return {
"analysis_results": [formatted_result]
}
except Exception as e:
print(f"Error analyzing file {file_path}: {e}")
return {}
あたりまえですが、プロジェクト内にある多数のファイルを直列で分析していると、処理時間が長くなってしまいます。なので、このノードは並列実行しています。そこら辺の実現方法は後述します。
summarizeノード
各ファイルの分析結果をまとめるノードです。
ただ、連結させて、project_analysis.mdというファイルに保存しているだけです。
連結させた情報からプロジェクト全体の概要を追記するぐらいはしてもよかったかもしれません。
ノードの実装
def summary_node(state: State):
"""分析結果をまとめて保存するノード"""
project_path = state["project_path"]
analysis_results = state["analysis_results"]
# 全ての分析結果を1つにまとめる
# 各結果が文字列であることを確認
string_results = [str(result) for result in analysis_results]
# 結果が空でないか確認
if not string_results:
print("Warning: No analysis results to save!")
string_results = ["No analysis results were generated."]
# 結果を結合
combined_analysis = "\n\n".join(string_results)
# project_analysis.mdに保存
analysis_file = os.path.join(project_path, "project_analysis.md")
with open(analysis_file, 'w', encoding='utf-8') as f:
f.write(combined_analysis)
return {
"agent_response": "Analysis completed and saved to project_analysis.md"
}
グラフ全体
とりあえず実装を見てみます。
class State(TypedDict):
queue: Queue
context: ContextRepository
project_path: str
current_phase: Annotated[str, operator.add] # 複数のノードから更新される可能性があるため、operator.addで注釈
agent_response: str
file_paths: List[str] # 分析対象のファイルパスリスト
analysis_results: Annotated[list, operator.add] # 分析結果のリスト
file_path: Optional[str] # 現在分析中のファイルパス
def build_graph():
"""グラフを構築する関数"""
workflow = StateGraph(State)
# ノードの追加
workflow.add_node("decide", decide_node)
workflow.add_node("analyze_file", analyze_file)
workflow.add_node("summary", summary_node)
# エントリーポイントの設定
workflow.set_entry_point("decide")
# 分析ファイルの処理を実行
workflow.add_conditional_edges(
"decide",
initiate_file_analysis,
["analyze_file"]
)
# 分析ファイルからサマリーへ
workflow.add_edge("analyze_file", "summary")
# サマリー処理後は終了
workflow.add_edge("summary", END)
return workflow.compile()
ポイントは、analyze_fileノードを並列実行するために、add_conditional_edgesを使っているところかなと思います。この関数は、decideノードから次に行くときに、initiate_file_analysis関数を使って、どのノードに行くかを決定しています。
なので、initiate_file_analysis関数の中を見てみましょう。
def initiate_file_analysis(state: State) -> List[Send]:
"""ファイル分析タスクを並列実行するためのエッジ関数"""
return [
Send("analyze_file", {"file_path": file_path})
for file_path in state["file_paths"]
]
意外とすっきり、こんな感じです。
LangGraphでは、SendAPIを使ってノードの並列実行を実現すると楽です。
forループの数だけanalyze_fileノードを実行するという感じです。その際に、第二引数で、各ノードで使用するStateを渡しています。
SendAPIを使うメリットは、並列数に動的に対応できるところですね
並列実行のポイントはもう一つあります。それは、すでに出てきたStateにあります。
並列実行されているanalyze_fileノードがそれぞれ別の出力で、同じState(ここではanalysis_results)を更新しようとします。上書き合戦が起こっては意味がないので、Stateをリストにして、更新時にリストに追加するようにしないといけないです。
class State(TypedDict):
queue: Queue
...
analysis_results: Annotated[list, operator.add] # 分析結果のリスト
file_path: Optional[str] # 現在分析中のファイルパス
Annotated[list, operator.add]とすると、更新時にリストに追加されるようになります。
最終的にanalysis_resultsは、並列実行された各ノードの分析結果のリストが格納されます。
これでいい感じにプロジェクトの分析ができるようになりました。
出力サンプル
...
=== analysis_project/agents.py ===
This file defines data models and an agent for analyzing source code files using a language model, extracting file overviews, defined classes and functions, and their dependencies.
== Source Code Structure ==
- Classes:
- CodeComponent: A Pydantic model representing a code element (class or function) with its name and a brief description.
- FileAnalysis: A Pydantic model that structures the analysis result including file path, overview, classes, functions, and dependencies.
- FileAnalysisAgent: An agent class extending BaseLLM that analyzes a file by invoking a language model with a detailed prompt and formats the analysis output.
- Functions:
- __init__: Initializes the FileAnalysisAgent with a system prompt and schema for file analysis.
- analyze: Analyzes the provided file content and path to generate a structured FileAnalysis result using the language model.
- format_analysis: Formats a FileAnalysis object into a human-readable string detailing the file structure and dependencies.
== Dependencies ==
This file depends on the following modules/packages:
- pydantic.BaseModel
- pydantic.Field
- typing.List
- typing.Dict
- llm.base_llm.BaseLLM
=== analysis_project/assistant.py ===
This file defines an analysis assistant function that builds and executes a workflow graph using a project path and message queue to generate an agent response.
== Source Code Structure ==
- Classes: None
- Functions:
- analysis_assistant: Constructs and invokes a workflow using the provided project path and message queue to obtain and return an agent response.
== Dependencies ==
This file depends on the following modules/packages:
- queue
- .graph (build_graph, State)
- .context (ContextRepository)
=== analysis_project/context.py ===
This file defines a ContextRepository class that stores and retrieves project info, analysis results, and design analysis details.
...
こんな感じです。これを人間が見るのはしんどいかもですが、LLMに入力させるのが主目的なのでこれでいいです。
最後に
いろいろ改善点ありますが、プロジェクトの分析エージェントが完成しました。
次は、
2. 影響範囲の調査・変更対象ファイルのリストアップ
ですかね。内容的には、わざわざマルチエージェントなグラフにする必要はなさそうなので、わりかしすぐに実現できるかもです。
最後までいったら、コードをGitに公開しようかなぁと思っています。
以上、最後まで読んでいただきありがとうございました！！！！！