智能代理驱动的DevOps：AI赋能的软件交付新时代

1
2

pip install "agenspy[mcp]" dspy openai
npm install -g @modelcontextprotocol/server-github

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

import os
from typing import Dict, List, Optional
import dspy
from dspy import Prediction
from agenspy import RealMCPClient

class PRReviewAgent(dspy.Module):
    """Agent for automated PR reviews."""
    
    def __init__(self, mcp_client, github_token: Optional[str] = None):
        super().__init__()
        self.mcp_client = mcp_client
        self.github_token = github_token

        # Define predictors with modern signature syntax using ->
        self.analyze_changes = dspy.Predict(

        )
        
        self.review_docs = dspy.Predict(
            "changes -> doc_updates: list[str], doc_suggestions: list[str]"
        )
        
        self.analyze_impact = dspy.Predict(
            "changes, codebase -> impact_analysis: str, risk_score: float"
        )

    def forward(self, pr_url: str) -> Prediction:
        """Review a pull request and provide comprehensive analysis."""
        print(f" Reviewing PR: {pr_url}")
        
        # Define default values

        default_edge_cases = ["No potential edge cases detected"]
        default_doc_updates = ["No documentation updates required"]
        default_doc_suggestions = ["Documentation appears to be complete"]
        default_impact = "No significant impact detected"
        default_risk = 0.0
        
        # Fetch PR data
        pr_data = self.mcp_client(
            context_request=f"Get PR details for {pr_url}",
            tool_name="get_pull_request",
            tool_args={"url": pr_url}
        )

        # Access the data properly from pr_data
        context_data = pr_data.get('context_data', {})
        diff_data = pr_data.get('diff', '') 

        try:
            # Analyze changes using modern predictor calls
            analysis = self.analyze_changes(pr_content=context_data)
            doc_review = self.review_docs(changes=diff_data)
            impact = self.analyze_impact(
                changes=diff_data,
                codebase=context_data
            )

            # Return a Prediction object with defaults
            return dspy.Prediction(

                edge_cases=getattr(analysis, 'edge_cases', []) or default_edge_cases,
                doc_updates=getattr(doc_review, 'doc_updates', []) or default_doc_updates,
                doc_suggestions=getattr(doc_review, 'doc_suggestions', []) or default_doc_suggestions,
                impact_analysis=getattr(impact, 'impact_analysis', '') or default_impact,
                risk_score=getattr(impact, 'risk_score', 0.0) or default_risk
            )
        except Exception as e:
            print(f"Analysis error: {str(e)}")
            # Return prediction with default values on error
            return dspy.Prediction(

                edge_cases=default_edge_cases,
                doc_updates=default_doc_updates,
                doc_suggestions=default_doc_suggestions,
                impact_analysis=default_impact,
                risk_score=default_risk
            )

def main():
    """Run PR Review Agent demo."""
    github_token = os.getenv('GITHUB_TOKEN')
    openai_key = os.getenv('OPENAI_API_KEY')
    
    if not all([github_token, openai_key]):
        raise EnvironmentError("Please set GITHUB_TOKEN and OPENAI_API_KEY")

    # Configure DSPy with modern configuration
    lm = dspy.LM('openai/gpt-4o-mini', api_key=openai_key)
    dspy.configure(lm=lm)

    # Setup MCP client
    mcp_client = RealMCPClient(
        ["npx", "-y", "@modelcontextprotocol/server-github"]
    )

    try:
        # Create and run agent
        agent = PRReviewAgent(mcp_client, github_token)
        result = agent.forward("https://github.com/SuperagenticAI/SuperXLab/pull/1")

        print("\n PR Review Results:")
        for key, value in result.items():
            print(f"\n{key.replace('_', ' ').title()}:")
            if isinstance(value, list):
                for item in value:
                    print(f"  - {item}")
            else:
                print(f"  {value}")
                
    except Exception as e:
        print(f"Error during PR review: {str(e)}")

if __name__ == "__main__":
    main()