GPT-5.1 Codex 实战指南与项目：构建 GitHub Issue 分析代理

OpenAI 已低调在开发者平台发布 GPT-5.1-Codex，开发者纷纷称其为当下最强的代码模型。与早期 Codex 版本不同，GPT-5.1-Codex 专为真实的软件工程、长时推理与工具型代理而打造。

在本教程中，我们将使用 OpenAI Agents 与 GPT-5.1 Codex 构建一个完整的 GitHub Issue 分析代理（GitHub Issue Analyzer Agent）。

该代理将能够：

• 直接从任意 GitHub 仓库抓取 Issue
• 理解、拆解并对 Issue 分类
• 仅检查仓库中相关的文件与目录
• 可选地使用 Firecrawl API 进行文档与网页检索
• 产出详细的、逐步的工程实施计划以解决 Issue

这个代理的行为像一位资深工程师：在动手前会先调研、阅读、推理并规划。您也可以查看我们关于 GPT-5.1 的指南，了解更多更新内容。

什么是 GPT-5.1-Codex？

GPT-5.1 Codex 是 GPT-5.1 的一个专业化版本，面向长时运行的代理型编码任务，而不仅仅是代码片段补全。它为真实的软件工程与代理式工作流而生，正是我们将 Issue 自动转化为计划这一流程的理想引擎。 

与通用模型不同，Codex 能像资深工程师一样理解代码库；它会阅读 Issue、基于架构进行推理、定位正确的目录，并只检查真正重要的文件。这使代理更快、更聪明且更具成本效率。

Codex 针对长时运行的代理型编码任务进行了优化。它可自然集成开发者工具，如 GitHub CLI 与 Firecrawl API，使我们的代理能够拉取 Issue、在线探索项目结构、读取特定文件，并按需收集文档。它能严格遵循指令，产出干净可靠的分析，并自适应推理力度：简单任务快速推进，复杂任务深入分析。

通过将强大的代码理解能力与工具感知的推理相结合，GPT-5.1 Codex 让我们的代理能够把一个 GitHub Issue 转化为准确、可执行的工程计划，而无需扫描整个仓库或幻想编造代码。它为整个工作流提供“工程直觉、结构与精确度”的支撑，是项目的中枢。

搭建 GitHub Issue 分析器

在开始之前，请先准备您的环境。您需要在机器上安装 Git。如果不确定，请运行 git --version 进行确认。您还需要一个 OpenAI 开发者平台账户，至少有 $6 额度，以保证 API 调用顺畅。

接着，注册免费的 Firecrawl 账号，并将 API Key 设置为环境变量。它们用于让分析器与 OpenAI 和 Firecrawl 通信：

export OPENAI_API_KEY=sk-...
export FIRECRAWL_API_KEY="fc-..." 技术支持

完成后，安装支撑该工作流的 Python 包。第一个 openai-agents 是一个轻量框架，让构建多代理流水线变得异常简单。第二个 firecrawl-py 负责爬取并从您的仓库或文档中抽取有用信息。

pip install openai-agents
pip install firecrawl-py技术支持

最后，请确保已安装并配置 GitHub CLI。以下命令可帮助您登录 GitHub 账户。

gh auth login技术支持

使用 GPT-5.1-Codex 构建 GitHub Issue 分析代理

我们将创建一个名为 "src" 的文件夹，包含所有代码文件。其中 "agents_pkg" 文件夹用于存放代理相关文件，"tools" 文件夹用于存放工具相关文件。

主应用程序 "app.py" 提供命令行界面（CLI），利用规划代理根据用户输入生成 GitHub Issue 报告。

项目目录结构如下：

1. Web 与 GitHub 工具

首先，我们将创建一个工具文件，通过简单的 Python 函数帮助代理访问 GitHub 与 Firecrawl API。

Firecrawl API 工具

先从工具开始。在 src/tools 目录下创建 firecrawl_tools.py 文件，并加入以下代码。

1. 使用环境变量中的 FIRECRAWL_API_KEY 创建并返回 Firecrawl 客户端；若缺失将抛出错误。

import json
import os

from agents import function_tool
from firecrawl import firecrawl


def _get_firecrawl_client():
    api_key = os.getenv("FIRECRAWL_API_KEY")
    if not api_key:
        raise RuntimeError("FIRECRAWL_API_KEY is not set")
    return firecrawl(api_key=api_key)技术支持

2. 使用 Firecrawl 进行聚焦的网页检索（如文档、博客、错误信息），并以 JSON 返回结果，供代理作为外部上下文使用。

@function_tool
def firecrawl_search(query: str, limit: int = 3) -> str:
    """
    Run a Firecrawl search for docs related to the issue or tech stack.

    Args:
        query: Search query (usually based on issue title / framework / error message).
        limit: Max number of results to return.

    Returns:
        JSON string of Firecrawl search results.
    """
    client = _get_firecrawl_client()
    results = client.search(query=query, limit=limit)
    return json.dumps(results)技术支持

3. 使用 Firecrawl 抓取单个 URL（markdown 格式），并将结构化页面内容以 JSON 返回，便于深入技术调研。

@function_tool
def firecrawl_scrape(url: str) -> str:
    """
    Scrape a single URL using Firecrawl for deeper research.

    Args:
        url: URL to scrape (docs, blog, README in another repo, etc.).

    Returns:
        JSON (markdown/structured) content from Firecrawl scrape.
    """
    client = _get_firecrawl_client()
    result = client.scrape(url=url, params={"formats": ["markdown"]})
    return json.dumps(result)技术支持

GitHub CLI 工具

然后，在 src/tools 目录下创建 github_tools.py 文件，并加入以下代码。

1. 通过 GitHub CLI 获取指定 Issue，并以 JSON 返回其详情，供代理阅读。

import base64
import json
import subprocess
from typing import List, Optional

from agents import function_tool


@function_tool
def get_github_issue(repo: str, issue_number: int) -> str:
    """
    Fetch a GitHub issue using the GitHub CLI.

    Args:
        repo: Repository in 'owner/name' format.
        issue_number: The issue number to fetch.

    Returns:
        A JSON string containing the issue fields (title, body, labels, URL, etc.),
        or an error payload if the command fails.
    """
    try:
        result = subprocess.run(
            [
                "gh",
                "issue",
                "view",
                str(issue_number),
                "--repo",
                repo,
                "--json",
                "number,title,body,labels,url,author,createdAt,state,assignees",
            ],
            capture_output=True,
            text=True,
            check=True,
        )
        return result.stdout
    except subprocess.CalledProcessError as e:
        return json.dumps(
            {
                "error": "Failed to fetch issue via GitHub CLI",
                "stderr": e.stderr,
                "repo": repo,
                "issue_number": issue_number,
            }
        )技术支持

2. 仅列出远程仓库中相关的文件（可按路径与扩展名过滤），避免代理扫描整个项目。

@function_tool
def list_repo_files_gh(
    repo: str,
    max_files: int = 80,
    extensions: Optional[List[str]] = None,
    path_prefixes: Optional[List[str]] = None,
) -> str:
    """
    List *relevant* files in the remote repo using GitHub CLI.

    Uses:
        gh api repos/{repo}/git/trees/HEAD?recursive=1

    The agent is expected to reason first which areas of the codebase are likely relevant
    (e.g. 'src/', 'app/', 'backend/api/', 'cli/'), and then call this tool with a small
    set of path_prefixes instead of scanning the entire project.

    Args:
        repo: Repository in 'owner/name' format (e.g. openai/openai-agents-python).
        max_files: Max number of files to return.
        extensions: Optional list of file extensions to keep (e.g. [".py", ".ts"]).
        path_prefixes: Optional list of path prefixes to include (e.g. ["src/", "app/api/"]).

    Returns:
        JSON string with file paths and filters applied.
    """
    try:
        result = subprocess.run(
            [
                "gh",
                "api",
                f"repos/{repo}/git/trees/HEAD?recursive=1",
            ],
            capture_output=True,
            text=True,
            check=True,
        )
    except subprocess.CalledProcessError as e:
        return json.dumps(
            {
                "error": "Failed to list repo files via GitHub CLI",
                "stderr": e.stderr,
                "repo": repo,
            }
        )

    try:
        data = json.loads(result.stdout)
    except json.JSONDecodeError:
        return json.dumps(
            {
                "error": "Failed to parse JSON from gh api",
                "raw": result.stdout[:2000],
                "repo": repo,
            }
        )

    tree = data.get("tree", [])

    if extensions is not None and not isinstance(extensions, list):
        extensions = [str(extensions)]
    exts = [e.lower() for e in (extensions or [])]

    if path_prefixes is not None and not isinstance(path_prefixes, list):
        path_prefixes = [str(path_prefixes)]
    prefixes = [p.strip() for p in (path_prefixes or []) if p.strip()]

    paths: List[str] = []
    for entry in tree:
        if entry.get("type") != "blob":
            continue  # only files
        path = entry.get("path", "")
        if not path:
            continue

        # If prefixes are provided, only keep files under those subtrees
        if prefixes and not any(path.startswith(pref) for pref in prefixes):
            continue

        if exts:
            suffix = "." + path.split(".")[-1].lower() if "." in path else ""
            if suffix not in exts:
                continue

        paths.append(path)
        if len(paths) >= max_files:
            break

    return json.dumps(
        {
            "repo": repo,
            "count": len(paths),
            "files": paths,
            "filtered_by_extensions": bool(exts),
            "filtered_by_prefixes": bool(prefixes),
        }
    )技术支持

3. 使用 GitHub CLI 下载并解码仓库中单个文件内容，将文本（必要时截断）以 JSON 返回。

@function_tool
def get_repo_file_gh(
    repo: str,
    path: str,
    ref: str = "",
    max_chars: int = 8000,
) -> str:
    """
    Read a file's contents from the remote repo using GitHub CLI.

    Uses:
        gh api repos/{repo}/contents/{path} [ -F ref=<branch> ]

    Args:
        repo: Repository in 'owner/name' format.
        path: File path in the repo (e.g. 'src/main.py').
        ref: Optional branch / commit / tag ref (default: repo's default branch).
        max_chars: Max characters of decoded content to return.

    Returns:
        JSON with file metadata and decoded content (truncated if needed),
        or an error payload if anything fails.
    """
    cmd = ["gh", "api", f"repos/{repo}/contents/{path}"]
    # Only add ref when explicitly set (GitHub default branch otherwise)
    if ref:
        cmd += ["-F", f"ref={ref}"]

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            check=True,
        )
    except subprocess.CalledProcessError as e:
        return json.dumps(
            {
                "error": "Failed to fetch file via GitHub CLI",
                "stderr": e.stderr,
                "repo": repo,
                "path": path,
                "ref": ref or "DEFAULT_BRANCH",
            }
        )

    try:
        data = json.loads(result.stdout)
    except json.JSONDecodeError:
        return json.dumps(
            {
                "error": "Failed to parse JSON from gh api (contents)",
                "raw": result.stdout[:2000],
                "repo": repo,
                "path": path,
            }
        )

    if data.get("type") != "file":
        return json.dumps(
            {
                "error": "Path is not a file",
                "repo": repo,
                "path": path,
                "data_type": data.get("type"),
            }
        )

    encoding = data.get("encoding")
    content_b64 = data.get("content", "")

    if encoding != "base64":
        return json.dumps(
            {
                "error": "Unexpected encoding",
                "repo": repo,
                "path": path,
                "encoding": encoding,
            }
        )

    try:
        # GitHub often includes newlines in base64 payload
        raw_bytes = base64.b64decode(content_b64)
        text = raw_bytes.decode("utf-8", errors="replace")
    except Exception as e:  # noqa: BLE001
        return json.dumps(
            {
                "error": f"Failed to decode file content: {e}",
                "repo": repo,
                "path": path,
                "encoding": encoding,
            }
        )

    truncated = text[:max_chars]
    return json.dumps(
        {
            "repo": repo,
            "path": path,
            "ref": ref or "DEFAULT_BRANCH",
            "truncated": len(text) > max_chars,
            "content": truncated,
        }
    )技术支持

2. 规划代理

这里我们定义一个 Issue Planner（问题规划） 代理，它会：

1. 读取 GitHub Issue
2. 判断代码库中哪些部分相关
3. 仅通过 GitHub CLI 工具检查一小组文件
4. 可选地调用 Firecrawl 查阅外部文档
5. 最终返回一份具体的、可执行的分步计划。

我们接入 GitHub 与 Firecrawl 工具，给代理详细的节省成本的工作指引，并指定运行在 gpt-5.1-codex 模型上。

在 src/agents_pkg 目录中创建 planner_agent.py 文件，并加入以下代码：

from agents import Agent

from tools.github_tools import (
    get_github_issue,
    list_repo_files_gh,
    get_repo_file_gh,
)
from tools.firecrawl_tools import (
    firecrawl_search,
    firecrawl_scrape,
)


def build_planner_agent() -> Agent:
    """
    Issue Planner agent that:
    - Reads the GitHub issue
    - Reasons about which parts of the repo are relevant
    - Uses GitHub CLI to inspect a *small* set of files online
    - Uses Firecrawl for external research
    - Outputs a concrete, step-by-step execution plan
    """
    return Agent(
        name="Issue Planner",
        instructions=(
            "You are a senior software engineer.\n"
            "Goal: Given a GitHub issue and the online repo (structure + files), plus optional "
            "external research, produce a clear, step-by-step execution plan to resolve the issue.\n\n"
            "CONTEXT:\n"
            "- All repository interaction must be done *online* via GitHub CLI tools.\n"
            "- You have tools to: read the issue, list files under certain paths, read specific files, "
            "  and call Firecrawl search/scrape for docs.\n\n"
            "IMPORTANT STRATEGY (BE SMART):\n"
            "- Be selective and cost-aware. Do NOT scan the whole project.\n"
            "- First, deeply read the issue and infer which part of the system it affects:\n"
            "  routing layer, CLI, API handlers, DB layer, tests, etc.\n"
            "- Based on this reasoning, decide a small list of path prefixes and file types.\n\n"
            "RECOMMENDED WORKFLOW:\n"
            "1. Call get_github_issue(repo, issue_number) to fully understand the problem.\n"
            "2. From the issue, infer a small list of path prefixes where relevant code likely lives,\n"
            "   e.g. ['src/', 'app/', 'backend/api/', 'cli/'] depending on the project style.\n"
            "3. Call list_repo_files_gh with:\n"
            "   - extensions like ['.py', '.ts', '.js', '.tsx', '.jsx']\n"
            "   - path_prefixes set to that small, targeted list\n"
            "   This keeps the search focused instead of scanning the entire project.\n"
            "4. From the returned file list, pick at most ~5-15 key files that are most likely related\n"
            "   (entrypoints, routers, handlers, services, tests).\n"
            "5. Call get_repo_file_gh(repo, path=...) only on those selected files to inspect the "
            "   actual implementation.\n"
            "6. If you need framework or library context (FastAPI, Click, React, etc.), use\n"
            "   firecrawl_search and firecrawl_scrape to pull official docs or good examples.\n\n"
            "OUTPUT FORMAT (execution plan):\n"
            "After you have enough context from the issue + targeted code inspection (+ optional research), "
            "output a concise but concrete plan with sections:\n"
            "   - Issue summary\n"
            "   - Project/codebase understanding (where this issue lives in the architecture)\n"
            "   - Key files / components to touch (with file paths)\n"
            "   - Step-by-step implementation plan (Step 1, Step 2, ...)\n"
            "   - Testing strategy (unit / integration / manual)\n"
            "   - Edge cases, risks, and any open questions\n\n"
            "The plan must be actionable for a mid-level developer. Avoid generic advice; tie your steps "
            "to the actual files and modules you inspected.\n"
        ),
        tools=[
            get_github_issue,
            list_repo_files_gh,
            get_repo_file_gh,
            firecrawl_search,
            firecrawl_scrape,
        ],
        model="gpt-5.1-codex",
    )技术支持

3. 主 CLI 应用

这是提供命令行界面（CLI）的主 Python 文件，将所有逻辑、回调与错误处理整合在一个文件中，并高效复用其他文件中的工具与代理。

1. 先完成 CLI 所需的一切导入与设置。包含标准库、在 Windows 上确保正确的 Unicode 输出（正确显示表情与符号），并导入 OpenAI 的运行器以及我们的规划代理。

2. 接着定义命令行选项，允许您在运行工具时传入仓库与 Issue 编号。

src/app.py：

import argparse
import asyncio
import json
import os
import pathlib
import sys
from datetime import datetime

# Set UTF-8 encoding for stdout to handle Unicode characters
if sys.platform == "win32":
    import codecs
    sys.stdout = codecs.getwriter("utf-8")(sys.stdout.detach())
    sys.stderr = codecs.getwriter("utf-8")(sys.stderr.detach())

from agents import Runner, ItemHelpers
from openai.types.responses import ResponseTextDeltaEvent

from agents_pkg.planner_agent import build_planner_agent


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Issue Planner: GPT-5.1-Codex + OpenAI Agents + GitHub CLI + Firecrawl\n"        )
    )
    parser.add_argument(
        "--repo",
        help="GitHub repo in 'owner/name' format (e.g. openai/openai-agents-python).",
    )
    parser.add_argument(
        "--issue",
        type=int,
        help="Issue number to plan for.",
    )
    return parser.parse_args()
技术支持

3. 然后收集输入并准备上下文。读取参数中的仓库与 Issue，确保已设置 OpenAI API Key，构建清晰的提示词，指导代理按步骤分析 Issue，并创建带时间戳的 markdown 文件用于保存最终计划。

def get_user_input(args: argparse.Namespace) -> tuple[str, int]:
    """Get repository and issue number from arguments or user input."""
    repo = args.repo or input("GitHub repo (owner/name): ").strip()
    issue_number = args.issue or int(input("Issue number: ").strip())
    return repo, issue_number


def validate_environment() -> None:
    """Validate that required environment variables are set."""
    if not os.getenv("OPENAI_API_KEY"):
        raise RuntimeError("OPENAI_API_KEY is not set")


def build_user_prompt(repo: str, issue_number: int) -> str:
    """Build the user prompt for the agent."""
    return (
        f"You are helping me plan how to implement GitHub issue #{issue_number} "
        f"in repo '{repo}'.\n\n"
        "Be selective and cost-aware:\n"
        "1. Use get_github_issue(repo, issue_number) to understand the problem.\n"
        "2. Based on the issue text, first reason about which directories and components "
        "   are likely relevant.\n"
        "3. Call list_repo_files_gh(repo, extensions=['.py', '.ts', '.js', '.tsx', '.jsx'], "
        "   path_prefixes=[<your inferred prefixes>]) to only explore those areas.\n"
        "4. From those results, choose a small set of the most relevant files and call "
        "   get_repo_file_gh(repo, path=...) on them.\n"
        "5. Optionally, use firecrawl_search and firecrawl_scrape if you need external docs.\n"
        "6. Finally, generate the execution plan in the structured format from your instructions."
    )


def setup_output_file(repo: str, issue_number: int) -> pathlib.Path:
    """Create output directory and return the markdown file path."""
    output_dir = pathlib.Path("output")
    output_dir.mkdir(exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    markdown_file = output_dir / f"execution_plan_{repo.replace('/', '_')}_issue_{issue_number}_{timestamp}.md"
    return markdown_file技术支持

4. 接着，我们让代理的思考过程可见。我们尝试从每个事件中提取干净的推理片段，并打印一行简短的“💭 Reasoning…”提示，让您看到模型在后台正在思考什么，而不是只展示底层细节。

def extract_reasoning_text(event_item) -> str | None:
    """Extract reasoning text from a reasoning event item."""
    reasoning_text = None
   
    if hasattr(event_item, 'raw_item'):
        raw = event_item.raw_item
        # Try multiple attribute names
        for attr_name in ['content', 'text', 'reasoning', 'message', 'delta']:
            if hasattr(raw, attr_name):
                val = getattr(raw, attr_name)
                if val and str(val).strip() and str(val) != 'None':
                    reasoning_text = str(val)
                    break
       
        # If still not found, try to access as dict-like
        if not reasoning_text:
            try:
                if hasattr(raw, '__dict__'):
                    for key, val in raw.__dict__.items():
                        if val and str(val).strip() and str(val) != 'None' and key in ['content', 'text', 'reasoning', 'message', 'delta']:
                            reasoning_text = str(val)
                            break
            except:
                pass
   
    # Also try direct attributes on event.item
    if not reasoning_text:
        for attr_name in ['content', 'text', 'reasoning', 'message']:
            if hasattr(event_item, attr_name):
                val = getattr(event_item, attr_name)
                if val and str(val).strip() and str(val) != 'None':
                    reasoning_text = str(val)
                    break
   
    return reasoning_text


def handle_reasoning_event(event_item) -> None:
    """Handle and display reasoning events."""
    reasoning_text = extract_reasoning_text(event_item)
   
    if reasoning_text and reasoning_text.strip():
        # Show first line or first 100 chars
        first_line = reasoning_text.split('\n')[0].strip()[:100]
        if len(reasoning_text.split('\n')[0].strip()) > 100:
            first_line += "..."
        print(f"\n💭 Reasoning: {first_line}", flush=True)
    else:
        # Don't show "None" - just show that reasoning is happening
        print(f"\n💭 Reasoning...", flush=True)技术支持

5. 随后我们处理工具调用与跟踪。检测代理使用的工具，将其参数（如 repo、path 或 query）格式化为简洁字符串，打印“🔧 Calling…”消息，并维护内部的活动工具映射，以便标记完成并在后续汇总。

def extract_tool_info(event_item) -> tuple[str | None, str | None]:
    """Extract tool name and ID from a tool call event item."""
    tool_name = None
    tool_id = None
   
    # First try raw_item which contains the actual tool call data
    if hasattr(event_item, 'raw_item'):
        raw = event_item.raw_item
        # Try accessing tool_call through various paths
        tool_call = None
        if hasattr(raw, 'tool_call'):
            tool_call = raw.tool_call
        elif hasattr(raw, 'function_call'):
            tool_call = raw.function_call
       
        if tool_call:
            # Try to get name from tool_call
            if hasattr(tool_call, 'name'):
                tool_name = tool_call.name
            elif hasattr(tool_call, 'function') and hasattr(tool_call.function, 'name'):
                tool_name = tool_call.function.name
            # Try to get ID
            if hasattr(tool_call, 'id'):
                tool_id = tool_call.id
            elif hasattr(tool_call, 'tool_call_id'):
                tool_id = tool_call.tool_call_id
       
        # Fallback: try direct attributes on raw
        if not tool_name:
            if hasattr(raw, 'name'):
                tool_name = getattr(raw, 'name')
            elif hasattr(raw, 'function') and hasattr(raw.function, 'name'):
                tool_name = raw.function.name
            # Try using getattr with different possible attribute names
            for attr_name in ['tool_name', 'function_name', 'name']:
                if hasattr(raw, attr_name):
                    tool_name = getattr(raw, attr_name, None)
                    if tool_name:
                        break
   
    # Fallback to direct attributes
    if not tool_name and hasattr(event_item, 'tool_call'):
        tool_call = event_item.tool_call
        if hasattr(tool_call, 'name'):
            tool_name = tool_call.name
        if hasattr(tool_call, 'id'):
            tool_id = tool_call.id
        elif hasattr(tool_call, 'function') and hasattr(tool_call.function, 'name'):
            tool_name = tool_call.function.name
    if not tool_name and hasattr(event_item, 'name'):
        tool_name = event_item.name
    if not tool_name and hasattr(event_item, 'function'):
        func = event_item.function
        if hasattr(func, 'name'):
            tool_name = func.name
   
    return tool_name, tool_id



def format_tool_arguments(tool_call_obj) -> str:
    """Format tool arguments for display."""
    if not tool_call_obj or not hasattr(tool_call_obj, 'arguments'):
        return ""
   
    try:
        args_dict = json.loads(tool_call_obj.arguments) if isinstance(tool_call_obj.arguments, str) else tool_call_obj.arguments
        if 'repo' in args_dict:
            tool_args = f" → {args_dict['repo']}"
            if 'issue_number' in args_dict:
                tool_args += f"#{args_dict['issue_number']}"
            return tool_args
        elif 'path' in args_dict:
            return f" → {args_dict['path']}"
        elif 'query' in args_dict:
            q = str(args_dict['query'])
            return f" → {q[:40]}..." if len(q) > 40 else f" → {q}"
        elif 'url' in args_dict:
            return f" → {args_dict['url']}"
        elif 'extensions' in args_dict or 'path_prefixes' in args_dict:
            parts = []
            if 'extensions' in args_dict:
                parts.append(f"ext={args_dict['extensions']}")
            if 'path_prefixes' in args_dict:
                parts.append(f"paths={args_dict['path_prefixes']}")
            return f" → {', '.join(parts)}"
    except:
        pass
   
    return ""


def handle_tool_call_event(event_item, active_tools: dict, tool_counter: int) -> tuple[int, bool]:
    """Handle tool call events and return updated tool_counter and whether event was handled."""
    tool_name, tool_id = extract_tool_info(event_item)
   
    if tool_name:
        tool_counter += 1
        tool_id = tool_id or f"tool_{tool_counter}"
        active_tools[tool_id] = tool_name
       
        # Get tool arguments if available
        tool_call_obj = None
        if hasattr(event_item, 'raw_item') and hasattr(event_item.raw_item, 'tool_call'):
            tool_call_obj = event_item.raw_item.tool_call
        elif hasattr(event_item, 'tool_call'):
            tool_call_obj = event_item.tool_call
       
        tool_args = format_tool_arguments(tool_call_obj)
        print(f"\n[{tool_counter}] 🔧 Calling: {tool_name}{tool_args}...", flush=True)
        return tool_counter, True
    else:
        # Still couldn't extract - try to inspect raw_item structure
        if hasattr(event_item, 'raw_item'):
            raw = event_item.raw_item
            try:
                raw_attrs = [attr for attr in dir(raw) if not attr.startswith('_')]
                # Look for attributes that might contain the tool name
                for attr in raw_attrs:
                    try:
                        val = getattr(raw, attr)
                        if isinstance(val, str) and ('get_github' in val.lower() or 'list_repo' in val.lower() or 'firecrawl' in val.lower()):
                            tool_name = val
                            break
                        # Check if it's a dict-like object with name
                        if hasattr(val, 'name'):
                            tool_name = val.name
                            break
                    except:
                        continue
               
                if tool_name:
                    tool_counter += 1
                    tool_id = tool_id or f"tool_{tool_counter}"
                    active_tools[tool_id] = tool_name
                    print(f"\n[{tool_counter}] 🔧 Calling: {tool_name}...", flush=True)
                    return tool_counter, True
                else:
                    # Print raw_item structure for debugging
                    print(f"\n[DEBUG] raw_item attrs: {raw_attrs[:10]}", flush=True)
            except Exception as e:
                print(f"\n[DEBUG] Error inspecting raw_item: {e}", flush=True)
   
    return tool_counter, False


def handle_tool_output_event(event_item, active_tools: dict, completed_tools: list) -> None:
    """Handle tool output events and track completed tools."""
    tool_id = None
   
    # Try raw_item first
    if hasattr(event_item, 'raw_item') and hasattr(event_item.raw_item, 'tool_call_id'):
        tool_id = event_item.raw_item.tool_call_id
    elif hasattr(event_item, 'tool_call_id'):
        tool_id = event_item.tool_call_id
    elif hasattr(event_item, 'raw_item') and hasattr(event_item.raw_item, 'tool_call'):
        if hasattr(event_item.raw_item.tool_call, 'id'):
            tool_id = event_item.raw_item.tool_call.id
    elif hasattr(event_item, 'tool_call'):
        if hasattr(event_item.tool_call, 'id'):
            tool_id = event_item.tool_call.id
        elif hasattr(event_item.tool_call, 'function') and hasattr(event_item.tool_call.function, 'name'):
            # Try to match by name if ID not available
            tool_name_match = event_item.tool_call.function.name
            for tid, tname in active_tools.items():
                if tname == tool_name_match:
                    tool_id = tid
                    break
   
    if tool_id and tool_id in active_tools:
        tool_name = active_tools.pop(tool_id)
        completed_tools.append(tool_name)
    elif active_tools:
        # Fallback: use the first active tool
        tool_id, tool_name = next(iter(active_tools.items()))
        active_tools.pop(tool_id)
        completed_tools.append(tool_name)技术支持

6. 现在我们接好流式循环与持久化。处理代理的流式事件，边到边打印令牌，实时显示推理与工具调用；若出现问题则优雅回退到非流式运行；最后将完整的执行计划写入带元数据的 markdown 文件。

async def process_streaming_events(result, repo: str, issue_number: int) -> str:
    """Process streaming events from the agent execution."""
    final_output = ""
    active_tools = {}  # Track active tool calls by ID
    tool_counter = 0
    completed_tools = []
    first_event_received = False
   
    # Stream the events as they come in
    async for event in result.stream_events():
        # Handle raw response events (token-by-token streaming) - print immediately
        if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
            if not first_event_received:
                first_event_received = True
            delta = event.data.delta
            print(delta, end="", flush=True)
            final_output += delta
        # Handle run item events (higher level updates)
        elif event.type == "run_item_stream_event":
            item_type = getattr(event.item, 'type', 'unknown')
           
            # Show reasoning events
            if item_type == "reasoning_item":
                handle_reasoning_event(event.item)
            elif item_type == "tool_call_item":
                tool_counter, handled = handle_tool_call_event(event.item, active_tools, tool_counter)
                if handled:
                    first_event_received = True
            elif item_type == "tool_call_output_item":
                handle_tool_output_event(event.item, active_tools, completed_tools)
            elif item_type == "message_output_item":
                message_text = ItemHelpers.text_message_output(event.item)
                if message_text and (not final_output or message_text not in final_output):
                    print(f"\n{message_text}", flush=True)
                    final_output += message_text
   
    print()  # Add newline after streaming
   
    # Show summary of tools used
    if completed_tools:
        print(f"---\n\n📊 Tools used ({len(completed_tools)}): {', '.join(completed_tools)}", flush=True)
   
    # If no streaming events occurred, fall back to final output
    if not final_output:
        final_output = result.final_output
        if final_output:
            print(final_output, flush=True)
   
    return final_output

async def run_agent_with_streaming(agent, user_prompt: str, repo: str, issue_number: int) -> str:
    """Run the agent with streaming support and fallback handling."""
    try:
        # Run agent with streaming (run_streamed is synchronous, returns immediately)
        result = Runner.run_streamed(
            agent,
            input=user_prompt,
            context={"repo": repo, "issue_number": issue_number},
        )
       
        return await process_streaming_events(result, repo, issue_number)
       
    except Exception as e:
        print(f"⚠️  Error: {e}", flush=True)
        # Fallback to standard async run
        result = await Runner.run(
            agent,
            input=user_prompt,
            context={"repo": repo, "issue_number": issue_number},
        )
        print(result.final_output, flush=True)
        return result.final_output

def save_output_to_file(markdown_file: pathlib.Path, repo: str, issue_number: int, final_output: str) -> None:
    """Save the final output to a markdown file."""
    with open(markdown_file, 'w', encoding='utf-8') as f:
        f.write(f"# GitHub Issue Analysis: {repo}#{issue_number}\n\n")
        f.write(f"**Generated on:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
        f.write(f"**Repository:** {repo}\n")
        f.write(f"**Issue Number:** {issue_number}\n\n")
        f.write("---\n\n")
        f.write(final_output)
   
    print(f"---\n\n✅ Saved: {markdown_file}", flush=True)技术支持

7. 最后在主入口中将一切串联起来：解析参数、获取仓库与 Issue，校验环境、构建代理与提示词，使用流式方式运行规划器（必要时回退为同步运行），并保存结果。

async def main() -> None:
    """Main entry point for the application."""
    args = parse_args()
    repo, issue_number = get_user_input(args)
    validate_environment()
   
    agent = build_planner_agent()
    user_prompt = build_user_prompt(repo, issue_number)
    markdown_file = setup_output_file(repo, issue_number)
   
    print(f"\n🔍 Analyzing {repo}#{issue_number}...\n")
   
    # Run the agent with streaming support
    try:
        final_output = await run_agent_with_streaming(agent, user_prompt, repo, issue_number)
    except Exception as e:
        print(f"⚠️  Error: {e}", flush=True)
        # Final fallback to standard sync run
        result = Runner.run_sync(
            agent,
            input=user_prompt,
            context={"repo": repo, "issue_number": issue_number},
        )
        print(result.final_output, flush=True)
        final_output = result.final_output

    # Save output to file
    save_output_to_file(markdown_file, repo, issue_number, final_output)


if __name__ == "__main__":
    asyncio.run(main())技术支持

注意：源代码、配置与文档可在 GitHub 仓库查看： kingabzpro/Issue-Analyzer。复现结果时请参考。

测试 GitHub Issue 分析器

我们的 CLI 应用有两种使用方式：交互模式，应用会逐步提示您提供仓库名与 Issue 编号；以及 CLI 模式，启动应用时需直接提供全部信息。

启动交互模式，输入以下命令：

python src/app.py技术支持

启动后，系统会提示您提供仓库名与 Issue 编号，应用将开始通过工具与推理来协助您。

很快您就会收到该 Issue 的摘要与解决思路。摘要包含所用工具的详情，以及保存信息的 markdown 文件位置。

您可以打开该 markdown 文件查看详细的 Issue 计划。

CLI 模式需要在命令中直接包含仓库名与 Issue 参数，如下所示：

python src/app.py --repo kingabzpro/Travel-with-Kimi-K2 --issue 1技术支持

整个过程都是流式的，您将看到代理使用了哪些工具、是否在有效地进行推理，并在最后实时输出生成的响应。

后续改进

当前版本的 GitHub Issue 分析器旨在理解问题并生成准确的执行计划。

然而，代理式工作流的真正潜力在于自动化计划生成之后的所有步骤。以下是您可以在系统中实现的一些重大升级：

1. 自动分支与 PR 创建

项目目标是将计划自动转化为可运行的 Pull Request，从而简化开发流程，提升工作效率。

该功能将包括代理直接创建新分支，便于实施更改。同时，代理会基于生成的计划应用代码修改，并运行 GitHub CLI 工作流，如 gh pr create 与 gh pr view。

此外，系统将自动生成 PR 描述、变更日志与关联 Issue 引用，并智能打标签，如 Bug 修复、功能增强或重构等类型。

总体而言，此举将代理有效升级为从 Issue 到 Pull Request 的全自动化系统，显著提升开发周期中的生产力与一致性。

2. 批量 Issue 分析与批处理工具执行

该升级将通过为 GitHub Issue 引入批处理能力来提升分析流程。团队可以对多个 Issue 进行并行或排队分析，从而更有效地处理大量问题。

此外，此升级还能帮助识别重复或相关问题，提升组织管理能力。Issue 也可按复杂度、子系统或影响范围等维度进行分类。

为进一步提效，团队可批量运行 GitHub 或 Firecrawl 工具，显著提升效率。最终，这一升级可使团队通过一条命令自动分拣数十甚至上百个 Issue。

3. PR 前测试、安全检查与验证

在创建 Pull Request（PR）之前，代理应当对拟议更改进行充分验证，而不是仅仅生成代码。该验证流程应包括多项计划特性：通过 GitHub Actions 或本地测试运行器执行单元测试，确保代码行为符合预期。

依赖项验证同样关键：检查导入是否准确、是否缺失模块，并解决版本不匹配等可能影响项目的问题。同时进行 Lint、格式化与类型检查，以维护代码质量。

必须确保修改不会破坏现有构建流水线，并能检测到任何 API 的破坏性变更或回归。遵循这些步骤可确保 PR 干净、安全、可用于生产部署。

关于 GPT-5.1 Codex 的最后思考

使用 GPT-5.1 Codex 与 openai-agents 构建高级、多步骤的代理，出乎意料地简单。您只需定义自己的工具，并清晰指示模型在何时、如何使用它们即可。

本教程中我选择 GitHub CLI，因为它快速、直观、易集成；但您同样可以使用 GitHub Python SDK、直接 API 调用，或其他 CLI/Bash 工具。灵活性才是关键。

您可以按需无限扩展这一套配置。

例如，您可以创建：

• 规划代理（我们已构建）
• 行动代理，根据计划应用代码更改
• 测试代理，运行测试并验证更改是否引入问题
• PR 代理，创建包含清晰摘要的 Pull Request

本教程的目标是展示 GPT-5.1 Codex 的真正能力：轻松调用工具、理解大型代码库、执行结构化推理，并能在无需频繁人工干预的情况下运行长链路自动化。

如果您希望进一步学习如何创建 AI 代理，推荐查看我们的 Google ADK AI 代理课程，以及我们的 AI 代理项目清单。