一个Python极简智能体框架:bu-agent-sdk,我用它搞了一个自动代码评审

# 使用 uv 包管理器安装
uv add bu-bu-agent-sdk

# 或使用 pip
pip install bu-bu-agent-sdk

import asyncio
from bu_agent_sdk import Agent, tool, TaskComplete
from bu_agent_sdk.llm import ChatAnthropic

# 定义工具：加法
@tool("Add two numbers")
async def add(a: int, b: int) -> int:
    """将两个数字相加"""
    return a + b

# 定义工具：任务完成信号
@tool("Signal task completion")
async def done(message: str) -> str:
    """当任务完成时调用此工具"""
    raise TaskComplete(message)

# 创建智能体
agent = Agent(
    llm=ChatAnthropic(model="claude-sonnet-4-20250514"),
    tools=[add, done],
)

# 执行查询
async def main():
    result = await agent.query("计算 25 + 17 等于多少？")
    print(result)

asyncio.run(main())

代码评审智能体
├── 工具集（Tools）
│   ├── read_file: 读取代码文件
│   ├── list_changed_files: 获取变更文件列表
│   ├── get_file_diff: 获取文件的 Git diff
│   ├── post_review_comment: 发布评审意见
│   └── done: 完成评审
├── 上下文管理
│   └── GitHubContext: 管理 GitHub API 交互
└── 评审逻辑
    └── 由 LLM 自主决策评审流程

from dataclasses import dataclass
from github import Github
from typing import List, Optional

@dataclass
class GitHubContext:
    """GitHub 操作的上下文管理器"""
    repo_name: str
    pr_number: int
    github_token: str
    
    def __post_init__(self):
        self.github = Github(self.github_token)
        self.repo = self.github.get_repo(self.repo_name)
        self.pr = self.repo.get_pull(self.pr_number)

from bu_agent_sdk import tool, TaskComplete, Depends
from typing import Annotated

def get_github_context() -> GitHubContext:
    """依赖注入：获取 GitHub 上下文"""
    # 这是一个占位符，实际的上下文会在运行时注入
    pass

@tool("List all files changed in this pull request")
async def list_changed_files(
    ctx: Annotated[GitHubContext, Depends(get_github_context)]
) -> str:
    """获取 PR 中所有变更的文件列表"""
    files = ctx.pr.get_files()
    file_list = [f.filename for f in files]
    return f"Changed files ({len(file_list)}):\n" + "\n".join(f"- {f}" for f in file_list)

@tool("Get the git diff for a specific file")
async def get_file_diff(
    filename: str,
    ctx: Annotated[GitHubContext, Depends(get_github_context)]
) -> str:
    """获取指定文件的 Git diff"""
    files = ctx.pr.get_files()
    for file in files:
        if file.filename == filename:
            return f"Diff for {filename}:\n```diff\n{file.patch}\n```"
    return f"File {filename} not found in this PR"

@tool("Read the full content of a file")
async def read_file(
    filename: str,
    ctx: Annotated[GitHubContext, Depends(get_github_context)]
) -> str:
    """读取文件的完整内容"""
    try:
        content = ctx.repo.get_contents(filename, ref=ctx.pr.head.sha)
        return f"Content of {filename}:\n```\n{content.decoded_content.decode('utf-8')}\n```"
    except Exception as e:
        return f"Error reading {filename}: {str(e)}"

@tool("Post a review comment on the pull request")
async def post_review_comment(
    body: str,
    ctx: Annotated[GitHubContext, Depends(get_github_context)]
) -> str:
    """在 PR 上发布评审评论"""
    ctx.pr.create_issue_comment(body)
    return "Review comment posted successfully"

@tool("Signal that code review is complete")
async def done(message: str) -> str:
    """标记代码评审完成"""
    raise TaskComplete(message)

from bu_agent_sdk import Agent
from bu_agent_sdk.llm import ChatAnthropic

# 定义系统提示词
SYSTEM_PROMPT = """你是一个专业的代码评审助手。你的任务是：

1. 首先查看 PR 中有哪些文件被修改
2. 对每个重要文件，获取其 diff 并仔细审查
3. 关注以下方面：
   - 代码质量和可读性
   - 潜在的 bug 或逻辑错误
   - 性能问题
   - 安全隐患
   - 最佳实践的遵守情况
4. 如果发现问题，使用 post_review_comment 工具发布详细的评审意见
5. 完成所有文件的评审后，调用 done 工具

请保持专业、建设性的态度，提供具体可行的改进建议。
"""

def create_code_review_agent(
    github_token: str,
    repo_name: str,
    pr_number: int
) -> Agent:
    """创建代码评审智能体"""
    
    # 创建 GitHub 上下文
    ctx = GitHubContext(
        repo_name=repo_name,
        pr_number=pr_number,
        github_token=github_token
    )
    
    # 创建智能体
    agent = Agent(
        llm=ChatAnthropic(model="claude-sonnet-4-20250514"),
        tools=[
            list_changed_files,
            get_file_diff,
            read_file,
            post_review_comment,
            done
        ],
        system_prompt=SYSTEM_PROMPT,
        dependency_overrides={get_github_context: lambda: ctx},
        require_done_tool=True  # 强制要求调用 done 工具
    )
    
    return agent

async def run_code_review(agent: Agent, pr_description: str):
    """执行代码评审，并打印实时进度"""
    
    print("🤖 Code Review Agent Started")
    print("=" * 50)
    
    task = f"Please review this pull request:\n{pr_description}"
    
    async for event in agent.query_stream(task):
        # 工具调用事件
        if hasattr(event, 'tool') and hasattr(event, 'args'):
            tool_name = event.tool
            print(f"🔧 Calling tool: {tool_name}")
            if event.args:
                print(f"   Args: {event.args}")
        
        # 工具结果事件
        elif hasattr(event, 'tool') and hasattr(event, 'result'):
            tool_name = event.tool
            result_preview = str(event.result)[:100]
            print(f"✅ {tool_name} completed")
            print(f"   Result preview: {result_preview}...")
        
        # 最终响应事件
        elif hasattr(event, 'content') and event.content:
            print(f"\n📝 Final Summary:\n{event.content}")
    
    print("=" * 50)
    print("✨ Code Review Completed!")

import os
import asyncio

async def main():
    # 从环境变量获取配置
    github_token = os.environ.get("GITHUB_TOKEN")
    repo_name = os.environ.get("GITHUB_REPOSITORY")  # 格式: owner/repo
    pr_number = int(os.environ.get("PR_NUMBER"))
    pr_description = os.environ.get("PR_DESCRIPTION", "")
    
    # 创建智能体
    agent = create_code_review_agent(
        github_token=github_token,
        repo_name=repo_name,
        pr_number=pr_number
    )
    
    # 执行评审
    await run_code_review(agent, pr_description)

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

name: AI Code Review

on:
  pull_request:
    types: [opened, synchronize, reopened]

jobs:
  code-review:
    runs-on: ubuntu-latest
    
    permissions:
      contents: read
      pull-requests: write
    
    steps:
      - name: Checkout code
        uses: actions/checkout@v4
        with:
          fetch-depth: 0  # 获取完整历史以便做 diff
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Install dependencies
        run: |
          pip install bu-bu-agent-sdk PyGithub anthropic
      
      - name: Run AI Code Review
        env:
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
          GITHUB_REPOSITORY: ${{ github.repository }}
          PR_NUMBER: ${{ github.event.pull_request.number }}
          PR_DESCRIPTION: ${{ github.event.pull_request.body }}
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: |
          python code_review_agent.py

@tool("Signal completion")
async def done(message: str) -> str:
    raise TaskComplete(message)

agent = Agent(
    llm=llm,
    tools=[..., done],
    require_done_tool=True,  # 开启自主模式
)

@tool("Read large log file", ephemeral=True)
async def read_log(filename: str) -> str:
    """读取日志文件（内容会被自动清理）"""
    with open(filename) as f:
        return f.read()

agent = Agent(
    llm=llm,
    tools=[read_log],
    ephemeral_limit=3  # 只保留最近 3 次临时工具的输出
)

agent = Agent(
    llm=llm,
    tools=tools,
    context_compaction_threshold=0.8,  # 达到 80% 上下文长度时触发压缩
)

from bu_agent_sdk.llm import ChatAnthropic, ChatOpenAI, ChatGoogle

# 使用 Claude
agent = Agent(llm=ChatAnthropic(model="claude-sonnet-4-20250514"), tools=tools)

# 切换到 GPT-4
agent = Agent(llm=ChatOpenAI(model="gpt-4o"), tools=tools)

# 或者使用 Gemini
agent = Agent(llm=ChatGoogle(model="gemini-2.0-flash"), tools=tools)

from typing import Annotated
from bu_agent_sdk import Depends

# 定义依赖
class DatabaseConnection:
    def __init__(self, db_url: str):
        self.db_url = db_url
    
    async def query(self, sql: str):
        # 执行数据库查询
        pass

def get_db() -> DatabaseConnection:
    """依赖提供者"""
    pass

# 在工具中使用依赖
@tool("Query user data")
async def get_user(
    user_id: int,
    db: Annotated[DatabaseConnection, Depends(get_db)]
) -> str:
    """查询用户数据"""
    result = await db.query(f"SELECT * FROM users WHERE id = {user_id}")
    return str(result)

# 创建智能体时注入实际的数据库连接
db_conn = DatabaseConnection("postgresql://localhost/mydb")

agent = Agent(
    llm=llm,
    tools=[get_user],
    dependency_overrides={get_db: lambda: db_conn}
)

from bu_agent_sdk.agent import ToolCallEvent, ToolResultEvent, FinalResponseEvent

async for event in agent.query_stream("do something"):
    match event:
        case ToolCallEvent(tool=name, args=args):
            print(f"Calling {name}")
        case ToolResultEvent(tool=name, result=result):
            print(f"{name} -> {result[:50]}")
        case FinalResponseEvent(content=text):
            print(f"Done: {text}")