LangChainで複数AIエージェント連携｜CI/CD自動化【実装3ステップ】

このやり方で何ができるか

企業のコードベース保守やセキュリティ監査では、数百万行のソースコード全体を高速に解析する必要があります。Claude Code のエージェント機能を活用することで、50並列実行で20時間以内に完了させることが可能になります。

これにより以下が実現します。

• 処理時間の短縮: 単一実行の数分の一に削減
• コスト効率化: バッチ間の待機時間削減で不要な課金を防止
• スケーラビリティ: 今後さらに大規模なコード解析にも対応可能

準備するもの

必要なソフトウェア

• Claude Code または Anthropic 公式 API（Claude 3系以上）- エージェントとして動く AI。API キーが必要
• Python 3.9以上 - プログラムを書く環境
• asyncio（標準装備）- 非同期処理ライブラリ
• httpx または requests - HTTP クライアント
• APIレート制限に対応したキューイングシステム - asyncio で統合可能

ファイル・権限の準備

1. 開発用のフォルダを新規作成
2. API キーを環境変数に設定（キーを直接コードに書かない）
3. 解析対象のコードディレクトリを用意する

手順

1. エージェント実行ユニットの設計（実装時間：2〜3時間）

Claude Code エージェントを複数並列で動かす場合、各エージェントが独立したタスクを処理する構造が必要です。

各エージェントは以下の役割を持つことが一般的です。

• 検査エージェント - コードの品質・セキュリティをチェック
• テストエージェント - プログラムが正しく動くか確認
• 修正エージェント - 見つかった問題を直す
• レポートエージェント - 結果をまとめる

それぞれが独立して動きますが、次のステップに移る前に「承認」が必要な設計にします。

以下は基本的なエージェント実行関数の例です。

import asyncio
import httpx
from typing import Optional, Dict, Any

async def run_code_analysis_agent(
    file_chunk: list[str],
    agent_id: int,
    api_key: str,
    analysis_type: str = "security"
) -> Dict[str, Any]:
    """
    単一のエージェント実行ユニット
    file_chunk: 解析対象のコード行のリスト
    agent_id: エージェントの識別番号（1〜50）
    """
    
    async with httpx.AsyncClient(timeout=300) as client:
        try:
            payload = {
                "model": "claude-3-5-sonnet-20241022",
                "max_tokens": 2000,
                "messages": [
                    {
                        "role": "user",
                        "content": f"""
                        以下のコード{len(file_chunk)}行を解析してください。
                        分析対象: {analysis_type}
                        
                        {chr(10).join(file_chunk)}
                        
                        以下の形式でJSON結果を返してください:
                        {{
                            "issues": [
                                {{"line": int, "type": str, "severity": str, "description": str}}
                            ],
                            "summary": str
                        }}
                        """
                    }
                ]
            }
            
            headers = {
                "x-api-key": api_key,
                "content-type": "application/json"
            }
            
            response = await client.post(
                "https://api.anthropic.com/v1/messages",
                json=payload,
                headers=headers
            )
            
            result = response.json()
            
            return {
                "agent_id": agent_id,
                "status": "success",
                "lines_processed": len(file_chunk),
                "response": result.get("content", [{}])[0].get("text", ""),
                "timestamp": asyncio.get_event_loop().time()
            }
            
        except Exception as e:
            return {
                "agent_id": agent_id,
                "status": "error",
                "error": str(e),
                "lines_processed": 0
            }

2. ファイル分割と並列実行の管理（実装時間：2〜3時間）

大規模コードベースを50並列で処理するには、事前にファイルを分割し、各エージェントに割り当てる必要があります。

import os
from pathlib import Path

async def distribute_and_process(
    code_directory: str,
    num_agents: int = 50,
    api_key: str = None
):
    """
    ディレクトリ内の全ソースコードを分割・並列処理
    """
    
    if api_key is None:
        api_key = os.environ.get("ANTHROPIC_API_KEY")
    
    # 全ソースファイルの収集
    code_files = []
    for ext in [".py", ".js", ".ts", ".java", ".cpp", ".go"]:
        code_files.extend(Path(code_directory).glob(f"**/*{ext}"))
    
    print(f"発見されたコードファイル: {len(code_files)}個")
    
    total_lines = 0
    file_contents = []
    
    for file_path in code_files:
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                lines = f.readlines()
                total_lines += len(lines)
                file_contents.append({
                    "path": str(file_path),
                    "lines": lines
                })
        except Exception as e:
            print(f"ファイル読み込みエラー {file_path}: {e}")
    
    print(f"合計行数: {total_lines:,}")
    
    # チャンク単位での分割
    lines_per_agent = max(total_lines // num_agents, 1000)
    chunks = []
    current_chunk = []
    
    for file_data in file_contents:
        for line in file_data["lines"]:
            current_chunk.append(line)
            if len(current_chunk) >= lines_per_agent:
                chunks.append(current_chunk)
                current_chunk = []
    
    if current_chunk:
        chunks.append(current_chunk)
    
    print(f"分割チャンク数: {len(chunks)}")
    
    # 並列実行
    tasks = [
        run_code_analysis_agent(
            chunk,
            i,
            api_key,
            analysis_type="security"
        )
        for i, chunk in enumerate(chunks[:num_agents])
    ]
    
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    return results

3. API レート制限への対応（設定時間：1〜2時間）

Anthropic API は分単位のレート制限（標準で約600リクエスト/分が目安）があります。50並列実行時は、リクエストをキューイングしてレート制限を回避する必要があります。

制御しないと起こるトラブル:

• 429エラー（Too Many Requests）が多発する
• 予定外のリトライで余分な課金が発生する
• 処理が途中で止まりデータが不整合になる

安全な設計のコツ:

1. セマフォで同時実行数を制限（最大10並列を目安に）
2. 各エージェントの権限を制限（API呼び出し回数など）
3. 実行前に「チェックポイント」を挟む
4. ログに記録して後から検証できるようにする

import asyncio
from collections import deque

class RateLimitedExecutor:
    """APIレート制限に対応した実行管理"""
    
    def __init__(self, requests_per_minute: int = 600):
        self.requests_per_minute = requests_per_minute
        self.min_interval = 60 / requests_per_minute
        self.last_request_time = 0
        self.queue = deque()
    
    async def execute_with_rate_limit(self, coro):
        current_time = asyncio.get_event_loop().time()
        time_since_last = current_time - self.last_request_time
        
        if time_since_last < self.min_interval:
            await asyncio.sleep(self.min_interval - time_since_last)
        
        self.last_request_time = asyncio.get_event_loop().time()
        return await coro
    
    async def batch_execute(self, tasks: list, max_concurrent: int = 10):
        """
        タスクをバッチで実行
        max_concurrent: 同時実行数（APIのレート制限に基づいて調整）
        """
        
        results = []
        semaphore = asyncio.Semaphore(max_concurrent)
        
        async def bounded_task(task):
            async with semaphore:
                return await self.execute_with_rate_limit(task)
        
        for result in await asyncio.gather(
            *[bounded_task(task) for task in tasks],
            return_exceptions=True
        ):
            results.append(result)
        
        return results

4. 結果の集約と統計（実装時間：1〜2時間）

50個のエージェントから返された結果を統合し、全体の解析サマリーを生成します。

from datetime import datetime
import json

async def aggregate_results(agent_results: list) -> Dict[str, Any]:
    
    successful = [r for r in agent_results if r.get("status") == "success"]
    failed = [r for r in agent_results if r.get("status") == "error"]
    
    total_processed = sum(r.get("lines_processed", 0) for r in successful)
    total_time = max([r.get("timestamp", 0) for r in successful]) - min(
        [r.get("timestamp", float('inf')) for r in successful]
    )
    
    # セキュリティ問題の集約
    all_issues = []
    for result in successful:
        try:
            response_text = result.get("response", "{}")
            if "```json" in response_text:
                json_str = response_text.split("```json")[1].split("```")[0]
            else:
                json_str = response_text
            
            parsed = json.loads(json_str)
            all_issues.extend(parsed.get("issues", []))
        except Exception as e:
            print(f"結果解析エラー (Agent {result.get('agent_id')}): {e}")
    
    severity_counts = {
        "critical": len([i for i in all_issues if i.get("severity") == "critical"]),
        "high": len([i for i in all_issues if i.get("severity") == "high"]),
        "medium": len([i for i in all_issues if i.get("severity") == "medium"]),
        "low": len([i for i in all_issues if i.get("severity") == "low"])
    }
    
    return {
        "timestamp": datetime.now().isoformat(),
        "summary": {
            "total_lines_processed": total_processed,
            "total_agents": len(agent_results),
            "successful_agents": len(successful),
            "failed_agents": len(failed),
            "total_execution_time_hours": total_time / 3600,
            "throughput_lines_per_minute": (total_processed / total_time) * 60 if total_time > 0 else 0
        },
        "issues": {
            "total_count": len(all_issues),
            "by_severity": severity_counts,
            "top_issues": sorted(
                all_issues,
                key=lambda x: {"critical": 0, "high": 1, "medium": 2, "low": 3}.get(x.get("severity"), 4)
            )[:10]
        },
        "performance": {
            "failed_agents": failed
        }
    }

5. 実際に動かしてテストする（テスト時間：2〜3時間）

小さなプロジェクトで試します。

async def main():
    
    code_directory = "./large_codebase"
    num_agents = 50
    
    print("=" * 50)
    print("Claude Code 大規模バッチ処理 開始")
    print("=" * 50)
    
    results = await distribute_and_process(
        code_directory=code_directory,
        num_agents=num_agents,
        api_key=os.environ.get("ANTHROPIC_API_KEY")
    )
    
    aggregated = await aggregate_results(results)
    
    print("\n" + "=" * 50)
    print("処理完了")
    print("=" * 50)
    print(f"処理行数: {aggregated['summary']['total_lines_processed']:,} 行")
    print(f"実行エージェント数: {aggregated['summary']['successful_agents']}/{aggregated['summary']['total_agents']}")
    print(f"実行時間: {aggregated['summary']['total_execution_time_hours']:.2f} 時間")
    print(f"スループット: {aggregated['summary']['throughput_lines_per_minute']:.0f} 行/分")
    print(f"検出問題数: {aggregated['issues']['total_count']}")
    
    with open("analysis_result.json", "w") as f:
        json.dump(aggregated, f, indent=2, ensure_ascii=False)
    
    print("\n結果をanalysis_result.jsonに保存しました")

if __name__ == "__main__":
    asyncio.run(main())

1. テスト用のコードリポジトリを準備
2. エージェントが期待通りに動くか確認
3. ログをチェックして、どの段階で何が起こったか把握
4. 必要に応じてエージェントの指示を修正

つまずきやすいところ

API レート制限エラー（429）

症状: 大量のリクエストで「Too Many Requests」エラーが発生

原因: 同時並列数が API の制限を超えている

解決策: RateLimitedExecutor を使用し、同時実行数を制限してください。

# 不正: 50並列を一気に実行
tasks = [run_code_analysis_agent(...) for _ in range(50)]
await asyncio.gather(*tasks)

# 正: 最大10並列に制限
executor = RateLimitedExecutor()
await executor.batch_execute(tasks, max_concurrent=10)

メモリ不足（OOM）

症状: 大規模ファイル処理時にメモリが枯渇

原因: 全ファイルをメモリに読み込んでいる

解決策: ストリーミング処理またはジェネレータを使用

def read_large_file_streaming(file_path: str, chunk_size: int = 10000):
    chunk = []
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            chunk.append(line)
            if len(chunk) >= chunk_size:
                yield chunk
                chunk = []
    if chunk:
        yield chunk

タイムアウトエラー

症状: 処理時間が長すぎてリクエストがタイムアウト

原因: Claude API のデフォルトタイムアウト（通常30秒）を超えた

解決策: タイムアウト値を適切に延長し、より小さなチャンクに分割

# タイムアウトを300秒に設定
async with httpx.AsyncClient(timeout=300) as client:
    response = await client.post(...)

# または、チャンク単位を減らす
lines_per_agent = 100000  # デフォルト: 500000 行

エージェント間のデータ形式が一致していない

複数のエージェントが返すデータ形式がバラバラだと、次のステップが受け取れません。解決方法は「すべてのエージェントに JSON 形式で返すよう指示する」ことに加え、APIリクエストで temperature=0 を指定して決定性を高めることです。

payload = {
    "model": "claude-3-5-sonnet-20241022",
    "max_tokens": 2000,
    "temperature": 0,  # 決定性を確保
    "messages": [...]
}

API キーの漏洩リスク

コードに直接キーを書くと、誤って公開リポジトリにアップロードする恐れがあります。環境変数（パソコンの設定に保存）を使って、コードには書かないようにしましょう。

ログが膨大になる

すべての動作を記録すると、ファイルサイズが大きくなります。重要な段階だけ詳細ログを取り、細かい部分は簡潔に記録するメリハリが大事です。

慣れてきたら試したいこと

クラウドスケーリング

AWS Lambda や Google Cloud Functions を使用すれば、さらに大規模な並列実行（数百〜数千並列）が可能になります。

def lambda_handler(event, context):
    file_chunk = event["file_chunk"]
    agent_id = event["agent_id"]
    
    result = asyncio.run(run_code_analysis_agent(
        file_chunk, agent_id, os.environ["ANTHROPIC_API_KEY"]
    ))
    
    return result

キャッシュ機能の活用

同じコード解析を複数回実行する場合、前回の結果を保存しておくことで、処理速度を上げられます。Redis 等のキャッシュストレージを活用してハッシュキーで管理すると効果的です。

インクリメンタル処理

システムが稼働中でも解析を継続する場合、前回の処理状態を記録して再開できる仕組みを作ります。

マルチモデル構成

セキュリティチェックと可読性解析など、異なる分析タイプに最適なモデルを使い分けることで、精度とコストを両立できます。

model_mapping = {
    "security": "claude-3-5-sonnet-20241022",  # 高精度必須
    "performance": "claude-3-opus-20250219",    # より詳細な分析
    "style": "claude-3-haiku-20250307"          # コスト優先
}

条件分岐の組み込み

「テスト結果が OK なら修正不要」「エラーの種類によって割り当てるエージェントを変える」など、より複雑な判断をエージェントに任せます。

リアルタイムモニタリング

エージェントの動作をダッシュボード（監視画面）で見守り、問題が起きたらすぐ人間が介入できる仕組みを作ります。

あわせて読みたい

• RAGパイプライン構築の実装手順：LangChainとベクトルDBで作る質問応答システム
• Claude＋GitHub Actionsで自動AIエージェント構築 マルチエージェントシステムの実践
• 本番環境でAIエージェントが暴走するのを防ぐ：料金・間違い・情報漏洩を守る実践ガイド