Codex 多场景工程实践：从代码生成到自动化流水线的统一抽象

from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Awaitable
import uuid

class Scene(str, Enum):
    code_review = "code_review"
    test_gen = "test_gen"
    doc_gen = "doc_gen"
    data_clean = "data_clean"
    ops_script = "ops_script"
    ci_fix = "ci_fix"

@dataclass
class CodexTask:
    scene: Scene
    payload: dict[str, Any]
    constraints: dict[str, Any] = field(default_factory=dict)
    task_id: str = field(default_factory=lambda: uuid.uuid4().hex[:8])
    result: dict[str, Any] | None = None
    status: str = "pending"

import os, json
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)

REVIEW_SYSTEM = """你是代码审查员。只输出 JSON：
{"issues": [{"file": str, "line": int, "severity": "high|mid|low",
"type": "bug|security|style|perf", "msg": str}]}
只报告 diff 中可见的问题，不推测未展示代码。无问题返回空数组。"""

async def review_diff(diff: str) -> dict:
    resp = await client.chat.completions.create(
        model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
        messages=[
            {"role": "system", "content": REVIEW_SYSTEM},
            {"role": "user", "content": diff[:12000]},
        ],
        response_format={"type": "json_object"},
        temperature=0,
    )
    return json.loads(resp.choices[0].message.content)

TEST_SYSTEM = """你是测试工程师。为给定函数生成 pytest 测试，
覆盖：正常路径、边界值、异常分支、空输入。
只输出 Python 代码，不输出解释。不要 mock 掉被测函数本身。"""

async def gen_tests(source: str) -> str:
    resp = await client.chat.completions.create(
        model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
        messages=[
            {"role": "system", "content": TEST_SYSTEM},
            {"role": "user", "content": source},
        ],
        temperature=0.2,
    )
    return resp.choices[0].message.content

CLEAN_SYSTEM = """你是数据工程师。根据表头和样例生成 Python 清洗函数。
要求：显式处理缺失值、类型转换失败、异常值；不修改原始文件；
不打印敏感字段。只输出函数代码。"""

async def gen_cleaner(schema: str) -> str:
    resp = await client.chat.completions.create(
        model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
        messages=[
            {"role": "system", "content": CLEAN_SYSTEM},
            {"role": "user", "content": schema},
        ],
        temperature=0.1,
    )
    return resp.choices[0].message.content

OPS_SYSTEM = """你是 SRE。根据错误日志给出最小修复建议。
只输出 JSON：{"cause": str, "fix": str, "commands": [str], "risk": "low|mid|high"}
禁止输出 rm -rf、curl|sh、关闭防火墙、提权等危险命令。"""

async def diagnose(log: str) -> dict:
    resp = await client.chat.completions.create(
        model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
        messages=[
            {"role": "system", "content": OPS_SYSTEM},
            {"role": "user", "content": log[-8000:]},
        ],
        response_format={"type": "json_object"},
        temperature=0,
    )
    return json.loads(resp.choices[0].message.content)

DANGEROUS = ("rm -rf", "curl | sh", "chmod 777", "sudo", "iptables -F")

def audit_commands(cmds: list[str]) -> list[str]:
    return [c for c in cmds if any(d in c for d in DANGEROUS)]

import asyncio, logging

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("codex")

HANDLERS: dict[Scene, Callable[[CodexTask], Awaitable[dict]]] = {}

def register(scene: Scene):
    def deco(fn):
        HANDLERS[scene] = fn
        return fn
    return deco

async def dispatch(task: CodexTask) -> CodexTask:
    handler = HANDLERS.get(task.scene)
    if not handler:
        task.status = "failed"
        return task
    try:
        task.result = await handler(task)
        task.status = "succeeded"
    except Exception as e:
        task.status = "failed"
        task.result = {"error": str(e)}
    log.info("scene=%s id=%s status=%s", task.scene, task.task_id, task.status)
    return task

async def run_all(tasks: list[CodexTask], concurrency: int = 3):
    sem = asyncio.Semaphore(concurrency)
    async def wrap(t):
        async with sem:
            return await dispatch(t)
    return await asyncio.gather(*(wrap(t) for t in tasks))