Hướng Dẫn GPT-5.1 Codex Với Dự Án Thực Hành: Xây Dựng Tác Nhân Phân Tích Vấn Đề GitHub

OpenAI đã âm thầm phát hành GPT-5.1-Codex trên nền tảng dành cho nhà phát triển, và các developer đã gọi đây là mô hình lập trình tốt nhất hiện nay. Khác với các phiên bản Codex trước, GPT-5.1-Codex được thiết kế cho kỹ nghệ phần mềm thực thụ, suy luận dài hơi và các tác nhân biết sử dụng công cụ.

Trong hướng dẫn này, chúng ta sẽ xây dựng trọn vẹn một Tác Nhân Phân Tích Issue GitHub bằng OpenAI Agents và GPT-5.1 Codex.

Tác nhân của chúng ta sẽ:

• Lấy issue trực tiếp từ bất kỳ kho GitHub nào
• Hiểu, phân rã và phân loại issue
• Chỉ kiểm tra các tệp và thư mục liên quan trong repo
• Tùy chọn tìm kiếm tài liệu & tìm web bằng Firecrawl API
• Tạo một kế hoạch kỹ thuật chi tiết, từng bước để xử lý issue

Tác nhân này hành xử như một kỹ sư cấp cao: nghiên cứu, đọc, suy luận và lập kế hoạch trước khi viết bất cứ điều gì. Bạn có thể xem hướng dẫn về GPT-5.1 riêng của chúng tôi để xem có gì mới khác.

GPT-5.1-Codex là gì?

GPT-5.1 Codex là phiên bản chuyên biệt của GPT-5.1, được xây dựng cho các tác vụ lập trình dạng tác nhân, chạy dài thay vì chỉ tự động hoàn thành đoạn mã. Nó được thiết kế có chủ đích cho kỹ nghệ phần mềm thực tế và các quy trình tác nhân, phù hợp để làm động cơ cho quy trình tự động hóa Issue-to-Plan của chúng ta. 

Khác với các mô hình tổng quát, Codex hiểu codebase như một kỹ sư cấp cao: đọc issue, suy luận về kiến trúc, xác định đúng thư mục, và chỉ kiểm tra các tệp thật sự quan trọng. Điều này giúp tác nhân nhanh hơn, thông minh hơn và tiết kiệm chi phí hơn rất nhiều.

Codex được tối ưu cho các tác vụ lập trình dạng tác nhân, chạy dài. Nó tích hợp tự nhiên với các công cụ developer như GitHub CLI và Firecrawl API, cho phép tác nhân lấy issue, khám phá cấu trúc dự án trực tuyến, đọc tệp cụ thể và thu thập tài liệu khi cần. Nó tuân thủ hướng dẫn chặt chẽ, tạo phân tích sạch và tin cậy, đồng thời điều chỉnh nỗ lực suy luận để xử lý nhanh các tác vụ đơn giản và đào sâu với các tác vụ phức tạp.

Bằng cách kết hợp khả năng hiểu mã mạnh mẽ với suy luận nhận thức công cụ, GPT-5.1 Codex giúp tác nhân của chúng ta biến một issue GitHub thành một kế hoạch kỹ thuật chính xác, có thể hành động, mà không cần quét toàn bộ repo hay bịa đặt mã. Nó là xương sống của quy trình vì mang lại trực giác kỹ thuật, cấu trúc và độ chính xác mà dự án phụ thuộc vào.

Thiết lập Tác Nhân Phân Tích Issue GitHub

Trước khi bắt đầu dự án, hãy đảm bảo môi trường của bạn đã sẵn sàng. Bạn cần cài Git trên máy. Nếu chưa chắc, chỉ cần chạy git --version để kiểm tra. Bạn cũng cần một tài khoản OpenAI Developer Platform với tối thiểu $6 tín dụng để các cuộc gọi API chạy không bị gián đoạn.

Tiếp theo, tạo tài khoản Firecrawl miễn phí và đặt khóa API của bạn thành biến môi trường. Đây là những gì cho phép bộ phân tích của bạn giao tiếp với OpenAI và Firecrawl:

export OPENAI_API_KEY=sk-...
export FIRECRAWL_API_KEY="fc-..." Được hỗ trợ bởi

Khi đã xong, cài các gói Python vận hành quy trình. Gói đầu tiên, openai-agents, là framework gọn nhẹ giúp xây pipeline đa tác nhân rất dễ dàng. Gói thứ hai, firecrawl-py, xử lý việc crawl và trích xuất thông tin hữu ích từ repo hoặc tài liệu của bạn.

pip install openai-agents
pip install firecrawl-pyĐược hỗ trợ bởi

Cuối cùng, đảm bảo GitHub CLI đã được cài và cấu hình. Lệnh sau sẽ giúp bạn đăng nhập vào tài khoản GitHub. 

gh auth loginĐược hỗ trợ bởi

Xây dựng Tác Nhân Phân Tích Issue GitHub Với GPT-5.1-Codex

Chúng ta sẽ tạo thư mục "src" chứa toàn bộ tệp mã. Thư mục "agents_pkg" sẽ chứa tất cả tệp tác nhân, trong khi thư mục "tools" sẽ chứa tất cả tệp công cụ. 

Ứng dụng chính, "app.py", có giao diện dòng lệnh (CLI) sử dụng các tác nhân lập kế hoạch để tạo báo cáo issue GitHub dựa trên đầu vào của người dùng.

Cấu trúc thư mục dự án của bạn sẽ như sau:

1. Công cụ Web và GitHub

Đầu tiên, chúng ta sẽ tạo tệp công cụ tác nhân giúp tác nhân truy cập GitHub và Firecrawl API bằng các hàm Python đơn giản.

Công cụ Firecrawl API

Bắt đầu với các công cụ. Tạo tệp firecrawl_tools.py trong thư mục src/tools và thêm đoạn mã sau. 

1. Tạo và trả về client Firecrawl sử dụng FIRECRAWL_API_KEY từ môi trường, và sẽ báo lỗi nếu thiếu khóa.

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)Được hỗ trợ bởi

2. Dùng Firecrawl để chạy tìm kiếm web có trọng tâm (ví dụ: tài liệu, blog, lỗi) và trả về kết quả dạng JSON để tác nhân dùng làm ngữ cảnh bên ngoài.

@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)Được hỗ trợ bởi

3. Quét một URL bằng Firecrawl (ở định dạng markdown) và trả về nội dung trang có cấu trúc dạng JSON để nghiên cứu kỹ thuật sâu hơn.

@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)Được hỗ trợ bởi

Công cụ GitHub CLI

Tiếp theo, tạo tệp github_tools.py trong thư mục src/tools và thêm đoạn mã sau. 

1. Lấy một issue GitHub cụ thể qua GitHub CLI và trả về chi tiết của nó dạng JSON để tác nhân đọc.

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,
            }
        )Được hỗ trợ bởi

2. Liệt kê chỉ các tệp liên quan trong repo GitHub từ xa (có thể lọc theo đường dẫn và phần mở rộng) để tác nhân không phải quét toàn bộ dự án.

@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),
        }
    )Được hỗ trợ bởi

3. Tải xuống và giải mã nội dung của một tệp trong repo bằng GitHub CLI, trả về văn bản (có thể cắt bớt nếu cần) dưới dạng 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,
        }
    )Được hỗ trợ bởi

2. Tác nhân lập kế hoạch 

Ở đây, chúng ta định nghĩa một tác nhân Issue Planner biết cách: 

1. Đọc một issue GitHub
2. Quyết định phần nào của codebase là liên quan
3. Chỉ kiểm tra một tập nhỏ tệp qua các công cụ GitHub CLI
4. Tùy chọn gọi Firecrawl để lấy tài liệu bên ngoài
5. Cuối cùng, trả về một kế hoạch thực thi cụ thể, từng bước. 

Chúng ta nối các công cụ GitHub và Firecrawl, cung cấp hướng dẫn chi tiết để tác nhân làm việc tiết kiệm chi phí, và chỉ định chạy trên mô hình gpt-5.1-codex.

Tạo tệp planner_agent.py trong thư mục src/agents_pkg và thêm đoạn mã sau:

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",
    )Được hỗ trợ bởi

3. Ứng dụng CLI chính

Đây là tệp Python chính cung cấp giao diện dòng lệnh (CLI) và tích hợp toàn bộ logic, callback, và xử lý lỗi trong một tệp thống nhất. Nó sử dụng hiệu quả các công cụ và tác nhân từ các tệp khác.

1. Trước hết, chúng ta thiết lập và import mọi thứ CLI cần. Bao gồm thư viện chuẩn, đảm bảo xuất Unicode đúng trên Windows để emoji và ký hiệu hiển thị chính xác, và import OpenAI runner cùng tác nhân lập kế hoạch của chúng ta. 

2. Tiếp theo, chúng ta định nghĩa các tùy chọn dòng lệnh, cho phép bạn truyền vào kho GitHub và số issue khi chạy công cụ.

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()
Được hỗ trợ bởi

3. Sau đó, chúng ta thu thập đầu vào và chuẩn bị ngữ cảnh. Đọc repo và issue từ tham số, đảm bảo OpenAI API key đã được đặt, xây dựng prompt rõ ràng để hướng dẫn tác nhân phân tích issue từng bước, và tạo tệp markdown đóng dấu thời gian để lưu kế hoạch cuối cùng.

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Được hỗ trợ bởi

4. Tiếp theo, chúng ta hiển thị tư duy của tác nhân. Chúng ta cố gắng trích xuất một đoạn suy luận gọn gàng từ mỗi sự kiện và in một dòng “💭 Đang suy luận…” ngắn, để thay vì nội bộ thô, bạn thấy gợi ý dễ đọc về việc mô hình đang làm gì ở hậu trường.

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)Được hỗ trợ bởi

5. Sau đó, chúng ta xử lý các lần gọi công cụ và theo dõi. Chúng ta phát hiện công cụ nào mà tác nhân đang dùng, định dạng đối số của nó (như repo, path, hoặc query) thành chuỗi gọn, in thông điệp “🔧 Đang gọi…”, và giữ một bản đồ nội bộ các công cụ đang hoạt động để đánh dấu chúng hoàn tất và sau đó tóm tắt mọi thứ đã chạy.

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)Được hỗ trợ bởi

6. Bây giờ, chúng ta nối vòng lặp streaming và lưu trữ. Xử lý các sự kiện streaming từ tác nhân, in token khi đến, hiển thị suy luận và gọi công cụ theo thời gian thực, chuyển rãnh mượt mà sang chế độ không streaming nếu có sự cố, và cuối cùng ghi toàn bộ kế hoạch thực thi vào tệp markdown kèm một số metadata hữu ích.

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)Được hỗ trợ bởi

7. Cuối cùng, chúng ta kết nối mọi thứ trong điểm vào chính. Phân tích tham số, lấy repo và issue, xác thực môi trường, xây dựng tác nhân và prompt, chạy planner với streaming (rơi về chạy đồng bộ nếu cần), và lưu kết quả. 

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())Được hỗ trợ bởi

Lưu ý: Mã nguồn, cấu hình và tài liệu có tại kho GitHub: kingabzpro/Issue-Analyzer. Vui lòng xem và dùng làm hướng dẫn khi tái lập kết quả.

Kiểm thử Tác Nhân Phân Tích Issue GitHub

Có hai cách dùng ứng dụng CLI của chúng tôi: Chế độ tương tác, ứng dụng sẽ lần lượt hỏi bạn tên kho và số issue, và Chế độ CLI, nơi bạn phải cung cấp toàn bộ thông tin khi khởi chạy ứng dụng.

Để bắt đầu chế độ tương tác, nhập lệnh sau:

python src/app.pyĐược hỗ trợ bởi

Khi khởi động, bạn sẽ được nhắc nhập tên kho và số issue, và ứng dụng sẽ bắt đầu sử dụng công cụ và suy luận để hỗ trợ bạn.

Kết quả là, chỉ trong vài giây, bạn sẽ nhận được tóm tắt issue và cách giải quyết. Bản tóm tắt bao gồm chi tiết về các công cụ đã dùng và vị trí tệp markdown nơi thông tin được lưu.

Bạn có thể mở tệp markdown để xem một kế hoạch xử lý issue chi tiết.

Chế độ CLI yêu cầu tên kho và tham số issue được đưa trực tiếp trong lệnh, như dưới đây:

python src/app.py --repo kingabzpro/Travel-with-Kimi-K2 --issue 1Được hỗ trợ bởi

Toàn bộ quá trình được stream, nghĩa là bạn sẽ thấy tác nhân đang dùng công cụ nào và liệu nó đang suy luận hiệu quả hay không. Phản hồi cuối cùng cũng sẽ được stream.

Cải tiến trong tương lai

Phiên bản hiện tại của Tác Nhân Phân Tích Issue GitHub được thiết kế để hiểu các issue và tạo kế hoạch thực thi chính xác. 

Tuy nhiên, tiềm năng thực sự của quy trình tác nhân nằm ở việc tự động hóa toàn bộ các bước sau khi kế hoạch đã được tạo. Dưới đây là một số nâng cấp lớn bạn có thể triển khai trong hệ thống:

1. Tự động tạo nhánh và PR

Mục tiêu của dự án là tự động chuyển một kế hoạch thành một pull request hoạt động. Điều này sẽ tinh gọn quy trình phát triển, cho phép luồng công việc hiệu quả hơn.

Tính năng này sẽ bao gồm khả năng tạo nhánh mới trực tiếp từ tác nhân, hỗ trợ việc áp dụng thay đổi. Ngoài ra, tác nhân sẽ áp dụng chỉnh sửa mã dựa trên kế hoạch đã tạo và chạy các workflow GitHub CLI như gh pr create và gh pr view.

Hơn nữa, hệ thống sẽ tự động tạo mô tả pull request, changelog, và liên kết issue. Hệ thống cũng sẽ áp dụng nhãn một cách thông minh, phân loại như sửa lỗi, cải tiến hay tái cấu trúc.

Tổng thể, sáng kiến này biến tác nhân thành một hệ thống tự động hóa từ Issue đến Pull Request hoàn chỉnh, nâng cao đáng kể năng suất và tính nhất quán trong chu trình phát triển.

2. Phân tích issue theo lô và chạy công cụ hàng loạt

Nâng cấp này sẽ cải thiện quá trình phân tích issue bằng cách bổ sung khả năng xử lý theo lô cho các issue GitHub. Thay vì tập trung vào từng vấn đề riêng lẻ, các nhóm có thể quét hàng loạt nhiều issue đồng thời. Tính năng mới cho phép phân tích song song hoặc theo hàng đợi cả backlog, giúp xử lý khối lượng issue lớn hiệu quả hơn.

Ngoài ra, nâng cấp này sẽ hỗ trợ xác định các issue trùng lặp hoặc liên quan, giúp tổ chức tốt hơn. Các issue có thể được phân loại theo tiêu chí như độ phức tạp, hệ thống con, hoặc mức độ ảnh hưởng. 

Để tinh gọn hơn nữa, các nhóm có thể chạy công cụ GitHub hoặc Firecrawl ở chế độ theo lô, cải thiện hiệu suất đáng kể. Cuối cùng, nâng cấp này mang lại một lệnh duy nhất có thể tự động phân loại hàng chục, thậm chí hàng trăm issue cùng lúc.

3. Kiểm thử trước PR, kiểm tra an toàn và xác thực

Trước khi tạo pull request (PR), điều cần thiết là tác nhân phải xác thực kỹ lưỡng các thay đổi đề xuất thay vì chỉ tạo ra chúng. Quá trình xác thực nên bao gồm một số tính năng đã lên kế hoạch, như chạy unit test qua GitHub Actions hoặc trình chạy cục bộ để đảm bảo mã hoạt động như mong đợi. 

Xác thực phụ thuộc cũng rất quan trọng. Nó bao gồm kiểm tra import, phát hiện module thiếu, và xử lý không khớp phiên bản có thể ảnh hưởng đến dự án. Hơn nữa, linting, định dạng, và kiểm tra kiểu đóng vai trò then chốt trong việc duy trì chất lượng mã. 

Cần đảm bảo rằng các chỉnh sửa không làm gián đoạn pipeline build hiện có và phát hiện bất kỳ thay đổi phá vỡ API hoặc hồi quy nào. Bằng cách tuân thủ các bước này, chúng ta có thể đảm bảo PR sạch, an toàn và sẵn sàng triển khai sản xuất.

Tổng kết về GPT-5.1 Codex

Xây dựng các tác nhân nâng cao, nhiều bước với GPT-5.1 Codex và openai-agents đơn giản đến bất ngờ. Điều bạn thực sự cần làm là định nghĩa công cụ của riêng bạn và đưa cho mô hình hướng dẫn rõ ràng về thời điểm và cách sử dụng chúng. 

Trong hướng dẫn này, tôi dùng GitHub CLI vì nó nhanh, trực quan và dễ tích hợp, nhưng bạn cũng có thể dùng GitHub Python SDK, gọi API trực tiếp, hoặc bất kỳ CLI hay tiện ích Bash nào khác. Tính linh hoạt mới là sức mạnh thực sự ở đây.

Bạn có thể mở rộng thiết lập này tùy thích. 

Ví dụ, bạn có thể tạo:

• Một tác nhân lập kế hoạch (chúng ta đã xây dựng)
• Một tác nhân hành động áp dụng thay đổi mã dựa trên kế hoạch
• Một tác nhân kiểm thử chạy test và xác nhận liệu thay đổi có làm hỏng gì không
• Một tác nhân PR mở pull request với phần tóm tắt gọn gàng

Mục tiêu của hướng dẫn này là cho thấy GPT-5.1 Codex thực sự có thể làm gì: Gọi công cụ mượt mà, hiểu các codebase lớn, thực hiện suy luận có cấu trúc, và có thể chạy chuỗi tự động hóa dài mà không cần người dùng can thiệp liên tục. 

Nếu bạn muốn tìm hiểu thêm về cách tạo AI agent, tôi khuyến nghị xem khóa học AI Agents với Google ADK của chúng tôi, cũng như danh sách các dự án AI agent để xây dựng.