Building a Simple GitHub PR Review Agent with ReAct

# ReAct loop
while current_step < self.max_steps:
    # Step 1: REASONING - Analyze the current state
    reasoning = await self._react_reasoning(agent, context)
    
    # Step 2: ACTION SELECTION - Select the next action
    action = await self._react_action_selection(agent, context, reasoning)
    
    # If the action is to finish, we're done
    if action.action_type == "finish":
        return action.output or {}
        
    # Step 3: OBSERVATION - Execute the selected tool
    if action.action_type == "use_tool" and action.tool_name:
        tool_name = action.tool_name
        tool_params = action.tool_params or {}
        
        # Execute the tool
        result = await self.tools[tool_name](tool_params)
        observation = str(result)
        
        # Update context with observation
        context["last_observation"] = observation
        context["intermediate_results"][f"step_{current_step}"] = result

class AgentReasoning(BaseModel):
    """Model for agent's reasoning step output."""
    content: str = Field(..., description="The agent's reasoning about the current state")

class AgentAction(BaseModel):
    """Model for agent's action selection output."""
    action_type: str = Field(..., description="Type of action: 'use_tool' or 'finish'")
    tool_name: Optional[str] = Field(None, description="Name of the tool to use")
    tool_params: Optional[Dict[str, Any]] = Field(None, description="Parameters for the tool")
    output: Optional[Dict[str, Any]] = Field(None, description="Final output if finishing")

class AgentOutput(BaseModel):
    """Model for the final PR review output."""
    review_summary: str = Field(..., description="Summary of the PR changes")
    issues_found: List[str] = Field(default_factory=list, description="List of issues found")
    suggestions: List[str] = Field(default_factory=list, description="List of suggestions")
    overall_assessment: str = Field(..., description="Overall assessment of the PR")

@trace(name="github_pr_tool.execute")
async def execute(self, params: Dict[str, Any]) -> Dict[str, Any]:
    """Execute the GitHub PR tool."""
    try:
        # Get PR information
        pr_info = await self._get_pr_info(params["owner"], params["repo"], params["pr_number"])
        
        # Check if repository is private and we're not authenticated
        if pr_info.get("private", False) and "Authorization" not in self.headers:
            return {
                "error": "This is a private repository. A GitHub token is required for access.",
                "status": "error"
            }
        
        # Get PR files and diff
        files = await self._get_pr_files(params["owner"], params["repo"], params["pr_number"])
        diff = await self._get_pr_diff(params["owner"], params["repo"], params["pr_number"])
        
        # Return structured result
        return {
            "pr_title": pr_info["title"],
            "pr_author": pr_info["user"]["login"],
            "changed_files": [f["filename"] for f in files],
            "additions": pr_info["additions"],
            "deletions": pr_info["deletions"],
            "diff": self._truncate_diff(diff),
            "pr_description": pr_info["body"] or "",
            "pr_url": pr_info["html_url"],
            "repository_private": pr_info.get("private", False),
            "status": "success"
        }
    except Exception as e:
        logger.error(f"Error executing GitHub PR tool: {e}", exc_info=True)
        return {"error": f"Error retrieving PR information: {str(e)}", "status": "error"}

async def _react_reasoning(self, agent: Dict[str, Any], context: Dict[str, Any]) -> AgentReasoning:
    """Generate reasoning based on the current context."""
    reasoning_instructions = """
    You are in the REASONING phase of a ReAct (Reasoning-Acting-Observation) loop.
    
    In this phase, you should:
    1. Analyze the current state and context
    2. Think step-by-step about what you know and what you need to find out
    3. Consider what tools or actions might be helpful
    4. Determine your next steps
    
    Your reasoning should be thorough, logical, and clear.
    
    Additionally, provide a confidence score from 0.0 to 1.0 indicating how
    confident you are in your reasoning.
    """
    
    result, _ = await opper.call(
        name="agent_reasoning",
        instructions=reasoning_instructions,
        input={
            "agent_instructions": agent.get("instructions", ""),
            "context": context,
            "step_number": context.get("current_step", 0),
            "last_observation": context.get("last_observation", None),
        },
        output_type=AgentReasoning,
    )
    return result

async def _react_action_selection(
    self, agent: Dict[str, Any], context: Dict[str, Any], reasoning: AgentReasoning
) -> AgentAction:
    """Select the next action based on reasoning."""
    # Get the list of available tools
    available_tools = list(self.tools.keys())
    
    action_instructions = """
    You are in the ACTION SELECTION phase of a ReAct (Reasoning-Acting-Observation) loop.
    
    Based on your prior reasoning, you must now decide on the next action to take.
    
    You have two options:
    1. Use a tool to gather more information or make progress:
       - action_type: "use_tool"
       - tool_name: Select from the available tools in the input
       - tool_params: Provide the necessary parameters for the tool
       
    2. Finish the task if you have enough information:
       - action_type: "finish"
       - output: Provide your final review with:
         - review_summary: A concise summary of the PR changes
         - issues_found: A list of issues or concerns
         - suggestions: A list of improvement suggestions
         - overall_assessment: Your final assessment of the PR
    """
    
    result, _ = await opper.call(
        name="agent_action",
        instructions=action_instructions,
        input={
            "reasoning": reasoning.content,
            "reasoning_confidence": reasoning.confidence,
            "context": context,
            "available_tools": available_tools,
            "agent_instructions": agent.get("instructions", ""),
            "step_number": context.get("current_step", 0),
        },
        output_type=AgentAction,
    )
    return result

@trace(name="agent_runner.run_agent")
async def run_agent(self, agent_id: str, agent: Dict[str, Any], input_data: Dict[str, Any]) -> Dict[str, Any]:
    """Run an agent with the given input data."""
    # ... implementation ...

# Agent configuration
PR_REVIEW_AGENT = {
    "instructions": """
    You are a GitHub PR reviewer. Your task is to review pull requests and provide helpful feedback.
    You should:
    1. Fetch the PR information using the github_pr_tool
    2. Analyze the changes and their impact
    3. Identify potential issues or improvements
    4. Provide a detailed review with actionable feedback
    
    Your final output should include:
    - A summary of the changes
    - List of issues found (if any)
    - Suggestions for improvement
    - Overall assessment
    """,
    "verbose": False,  # Will be set from command line args
}

# Initialize services
agent_runner = AgentRunnerService()

# Initialize GitHub PR tool
github_token = os.getenv("GITHUB_TOKEN")  # Optional for public repositories
github_pr_tool = GitHubPRTool(github_token)

# Register tools
agent_runner.register_tools({
    "github_pr_tool": github_pr_tool.execute
})

# Run the agent
result = await agent_runner.run_agent(
    agent_id="github_pr_reviewer",
    agent=PR_REVIEW_AGENT,
    input_data={
        "owner": args.owner,
        "repo": args.repo,
        "pr_number": args.pr_number,
    }
)