CrewAI で GitHub リポジトリ要約ツールを構築する

import os
import subprocess
from typing import Type
from pydantic import BaseModel, Field
from crewai.tools import BaseTool
from langchain_openai import ChatOpenAI

# Clone the repo if not already cloned
REPO_URL = "https://github.com/karpathy/nanogpt.git"
REPO_DIR = "./nanogpt"

if not os.path.exists(REPO_DIR):
    subprocess.run(["git", "clone", REPO_URL])

# --- Define the Repo Explorer Tool ---
class RepoQueryInput(BaseModel):
    query: str = Field(..., description="Describe what you are looking for in the codebase.")

class RepoExplorerTool(BaseTool):
    name: str = "Repo Explorer"
    description: str = "Given a query, find relevant file paths from the nanogpt repo."
    args_schema: Type[BaseModel] = RepoQueryInput

    def _run(self, query: str) -> str:
        file_list = []
        for root, _, files in os.walk(REPO_DIR):
            for file in files:
                full_path = os.path.join(root, file)
                file_list.append(full_path)

        llm = ChatOpenAI(model_name="gpt-4o-mini", temperature=0)
        prompt = f"""
You are a codebase expert. Here is a list of files:\n{file_list}\n\n
Based on the user query "{query}", return ONLY the full file paths that are most relevant.
Respond with one file path per line, no extra text.
"""
        response = llm.invoke(prompt)
        return response.content.strip()

# --- Define the View File Content Tool ---
class ViewFileInput(BaseModel):
    filepath: str = Field(..., description="Full path to the file you want to read.")

class ViewFileTool(BaseTool):
    name: str = "View File Content"
    description: str = "Reads the full content of a given file path from the nanogpt repo."
    args_schema: Type[BaseModel] = ViewFileInput

    def _run(self, filepath: str) -> str:
        filepath = filepath.strip()  # <<< add this
        try:
            with open(filepath, "r", encoding="utf-8") as f:
                content = f.read()
            return content
        except Exception as e:
            return f"Error reading file: {str(e)}"

# --- Instantiate the tools ---
repo_explorer_tool = RepoExplorerTool()
view_file_tool = ViewFileTool()

# --- Example usage ---
if __name__ == "__main__":
    # Step 1: Find files related to training loop
    query = "Where is the training loop defined?"
    result_paths = repo_explorer_tool._run(query=query)
    print("Relevant Files:\n", result_paths)

    # Step 2: Read the first file found
    first_file = result_paths.splitlines()[0]
    file_content = view_file_tool._run(filepath=first_file)
    print("\nContent of the first relevant file:\n")
    print(file_content[:3000])  # only print first 3000 chars to keep output readable

import os
import subprocess
from typing import Type, List
from pydantic import BaseModel, Field
from crewai import Agent, Task, Crew, Process
from crewai.tools import BaseTool
from langchain_openai import ChatOpenAI
import weave; weave.init("crewai_git_documenter")
# --- Repo Setup ---

REPO_URL = os.getenv('REPO_URL', 'https://github.com/LiveCodeBench/LiveCodeBench')
REPO_NAME = REPO_URL.split("/")[-1].replace(".git", "")
REPO_DIR = f"./{REPO_NAME}"

if not os.path.exists(REPO_DIR):
    subprocess.run(["git", "clone", REPO_URL])

# --- Tools ---

class RepoQueryInput(BaseModel):
    query: str = Field(..., query="Describe what you are looking for in the codebase.")

class RepoExplorerTool(BaseTool):
    name: str = "Repo Explorer"
    description: str = "Given a query, find relevant file paths from the repo."
    args_schema: Type[BaseModel] = RepoQueryInput

    def _run(self, query: str) -> str:
        file_list = []
        for root, _, files in os.walk(REPO_DIR):
            for file in files:
                full_path = os.path.join(root, file)
                file_list.append(full_path)

        llm = ChatOpenAI(model_name="gpt-4o-mini", temperature=0)
        prompt = f"""
Here is a list of files:\n{file_list}\n\n
User query: "{query}"
Return ONLY full file paths that match. One per line.
"""
        response = llm.invoke(prompt)
        return response.content.strip()

class ViewFileInput(BaseModel):
    filepath: str = Field(..., description="Full path to the file you want to read.")

class ViewFileTool(BaseTool):
    name: str = "View File Content"
    description: str = "Reads the full content of a given file path from the repo."
    args_schema: Type[BaseModel] = ViewFileInput

    def _run(self, filepath: str) -> str:
        filepath = filepath.strip()
        try:
            with open(filepath, "r", encoding="utf-8") as f:
                return f.read()
        except Exception as e:
            return f"Error reading file: {str(e)}"

class SummarizeFileTool(BaseTool):
    name: str = "Summarize File Content"
    description: str = "Summarizes a given file's content into a short description."
    args_schema: Type[BaseModel] = ViewFileInput

    def _run(self, filepath: str) -> str:
        filepath = filepath.strip()
        try:
            with open(filepath, "r", encoding="utf-8") as f:
                content = f.read()
            llm = ChatOpenAI(model_name="gpt-4o-mini", temperature=0)
            prompt = f"Summarize the following code file in 5 sentences:\n\n{content}"
            response = llm.invoke(prompt)
            return response.content.strip()
        except Exception as e:
            return f"Error summarizing file: {str(e)}"

# --- Instantiate tools ---
repo_explorer_tool = RepoExplorerTool()
view_file_tool = ViewFileTool()
summarize_file_tool = SummarizeFileTool()

# --- Agents ---

demo_generator_agent = Agent(
    role="Demo Generator",
    goal="Write and expand a basic usage example for the repo.",
    backstory="Expert technical writer.",
    tools=[repo_explorer_tool, view_file_tool],
    allow_delegation=False,
    verbose=True,
    llm=ChatOpenAI(model_name="gpt-4o-mini", temperature=0.5)
)

followup_analyst_agent = Agent(
    role="Follow-Up Analyst",
    goal="Analyze the usage guide, find missing info, suggest improvements.",
    backstory="Developer experience expert.",
    allow_delegation=False,
    verbose=True,
    llm=ChatOpenAI(model_name="gpt-4o-mini", temperature=0.3)
)

html_coder_agent = Agent(
    role="HTML Coder",
    goal="Turn the improved usage guide into a clean HTML page.",
    backstory="Frontend dev specializing in docs websites. DO NOT GENERATE IMAGES MAN eg  dont do src=data:image/png etc etc",
    allow_delegation=False,
    verbose=True,
    llm=ChatOpenAI(model_name="gpt-4o-2024-08-06", temperature=0.4)
)

repo_summarizer_agent = Agent(
    role="Repo Summarizer",
    goal="Summarize all important files in the repo for easier understanding.",
    backstory="Documentation architect.",
    tools=[repo_explorer_tool, view_file_tool, summarize_file_tool],
    allow_delegation=False,
    verbose=True,
    llm=ChatOpenAI(model_name="gpt-4o-mini", temperature=0.4)
)

diagram_generator_agent = Agent(
    role="Diagram Generator",
    goal="Using file summaries, generate a full architecture diagram of the repo.",
    backstory="System architect skilled in explaining large codebases.",
    allow_delegation=False,
    verbose=True,
    llm=ChatOpenAI(model_name="gpt-4o-mini", temperature=0.4)
)


shell_script_agent = Agent(
    role="Shell Script Automation Engineer",
    goal="Create a working shell script that sets up all requirements and runs the repo's demo or basic usage.",
    backstory="Expert in bash scripting, devops, and repo onboarding automations.",
    tools=[repo_explorer_tool, view_file_tool, summarize_file_tool],  # Optional, useful for reading files
    allow_delegation=False,
    verbose=True,
    llm=ChatOpenAI(model_name="gpt-4o-mini", temperature=0.3)
)

# --- Tasks ---


demo_generation_task = Task(
    description=f"Write a beginner-friendly usage guide explaining how to use {REPO_NAME}. Include a minimal example command.",
    expected_output="Markdown snippet with instructions.",
    agent=demo_generator_agent,
    output_file="basic_usage.md"
)

followup_task = Task(
    description="Read 'basic_usage.md', find confusing parts, generate 5+ questions and improvements.",
    expected_output="List of questions + feedback report.",
    agent=followup_analyst_agent,
    context=[demo_generation_task],
    output_file="feedback_report.md"
)

improvement_task = Task(
    description="Use 'feedback_report.md' to expand and improve the usage guide.",
    expected_output="New improved guide.",
    agent=demo_generator_agent,
    context=[followup_task],
    output_file="improved_usage.md"
)

repo_summarization_task = Task(
    description="Scan the repo, summarize the major files (training, models, data utils, etc.).",
    expected_output="List of file summaries.",
    agent=repo_summarizer_agent,
    output_file="repo_summary.md"
)

diagram_generation_task = Task(
    description="Using 'repo_summary.md', create an overall architecture diagram of the repo in text/plantuml format.",
    expected_output="Architecture diagram text.",
    agent=diagram_generator_agent,
    context=[repo_summarization_task],
    tools=[repo_explorer_tool, view_file_tool, summarize_file_tool],

    output_file="architecture_diagram.md"
)

shell_script_task = Task(
    description=(
        "Read the improved usage guide and repo summary. Write a BASH shell script that fully sets up the repo from scratch, "
        "including dependency installation, environment setup, any required downloads, and running the minimal example demo. "
        "The script should be idempotent (safe to rerun), and **explain each step with comments**. Output the script in a codeblock."
    ),
    expected_output="A complete shell script (setup_and_run_demo.sh) with comments.",
    agent=shell_script_agent,
    context=[improvement_task, repo_summarization_task],  # Uses improved usage and file summaries
    output_file="setup_and_run_demo.sh"
)



html_generation_task = Task(
    description="Combine 'improved_usage.md', and 'architecture_diagram.md' into a final HTML file. Make sure to append the full setup_and_run_demo.sh script at the end",
    expected_output="Final styled HTML guide. Follow styling similar to Wandb's website styling",
    agent=html_coder_agent,
    context=[improvement_task, diagram_generation_task, shell_script_task],
    output_file="final_guide.html"
)


# --- Run Crew ---
crew = Crew(
    agents=[
        demo_generator_agent,
        followup_analyst_agent,
        repo_summarizer_agent,
        diagram_generator_agent,
        html_coder_agent,
        shell_script_agent,  # <-- Add here
    ],
    tasks=[
        demo_generation_task,
        followup_task,
        improvement_task,
        repo_summarization_task,
        diagram_generation_task,
        shell_script_task,    # <-- Add here, before html_generation_task
        html_generation_task,
    ],
    process=Process.sequential,
    verbose=True
)

if __name__ == "__main__":
    result = crew.kickoff()
    print("\n\nFinal HTML Guide and Repo Diagram Created:\n")
    print(result)