🌟

Claude-Engineer徹底解剖: AI駆動開発の舞台裏をMermaid図で完全解説

2024/07/18に公開

CLI

はじめに

GitHubで公開されている革新的なAI搭載CLIツール「Claude-Engineer」。それは、Anthropic社の先進的な大規模言語モデル"Claude"のパワーを開発者の手に届ける、まさに"夢のツール"と言えるでしょう。しかし、その洗練された機能の裏側で、どのようにコードが織りなす複雑な連携を実現しているのでしょうか？本稿では、Claude-Engineerのソースコードを徹底的に解き明かし、Mermaid図を用いてその全貌を視覚的に解説します。初心者の方でも理解できるように、コードの動作を一つずつ丁寧に説明していくので、安心して読み進めてください。さあ、AI駆動開発の舞台裏へとご案内しましょう！

全体像

Claude-Engineerは、Anthropic社の提供する大規模言語モデル「Claude」のAPIを利用し、ソフトウェア開発を支援するコマンドラインインターフェースツールです。

ユーザーはClaudeとチャット形式で対話し、以下のような操作を実現できます。

ファイルシステム操作：新規フォルダ/ファイルの作成、ファイルの読み書きなど
コード生成：Claudeに指示を出し、Pythonコードを生成
ウェブ検索：最新の情報を取得
コード実行：生成したコードを仮想環境内で実行し、結果を確認

Claude-Engineerの構造

Claude-Engineerは複数のモジュールが連携して動作する、洗練された設計がされています。以下に、主要なモジュールと、それらの関係性を示すMermaid図を示します。

各モジュールの役割

ユーザーインターフェース: ユーザーからの入力を処理し、対応する処理を実行します。
AIモデル: Claude の各モデル(MAINMODEL, TOOLCHECKERMODEL, CODEEDITORMODEL, CODEEXECUTIONMODEL)は、それぞれ異なる役割を担い、対話処理、ツールの実行と検証、コード編集、コード実行結果の分析などを実行します。
ツール: ファイル操作、コード編集、ウェブ検索、コード実行といった具体的な機能を提供します。
出力: 処理結果をコンソールやファイルに出力します。

モジュール間の関係性

ユーザーインターフェースは、ユーザー入力に基づいてAIモデルを呼び出し、処理を実行します。
AIモデルは、タスクに応じてツールを呼び出したり、他のAIモデルと連携します。
ツールは、AIモデルからの指示を受けて動作し、結果を返します。
出力モジュールは、AIモデルやツールからの出力をフォーマットし、ユーザーに表示します。

Claude-Engineer 関数呼び出しフロー

以下は、Claude-Engineer の主要な関数と、それらの呼び出し関係を表す Mermaid 図です。ユーザー入力から処理結果の出力まで、どのように関数が連携して動作するのかを視覚化しています。

図の説明

main 関数: プログラムのエントリポイント。ユーザー入力を受け取り、コマンドを解析し、処理を適切な関数に委譲します。
コマンド解析: ユーザー入力を解釈し、「exit」「reset」「save chat」「image」「automode」、その他のコマンドに分類します。
制御コマンド: exit, reset, save chat などのコマンドを実行します。
画像処理: image コマンドを受け取ると、encode_image_to_base64 関数を呼び出して画像を Base64 形式にエンコードします。
自動モード制御: automode コマンドを受け取ると、自動モードの処理を実行します。
chat_with_claude 関数: ユーザー入力とコンテキストに基づいて、Claude との対話を処理します。
update_system_prompt 関数: Claude に送信するプロンプトを、現在のコンテキストに応じて動的に生成します。
Anthropic API 呼び出し: 生成されたプロンプトを使用して、Claude API を呼び出します。
ツール呼び出し判定: Claude から返された応答に、ツールの実行指示が含まれているかどうかを判定します。
execute_tool 関数: ツール名に応じて、適切なツール関数 (create_folder, create_file, read_file など) を呼び出します。
ツール関数: 具体的なファイル操作、コード編集、ウェブ検索、コード実行などの処理を行います。
結果出力: ツールの処理結果または Claude からの応答を、ユーザーにわかりやすく出力します。
コンソール出力: 処理結果をコンソールに表示します。

補足

図は主要な関数のみを示しており、全ての関数を含んでいるわけではありません。
実際の実装では、エラー処理なども行われますが、図では簡略化のため省略しています。

コード解説

環境設定と初期化 (`main.py`)

import os
from dotenv import load_dotenv
import json
from tavily import TavilyClient
import base64
from PIL import Image
import io
import re
from anthropic import Anthropic, APIStatusError, APIError
import difflib
import time
from rich.console import Console
from rich.panel import Panel
from rich.syntax import Syntax
from rich.markdown import Markdown
import asyncio
import aiohttp
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn
import datetime
import venv
import subprocess
import sys
import signal
import logging
from typing import Tuple, Optional

# 仮想環境の設定
def setup_virtual_environment() -> Tuple[str, str]:
    venv_name = "code_execution_env"
    venv_path = os.path.join(os.getcwd(), venv_name)
    try:
        if not os.path.exists(venv_path):
            venv.create(venv_path, with_pip=True)
        
        # activateスクリプトのパスを取得
        if sys.platform == "win32":
            activate_script = os.path.join(venv_path, "Scripts", "activate.bat")
        else:
            activate_script = os.path.join(venv_path, "bin", "activate")
        
        return venv_path, activate_script
    except Exception as e:
        logging.error(f"Error setting up virtual environment: {str(e)}")
        raise

# .envファイルから環境変数をロード
load_dotenv()

# Anthropic APIキーの設定
anthropic_api_key = os.getenv("ANTHROPIC_API_KEY")
if not anthropic_api_key:
    raise ValueError("ANTHROPIC_API_KEY not found in environment variables")
client = Anthropic(api_key=anthropic_api_key)

# Tavily APIキーの設定
tavily_api_key = os.getenv("TAVILY_API_KEY")
if not tavily_api_key:
    raise ValueError("TAVILY_API_KEY not found in environment variables")
tavily = TavilyClient(api_key=tavily_api_key)

# コンソール出力用のRichライブラリ設定
console = Console()

# トークン追跡用の変数
main_model_tokens = {"input": 0, "output": 0}
tool_checker_tokens = {"input": 0, "output": 0}
code_editor_tokens = {"input": 0, "output": 0}
code_execution_tokens = {"input": 0, "output": 0}

# 会話履歴
conversation_history = []

# ファイルの内容
file_contents = {}

# コードエディタのメモリ
code_editor_memory = []

# 自動モードフラグ
automode = False

# 実行中プロセス
running_processes = {}

# 定数
CONTINUATION_EXIT_PHRASE = "AUTOMODE_COMPLETE"
MAX_CONTINUATION_ITERATIONS = 25
MAX_CONTEXT_TOKENS = 200000  # Reduced to 200k tokens for context window

# 使用するAIモデル
MAINMODEL = "claude-3-5-sonnet-20240620"  # Maintains conversation history and file contents
TOOLCHECKERMODEL = "claude-3-5-sonnet-20240620"
CODEEDITORMODEL = "claude-3-5-sonnet-20240620"
CODEEXECUTIONMODEL = "claude-3-haiku-20240307"

# システムプロンプト
BASE_SYSTEM_PROMPT = """
# ...(省略)...
"""

AUTOMODE_SYSTEM_PROMPT = """
# ...(省略)...
"""

.envファイルから環境変数(APIキーなど)を読み込みます。
Anthropic APIクライアントとTavily APIクライアントを初期化します。
コンソール出力を美しく見せるRichライブラリを設定します。
グローバル変数を定義し、初期化します。
定数、使用するAIモデル、システムプロンプトを定義します。

システムプロンプトの動的更新 (`update_system_prompt`)

def update_system_prompt(current_iteration: Optional[int] = None, max_iterations: Optional[int] = None) -> str:
    global file_contents
    chain_of_thought_prompt = """
    # ...(省略)...
    """
    
    file_contents_prompt = "\n\nFile Contents:\n"
    for path, content in file_contents.items():
        file_contents_prompt += f"\n--- {path} ---\n{content}\n"
    
    # モードに応じてプロンプトを返す
    if automode:
        # 自動モードの場合
        iteration_info = ""
        if current_iteration is not None and max_iterations is not None:
            iteration_info = f"You are currently on iteration {current_iteration} out of {max_iterations} in automode."
        return BASE_SYSTEM_PROMPT + file_contents_prompt + "\n\n" + AUTOMODE_SYSTEM_PROMPT.format(iteration_info=iteration_info) + "\n\n" + chain_of_thought_prompt
    else:
        # 通常モードの場合
        return BASE_SYSTEM_PROMPT + file_contents_prompt + "\n\n" + chain_of_thought_prompt

Claudeに送信するプロンプトを、モード(自動モード/通常モード)と現在のファイル内容に応じて動的に生成します。
BASE_SYSTEM_PROMPTは基本的な指示、AUTOMODE_SYSTEM_PROMPTは自動モード特有の指示を定義した変数です。
file_contents_promptは、現在読み込まれているファイルの内容を文字列として埋め込みます。

ファイル操作 (`create_folder`, `create_file`, `read_file`, `list_files`)

def create_folder(path):
    try:
        os.makedirs(path, exist_ok=True)
        return f"Folder created: {path}"
    except Exception as e:
        return f"Error creating folder: {str(e)}"

def create_file(path, content=""):
    global file_contents
    try:
        with open(path, 'w') as f:
            f.write(content)
        file_contents[path] = content
        return f"File created and added to system prompt: {path}"
    except Exception as e:
        return f"Error creating file: {str(e)}"

def read_file(path):
    global file_contents
    try:
        with open(path, 'r') as f:
            content = f.read()
        file_contents[path] = content
        return f"File '{path}' has been read and stored in the system prompt."
    except Exception as e:
        return f"Error reading file: {str(e)}"

def list_files(path="."):
    try:
        files = os.listdir(path)
        return "\n".join(files)
    except Exception as e:
        return f"Error listing files: {str(e)}"

これらの関数はOS標準ライブラリを利用し、ファイルシステム操作を実現しています。

各関数の機能:

create_folder: 指定されたパスにフォルダを作成します。既に存在する場合はエラーとなりません。
create_file: 指定されたパスにファイルを作成し、内容を書き込みます。file_contentsにも内容を保存します。
read_file: 指定されたパスのファイルを読み込み、内容をfile_contentsに保存します。
list_files: 指定されたパス内のファイルとディレクトリの一覧を取得します。

コードの差分表示 (`highlight_diff`, `generate_and_apply_diff`)

def highlight_diff(diff_text):
    return Syntax(diff_text, "diff", theme="monokai", line_numbers=True)

def generate_and_apply_diff(original_content, new_content, path):
    # diffを生成
    diff = list(difflib.unified_diff(
        original_content.splitlines(keepends=True),
        new_content.splitlines(keepends=True),
        fromfile=f"a/{path}",
        tofile=f"b/{path}",
        n=3
    ))

    # 変更がない場合はメッセージを返す
    if not diff:
        return "No changes detected."

    try:
        # ファイルに新しい内容を書き込む
        with open(path, 'w') as f:
            f.writelines(new_content)

        # diffをハイライト表示
        diff_text = ''.join(diff)
        highlighted_diff = highlight_diff(diff_text)

        diff_panel = Panel(
            highlighted_diff,
            title=f"Changes in {path}",
            expand=False,
            border_style="cyan"
        )

        console.print(diff_panel)

        # 追加・削除された行数をカウント
        added_lines = sum(1 for line in diff if line.startswith('+') and not line.startswith('+++'))
        removed_lines = sum(1 for line in diff if line.startswith('-') and not line.startswith('---'))

        # 変更の概要を作成
        summary = f"Changes applied to {path}:\n"
        summary += f"  Lines added: {added_lines}\n"
        summary += f"  Lines removed: {removed_lines}\n"

        return summary

    except Exception as e:
        # エラー処理
        error_panel = Panel(
            f"Error: {str(e)}",
            title="Error Applying Changes",
            style="bold red"
        )
        console.print(error_panel)
        return f"Error applying changes: {str(e)}"

これらの関数は、コード変更の前後を表示する際に、差分をわかりやすく表示する機能を提供します。

各関数の機能:

highlight_diff: difflibライブラリで生成された差分情報を、Richライブラリを用いてシンタックスハイライト表示します。
generate_and_apply_diff: ファイルの変更前後の内容を受け取り、差分を生成、表示します。highlight_diff関数を呼び出してシンタックスハイライト表示を行います。

コード編集 (`generate_edit_instructions`, `parse_search_replace_blocks`, `edit_and_apply`, `apply_edits`)

async def generate_edit_instructions(file_content, instructions, project_context):
    global code_editor_tokens, code_editor_memory
    try:
        # メモリの内容を整形
        memory_context = "\n".join([f"Memory {i+1}:\n{mem}" for i, mem in enumerate(code_editor_memory)])

        # CODEEDITORMODELに送信するプロンプト
        system_prompt = f"""
        # ...(省略)...
        """

        # CODEEDITORMODELへのAPI呼び出し
        response = client.messages.create(
            model=CODEEDITORMODEL,
            max_tokens=8000,
            system=system_prompt,
            extra_headers={"anthropic-beta": "max-tokens-3-5-sonnet-2024-07-15"},
            messages=[
                {"role": "user", "content": "Generate SEARCH/REPLACE blocks for the necessary changes."}
            ]
        )

        # トークン使用量を更新
        code_editor_tokens['input'] += response.usage.input_tokens
        code_editor_tokens['output'] += response.usage.output_tokens

        # レスポンスからSEARCH/REPLACEブロックを抽出
        edit_instructions = parse_search_replace_blocks(response.content[0].text)

        # コードエディタメモリを更新
        code_editor_memory.append(f"Edit Instructions:\n{response.content[0].text}")

        return edit_instructions

    except Exception as e:
        console.print(f"Error in generating edit instructions: {str(e)}", style="bold red")
        return []  # エラーが発生した場合は空のリストを返す

def parse_search_replace_blocks(response_text):
    blocks = []
    lines = response_text.split('\n')
    current_block = {}
    current_section = None

    # レスポンスを解析し、SEARCH/REPLACEブロックを抽出
    for line in lines:
        # ...(SEARCH/REPLACEブロックのパース処理)...

    return blocks


async def edit_and_apply(path, instructions, project_context, is_automode=False):
    global file_contents
    try:
        # ファイルの内容を取得
        original_content = file_contents.get(path, "")
        if not original_content:
            with open(path, 'r') as file:
                original_content = file.read()
            file_contents[path] = original_content

        # 編集指示を生成
        edit_instructions = await generate_edit_instructions(original_content, instructions, project_context)
        
        # 編集指示がある場合
        if edit_instructions:
            # 編集指示を表示
            console.print(Panel("The following SEARCH/REPLACE blocks have been generated:", title="Edit Instructions", style="cyan"))
            for i, block in enumerate(edit_instructions, 1):
                console.print(f"Block {i}:")
                console.print(Panel(f"SEARCH:\n{block['search']}\n\nREPLACE:\n{block['replace']}", expand=False))

            # 編集を適用
            edited_content, changes_made = await apply_edits(path, edit_instructions, original_content)

            # 変更があった場合
            if changes_made:
                # 差分を生成して表示
                diff_result = generate_and_apply_diff(original_content, edited_content, path)

                console.print(Panel("The following changes will be applied:", title="File Changes", style="cyan"))
                console.print(diff_result)

                # 自動モードでない場合は確認
                if not is_automode:
                    confirm = console.input("[bold yellow]Do you want to apply these changes? (yes/no): [/bold yellow]")
                    if confirm.lower() != 'yes':
                        return "Changes were not applied."

                # ファイルに書き込み
                with open(path, 'w') as file:
                    file.write(edited_content)
                file_contents[path] = edited_content  # file_contentsを更新
                console.print(Panel(f"File contents updated in system prompt: {path}", style="green"))
                return f"Changes applied to {path}:\n{diff_result}"
            else:
                return f"No changes needed for {path}"
        else:
            return f"No changes suggested for {path}"
    except Exception as e:
        return f"Error editing/applying to file: {str(e)}"



async def apply_edits(file_path, edit_instructions, original_content):
    changes_made = False
    edited_content = original_content
    total_edits = len(edit_instructions)

    # 進捗表示
    with Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        BarColumn(),
        TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
        console=console
    ) as progress:
        edit_task = progress.add_task("[cyan]Applying edits...", total=total_edits)

        for i, edit in enumerate(edit_instructions, 1):
            search_content = edit['search']
            replace_content = edit['replace']
            
            # 編集を適用
            if search_content in edited_content:
                edited_content = edited_content.replace(search_content, replace_content)
                changes_made = True
                
                # 差分を表示
                diff_result = generate_and_apply_diff(search_content, replace_content, file_path)
                console.print(Panel(diff_result, title=f"Changes in {file_path} ({i}/{total_edits})", style="cyan"))

            # 進捗を更新
            progress.update(edit_task, advance=1)

    return edited_content, changes_made

これらの関数は、Claudeの力を借りてコードを編集する機能を提供します。CODEEDITORMODELという specialized AI model を利用します。

各関数の機能:

generate_edit_instructions: CODEEDITORMODEL を使用して、ファイルの編集指示(SEARCH/REPLACEブロック)を生成します。
parse_search_replace_blocks: generate_edit_instructions から返された指示(SEARCH/REPLACEブロック)を解析します。
edit_and_apply: generate_edit_instructionsで生成された指示に基づいてファイルの編集を行い、apply_edits関数を呼び出して変更を適用します。
apply_edits: SEARCH/REPLACE ブロックに基づいてファイルの内容を更新します。Richライブラリを用いて、変更内容をわかりやすく表示します。

コード実行 (`execute_code`, `send_to_ai_for_executing`)

async def execute_code(code, timeout=10):
    global running_processes
    # 仮想環境の設定
    venv_path, activate_script = setup_virtual_environment()
    
    # プロセスIDを生成
    process_id = f"process_{len(running_processes)}"
    
    # コードを一時ファイルに書き込み
    with open(f"{process_id}.py", "w") as f:
        f.write(code)
    
    # 実行コマンドの準備
    if sys.platform == "win32":
        command = f'"{activate_script}" && python3 {process_id}.py'
    else:
        command = f'source "{activate_script}" && python3 {process_id}.py'
    
    # サブプロセスを作成してコマンドを実行
    process = await asyncio.create_subprocess_shell(
        command,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        shell=True,
        preexec_fn=None if sys.platform == "win32" else os.setsid
    )
    
    # プロセスをrunning_processesに保存
    running_processes[process_id] = process
    
    try:
        # プロセスの出力を読み込む
        stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
        stdout = stdout.decode()
        stderr = stderr.decode()
        return_code = process.returncode
    except asyncio.TimeoutError:
        # タイムアウトした場合
        stdout = "Process started and running in the background."
        stderr = ""
        return_code = "Running"
    
    execution_result = f"Process ID: {process_id}\n\nStdout:\n{stdout}\n\nStderr:\n{stderr}\n\nReturn Code: {return_code}"
    return process_id, execution_result

async def send_to_ai_for_executing(code, execution_result):
    global code_execution_tokens

    try:
        # CODEEXECUTIONMODELに送信するプロンプト
        system_prompt = f"""
        # ...(省略)...
        """

        # CODEEXECUTIONMODELへのAPI呼び出し
        response = client.messages.create(
            model=CODEEXECUTIONMODEL,
            max_tokens=2000,
            system=system_prompt,
            messages=[
                {
                    "role": "user",
                    "content": f"Analyze this code execution from the 'code_execution_env' virtual environment:\n\nCode:\n{code}\n\nExecution Result:\n{execution_result}",
                }
            ],
        )

        # トークン使用量を更新
        code_execution_tokens["input"] += response.usage.input_tokens
        code_execution_tokens["output"] += response.usage.output_tokens

        analysis = response.content[0].text

        return analysis

    except Exception as e:
        console.print(
            f"Error in AI code execution analysis: {str(e)}", style="bold red"
        )
        return f"Error analyzing code execution from 'code_execution_env': {str(e)}"

これらの関数は、生成されたPythonコードを仮想環境内で実行し、結果を分析します。

各関数の機能:

execute_code: 指定されたPythonコードを仮想環境内で実行します。asyncioライブラリを利用し、非同期処理を実現しています。
send_to_ai_for_executing: CODEEXECUTIONMODEL を使用して、execute_code の実行結果を分析します。

ウェブ検索 (`tavily_search`)

def tavily_search(query):
    try:
        response = tavily.qna_search(query=query, search_depth="advanced")
        return response
    except Exception as e:
        return f"Error performing search: {str(e)}"

Tavily APIを使用してウェブ検索を実行します。

プロセス管理 (`stop_process`)

def stop_process(process_id):
    global running_processes
    if process_id in running_processes:
        process = running_processes[process_id]
        if sys.platform == "win32":
            process.terminate()
        else:
            os.killpg(os.getpgid(process.pid), signal.SIGTERM)
        del running_processes[process_id]
        return f"Process {process_id} has been stopped."
    else:
        return f"No running process found with ID {process_id}."

execute_code 関数で実行されたプロセスを停止します。

チャットログの保存 (`save_chat`)

def save_chat():
    # ファイル名を生成
    now = datetime.datetime.now()
    filename = f"Chat_{now.strftime('%H%M')}.md"
    
    # 会話履歴をフォーマット
    formatted_chat = "# Claude-3-Sonnet Engineer Chat Log\n\n"
    for message in conversation_history:
        if message['role'] == 'user':
            formatted_chat += f"## User\n\n{message['content']}\n\n"
        elif message['role'] == 'assistant':
            if isinstance(message['content'], str):
                formatted_chat += f"## Claude\n\n{message['content']}\n\n"
            elif isinstance(message['content'], list):
                for content in message['content']:
                    if content['type'] == 'tool_use':
                        formatted_chat += f"### Tool Use: {content['name']}\n\n```json\n{json.dumps(content['input'], indent=2)}\n```\n\n"
                    elif content['type'] == 'text':
                        formatted_chat += f"## Claude\n\n{content['text']}\n\n"
        elif message['role'] == 'user' and isinstance(message['content'], list):
            for content in message['content']:
                if content['type'] == 'tool_result':
                    formatted_chat += f"### Tool Result\n\n```\n{content['content']}\n```\n\n"
    
    # ファイルに保存
    with open(filename, 'w', encoding='utf-8') as f:
        f.write(formatted_chat)
    
    return filename

現在のチャットログをMarkdownファイルに保存します。

メインループ (`main`)

async def main():
    global automode, conversation_history
    # 起動メッセージを表示
    console.print(
        Panel(
            "Welcome to the Claude-3-Sonnet Engineer Chat with Multi-Agent and Image Support!",
            title="Welcome",
            style="bold green",
        )
    )
    # 使い方を表示
    console.print("Type 'exit' to end the conversation.")
    console.print("Type 'image' to include an image in your message.")
    console.print(
        "Type 'automode [number]' to enter Autonomous mode with a specific number of iterations."
    )
    console.print("Type 'reset' to clear the conversation history.")
    console.print("Type 'save chat' to save the conversation to a Markdown file.")
    console.print(
        "While in automode, press Ctrl+C at any time to exit the automode to return to regular chat."
    )

    # メインループ
    while True:
        # ユーザー入力を取得
        user_input = console.input("[bold cyan]You:[/bold cyan] ")

        # 終了コマンド
        if user_input.lower() == "exit":
            # 終了メッセージを表示
            console.print(
                Panel(
                    "Thank you for chatting. Goodbye!",
                    title_align="left",
                    title="Goodbye",
                    style="bold green",
                )
            )
            break

        # リセットコマンド
        if user_input.lower() == "reset":
            # 会話履歴をリセット
            reset_conversation()
            continue

        # チャットログ保存コマンド
        if user_input.lower() == "save chat":
            # チャットログを保存
            filename = save_chat()
            # 保存完了メッセージを表示
            console.print(
                Panel(
                    f"Chat saved to {filename}", title="Chat Saved", style="bold green"
                )
            )
            continue

        # 画像入力コマンド
        if user_input.lower() == "image":
            # 画像パスを取得
            image_path = (
                console.input("[bold cyan]Drag and drop your image here, then press enter:[/bold cyan] ")
                .strip()
                .replace("'", "")
            )

            # 画像パスが有効な場合
            if os.path.isfile(image_path):
                # 画像に対するプロンプトを取得
                user_input = console.input(
                    "[bold cyan]You (prompt for image):[/bold cyan] "
                )
                # Claudeと対話
                response, _ = await chat_with_claude(user_input, image_path)
            # 画像パスが無効な場合
            else:
                # エラーメッセージを表示
                console.print(
                    Panel(
                        "Invalid image path. Please try again.",
                        title="Error",
                        style="bold red",
                    )
                )
                continue

        # 自動モードコマンド
        elif user_input.lower().startswith("automode"):
            try:
                # 反復回数を取得
                parts = user_input.split()
                if len(parts) > 1 and parts[1].isdigit():
                    max_iterations = int(parts[1])
                else:
                    max_iterations = MAX_CONTINUATION_ITERATIONS

                # 自動モードを有効化
                automode = True
                # 自動モード開始メッセージを表示
                console.print(
                    Panel(
                        f"Entering automode with {max_iterations} iterations. Please provide the goal of the automode.",
                        title_align="left",
                        title="Automode",
                        style="bold yellow",
                    )
                )
                console.print(
                    Panel(
                        "Press Ctrl+C at any time to exit the automode loop.",
                        style="bold yellow",
                    )
                )
                # ユーザー入力(目標)を取得
                user_input = console.input("[bold cyan]You:[/bold cyan] ")

                iteration_count = 0
                try:
                    # 自動モードループ
                    while automode and iteration_count < max_iterations:
                        # Claudeと対話
                        response, exit_continuation = await chat_with_claude(
                            user_input,
                            current_iteration=iteration_count + 1,
                            max_iterations=max_iterations,
                        )

                        # 自動モード終了条件
                        if exit_continuation or CONTINUATION_EXIT_PHRASE in response:
                            # 自動モード完了メッセージを表示
                            console.print(
                                Panel(
                                    "Automode completed.",
                                    title_align="left",
                                    title="Automode",
                                    style="green",
                                )
                            )
                            # 自動モードを無効化
                            automode = False
                        # 自動モード継続
                        else:
                            # 反復完了メッセージを表示
                            console.print(
                                Panel(
                                    f"Continuation iteration {iteration_count + 1} completed. Press Ctrl+C to exit automode. ",
                                    title_align="left",
                                    title="Automode",
                                    style="yellow",
                                )
                            )
                            # 次の反復のためのユーザー入力を設定
                            user_input = "Continue with the next step. Or STOP by saying 'AUTOMODE_COMPLETE' if you think you've achieved the results established in the original request."
                        # 反復回数を増加
                        iteration_count += 1

                        # 最大反復回数に達した場合
                        if iteration_count >= max_iterations:
                            # 最大反復回数到達メッセージを表示
                            console.print(
                                Panel(
                                    "Max iterations reached. Exiting automode.",
                                    title_align="left",
                                    title="Automode",
                                    style="bold red",
                                )
                            )
                            # 自動モードを無効化
                            automode = False
                # Ctrl+Cによる中断
                except KeyboardInterrupt:
                    # 自動モード中断メッセージを表示
                    console.print(
                        Panel(
                            "\nAutomode interrupted by user. Exiting automode.",
                            title_align="left",
                            title="Automode",
                            style="bold red",
                        )
                    )
                    # 自動モードを無効化
                    automode = False
                    # ユーザー入力があればアシスタントの応答を追加
                    if (
                        conversation_history
                        and conversation_history[-1]["role"] == "user"
                    ):
                        conversation_history.append(
                            {
                                "role": "assistant",
                                "content": "Automode interrupted. How can I assist you further?",
                            }
                        )
            # Ctrl+Cによる中断
            except KeyboardInterrupt:
                # 自動モード中断メッセージを表示
                console.print(
                    Panel(
                        "\nAutomode interrupted by user. Exiting automode.",
                        title_align="left",
                        title="Automode",
                        style="bold red",
                    )
                )
                # 自動モードを無効化
                automode = False
                # ユーザー入力があればアシスタントの応答を追加
                if conversation_history and conversation_history[-1]["role"] == "user":
                    conversation_history.append(
                        {
                            "role": "assistant",
                            "content": "Automode interrupted. How can I assist you further?",
                        }
                    )

            # 自動モード終了メッセージを表示
            console.print(Panel("Exited automode. Returning to regular chat.", style="green"))
        # 通常の対話処理
        else:
            response, _ = await chat_with_claude(user_input)

ユーザーからの入力を処理するメインループです。
入力内容に応じて、chat_with_claude関数を呼び出してClaudeと対話したり、imageコマンドで画像処理を行ったり、automodeコマンドで自動モードに移行したりします。

まとめ

Claude-Engineer は、大規模言語モデル Claude の力を借りて、ファイル操作、コード生成、ウェブ検索、コード実行などをチャット形式で行える、強力な開発支援ツールです。

本記事では、そのソースコードを詳細に解説し、Mermaid図解を用いて処理の流れを可視化しました。コードの動作を理解することで、Claude-Engineerをより効果的に活用し、開発効率の向上に繋げることができるでしょう。

はじめに

全体像

Claude-Engineerの構造

各モジュールの役割

モジュール間の関係性

Claude-Engineer 関数呼び出しフロー

図の説明

補足

コード解説

環境設定と初期化 (main.py)

システムプロンプトの動的更新 (update_system_prompt)

ファイル操作 (create_folder, create_file, read_file, list_files)

コードの差分表示 (highlight_diff, generate_and_apply_diff)

コード編集 (generate_edit_instructions, parse_search_replace_blocks, edit_and_apply, apply_edits)

コード実行 (execute_code, send_to_ai_for_executing)

ウェブ検索 (tavily_search)

プロセス管理 (stop_process)

チャットログの保存 (save_chat)

メインループ (main)

まとめ

Discussion

環境設定と初期化 (`main.py`)

システムプロンプトの動的更新 (`update_system_prompt`)

ファイル操作 (`create_folder`, `create_file`, `read_file`, `list_files`)

コードの差分表示 (`highlight_diff`, `generate_and_apply_diff`)

コード編集 (`generate_edit_instructions`, `parse_search_replace_blocks`, `edit_and_apply`, `apply_edits`)

コード実行 (`execute_code`, `send_to_ai_for_executing`)

ウェブ検索 (`tavily_search`)

プロセス管理 (`stop_process`)

チャットログの保存 (`save_chat`)

メインループ (`main`)