编程 OmX(oh-my-codex)深度解析:给 Codex 加上工作流引擎,让 AI 助手从「单打独斗」变「团队协作」

2026-05-13 23:44:56 +0800 CST views 9

OmX(oh-my-codex)深度解析:给 Codex 加上工作流引擎,让 AI 助手从「单打独斗」变「团队协作」

引言:AI 编程助手的「单兵作战」困境

如果你用过 Claude Code 或 OpenAI Codex CLI,一定有过这种感受:

# 你让 Codex 重构一个模块
$ codex "帮我重构 src/auth 模块,拆分为 token + session 两个子模块"

# Codex 开始工作...
# 它很努力,但它:
# 1. 没有「规划」阶段——直接开始写代码
# 2. 没有「记忆」——每次对话都是全新的
# 3. 没有「协作」——不能调用其他 Agent 帮忙
# 4. 没有「审查」——自己的代码自己测,没有第三方视角

# 结果:代码写完了,但架构混乱、测试不全、文档缺失

根本问题:现有 AI 编程助手是「单兵作战」模式——一个人(一个模型)做所有事。

2026 年 2 月,GitHub 用户 Yechan-Heo 开源了 OmX(oh-my-codex)——一个基于命令行的 AI 编程工作流引擎。短短两个月,Star 数飙升至 16000+

OmX 的核心定位:不是替代 Codex,而是给 Codex 加一层工作流引擎。

┌─────────────────────────────────────────────────┐
│          AI 编程助手的演进                          │
│                                                 │
│  v1.0: 纯代码补全(Copilot)                    │
│        ↓                                        │
│  v2.0: 对话式编程(Claude Code、Codex CLI)    │
│        ↓                                        │
│  v3.0: 工作流编排(OmX)← 我们现在在这里       │
│        ↓                                        │
│  v4.0: 自主团队(?)                          │
└─────────────────────────────────────────────────┘

本文将从架构、原理、代码实战三个维度,深度解析 OmX 的技术实现。


第一章:OmX 的核心架构——工作流引擎 + Agent 团队

1.1 传统 Codex CLI 的执行流程

用户输入 → Codex CLI → 调用 LLM → 生成代码 → 写入文件

问题:

  • 没有规划:直接生成代码,容易走入死胡同
  • 没有验证:生成的代码没有经过测试/审查
  • 没有记忆:每次会话独立,无法积累经验

1.2 OmX 的工作流架构

┌──────────────────────────────────────────────────────────┐
│                      OmX 工作流                         │
│                                                          │
│  用户输入                                                │
│      │                                                 │
│      ▼                                                 │
│  ┌──────────────┐                                      │
│  │ Planner Agent│  规划阶段:拆解任务、制定执行计划       │
│  └──────┬───────┘                                      │
│          │ 输出:任务列表 + 执行顺序                     │
│          ▼                                                 │
│  ┌──────────────┐                                      │
│  │Executor Agent │  执行阶段:调用 Codex CLI 写代码      │
│  │  (Codex CLI) │                                     │
│  └──────┬───────┘                                      │
│          │ 输出:代码文件                                 │
│          ▼                                                 │
│  ┌──────────────┐                                      │
│  │Reviewer Agent │  审查阶段:检查代码质量、安全性        │
│  └──────┬───────┘                                      │
│          │ 输出:审查报告 + 修改建议                     │
│          ▼                                                 │
│  ┌──────────────┐                                      │
│  │  HUD 实时显示 │  可视化:当前进度、Agent 状态         │
│  └──────────────┘                                      │
│                                                          │
│  支持:Hooks(钩子)、Checkpoints(检查点)、            │
│         Agent 间通信、共享记忆                            │
└──────────────────────────────────────────────────────────┘

关键创新:OmX 不是「另一个 AI 助手」,而是「AI 助手的编排层」。

Codex CLI = 引擎(执行代码生成)
OmX       = 方向盘 + 导航仪(规划路径、协调多个引擎)

1.3 OmX 的核心组件

组件功能类比
Planner任务拆解、执行计划生成项目经理
Executor调用 Codex CLI 执行代码生成开发工程师
Reviewer代码审查、安全检测代码审查员
HUD实时可视化(Heads-Up Display)监控大屏
Hooks生命周期钩子(before/after each step)Git Hooks
Checkpoints中间状态快照,支持回滚游戏存档

第二章:Planner Agent——从「直接写代码」到「先规划再执行」

2.1 传统方式的痛点

# 用户需求:「帮我实现一个 RPC 框架」
# 传统 Codex CLI 的处理方式:

# 1. 直接开始写代码(没有规划)
codex_terminal = """
用户:帮我实现一个 RPC 框架

(Codex 开始输出代码)
class RPCServer:
    def __init__(self):
        self.handlers = {}
    
    def register(self, name, handler):
        self.handlers[name] = handler
    
    def call(self, name, args):
        return self.handlers[name](*args)

# 问题:
# - 没有考虑序列化协议(JSON?Protobuf?MsgPack?)
# - 没有考虑传输层(TCP?HTTP?WebSocket?)
# - 没有考虑服务发现、负载均衡、熔断...
# - 写出来的是一个玩具,不是生产级框架
"""

# 2. 没有验证机制
# Codex 自己写代码,自己测试,自己审查 → 容易遗漏边界情况

2.2 OmX Planner 的工作方式

# OmX 的处理方式:先规划,再执行

# 第一阶段:Planning(规划)
planner_output = """
用户需求:帮我实现一个 RPC 框架

## 任务拆解:
1. 设计序列化协议(选择 Protobuf)
2. 设计传输层(基于 HTTP/2)
3. 实现服务端框架(注册、调用、中间件)
4. 实现客户端框架(连接池、服务发现、负载均衡)
5. 实现中间件(日志、重试、熔断、限流)
6. 编写单元测试(覆盖率 > 80%)
7. 编写集成测试(真实网络调用)
8. 编写文档(API 文档 + 使用示例)

## 执行顺序:
1 → 2 → 3 → 4 → 5 → 6 → 7 → 8

## 预估时间:
每个任务约 15-30 分钟,总计 2-4 小时

## 风险评估:
- 序列化协议选择:Protobuf 成熟但需要学习成本
- HTTP/2 实现:可以用 hyper(Rust)或 http2(Go)
- 中间件设计:需要支持链式调用(责任链模式)
"""

# 第二阶段:Execution(执行)
# OmX 按照规划,逐一调用 Executor Agent 执行每个任务
# 每个任务完成后,调用 Reviewer Agent 审查

# 第三阶段:Review(审查)
reviewer_output = """
任务 3(实现服务端框架)审查报告:

✅ 优点:
- 代码结构清晰,使用了插件模式
- 错误处理完善,所有 export function 都有 try-catch

⚠️ 建议修改:
- 注册中心应该用接口抽象,不要硬编码为 Consul
- 连接池应该用 sync.Pool,而不是每次新建
- 需要增加 TLS 支持(当前是明文传输)

❌ 必须修改:
- 没有做内存泄漏检测(长期运行的 RPC 服务必须考虑)
- 没有做优雅关闭(会丢失正在处理的请求)
"""

2.3 Planner 的实现原理

class PlannerAgent:
    """OmX 的规划 Agent"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
        self.planning_prompt = self.load_planning_prompt()
        
    def plan(self, user_request: str) -> dict:
        """
        将用户需求转化为执行计划
        Args:
            user_request: 用户需求(自然语言)
        Returns:
            plan: 执行计划(任务列表 + 执行顺序 + 风险评估)
        """
        # 1. 调用 LLM 生成规划
        prompt = self.planning_prompt.format(user_request=user_request)
        response = self.llm.complete(prompt)
        
        # 2. 解析 LLM 输出(JSON 格式)
        plan = self.parse_plan(response)
        
        # 3. 验证计划的合理性
        plan = self.validate_plan(plan)
        
        return plan
    
    def load_planning_prompt(self) -> str:
        """加载规划 Prompt(系统内置)"""
        return """
你是 一个 资深 软件架构师。

用户会给你一个开发需求,你需要:
1. 将需求拆解为具体的任务列表
2. 确定任务的执行顺序(依赖关系)
3. 评估每个任务的风险和预估时间
4. 输出 JSON 格式的执行计划

输出格式:
{
  "tasks": [
    {"id": 1, "name": "设计序列化协议", "deps": [], "estimated_time": "30m"},
    {"id": 2, "name": "实现传输层", "deps": [1], "estimated_time": "1h"},
    ...
  ],
  "risks": ["序列化兼容性", "性能瓶颈"],
  "suggestions": ["使用 Protobuf", "考虑 HTTP/2"]
}

规则:
- 每个任务必须明确、可执行
- 必须考虑测试、文档、部署
- 必须评估技术风险和备选方案
"""
    
    def parse_plan(self, llm_response: str) -> dict:
        """解析 LLM 输出的 JSON"""
        import json
        # 提取 JSON(LLM 可能输出 markdown 代码块)
        json_str = self.extract_json(llm_response)
        plan = json.loads(json_str)
        return plan
    
    def validate_plan(self, plan: dict) -> dict:
        """验证计划的合理性"""
        tasks = plan["tasks"]
        
        # 检查依赖关系是否形成环
        if self.has_cycle(tasks):
            raise ValueError("任务依赖存在环!")
        
        # 检查是否遗漏了测试和文档
        task_names = [t["name"] for t in tasks]
        if not any("测试" in name or "test" in name.lower() for name in task_names):
            # 自动添加测试任务
            tasks.append({
                "id": len(tasks) + 1,
                "name": "编写单元测试",
                "deps": [t["id"] for t in tasks],
                "estimated_time": "1h"
            })
        
        return {"tasks": tasks, "risks": plan["risks"], "suggestions": plan["suggestions"]}

第三章:Executor Agent——调用 Codex CLI 执行代码生成

3.1 Executor 的工作流程

class ExecutorAgent:
    """OmX 的执行 Agent(调用 Codex CLI)"""
    
    def __init__(self, codex_cli_path: str):
        self.codex_cli = codex_cli_path
        
    def execute_task(self, task: dict, context: dict) -> dict:
        """
        执行单个任务
        Args:
            task: 任务描述
            context: 上下文(之前任务的输出、共享记忆等)
        Returns:
            result: 执行结果(代码文件列表 + 执行日志)
        """
        # 1. 构建 Codex CLI 的 Prompt
        prompt = self.build_prompt(task, context)
        
        # 2. 调用 Codex CLI
        result = self.call_codex(prompt)
        
        # 3. 解析 Codex 输出(提取代码文件)
        files = self.extract_files(result)
        
        # 4. 写入文件系统
        self.write_files(files)
        
        return {"files": files, "log": result["log"]}
    
    def build_prompt(self, task: dict, context: dict) -> str:
        """构建 Codex CLI 的 Prompt"""
        return f"""
你是 一个 资深 开发者。

## 任务
{task['name']}

## 依赖的任务(已完成)
{self.format_context(context)}

## 要求
- 代码必须包含完整的单元测试
- 代码必须包含详细的注释
- 代码必须处理边界情况(错误处理、输入验证)
- 代码必须考虑性能(时间复杂度、空间复杂度)

## 输出格式
为每个文件输出:
```python
# filepath: src/xxx.py
代码内容...

"""

def call_codex(self, prompt: str) -> dict:
    """调用 Codex CLI"""
    import subprocess
    
    # 调用 Codex CLI(通过子进程)
    result = subprocess.run(
        [self.codex_cli, "--non-interactive", "--prompt", prompt],
        capture_output=True,
        text=True,
        timeout=300  # 5 分钟超时
    )
    
    if result.returncode != 0:
        raise Exception(f"Codex CLI 执行失败:{result.stderr}")
    
    return {"output": result.stdout, "log": result.stderr}

def extract_files(self, codex_result: dict) -> list:
    """从 Codex 输出中提取代码文件"""
    import re
    
    output = codex_result["output"]
    files = []
    
    # 匹配 ```python ... ``` 代码块
    pattern = r'```(\w+)\n# filepath: (.+?)\n(.*?)```'
    matches = re.findall(pattern, output, re.DOTALL)
    
    for lang, filepath, code in matches:
        files.append({
            "path": filepath.strip(),
            "language": lang,
            "content": code.strip()
        })
    
    return files

def write_files(self, files: list):
    """将代码写入文件系统"""
    import os
    
    for file in files:
        path = file["path"]
        content = file["content"]
        
        # 创建目录
        os.makedirs(os.path.dirname(path), exist_ok=True)
        
        # 写入文件
        with open(path, "w") as f:
            f.write(content)
        
        print(f"✅ 已写入:{path}")

### 3.2 Hooks——生命周期钩子

OmX 支持在任务执行的不同阶段插入自定义逻辑(类似 Git Hooks):

```python
# .omx/hooks.py
def before_task(task):
    """每个任务执行前触发"""
    print(f"🚀 开始执行任务:{task['name']}")
    
    # 检查依赖是否安装
    if task["name"] == "实现 RPC 框架":
        if not command_exists("protoc"):
            raise Exception("请先安装 Protobuf 编译器:brew install protobuf")

def after_task(task, result):
    """每个任务执行后触发"""
    print(f"✅ 任务完成:{task['name']}")
    
    # 自动运行测试
    if "test" in task["name"].lower():
        run_tests()

def before_review(task, code):
    """审查前触发"""
    # 自动运行 linter
    lint_result = run_linter(code)
    if lint_result["errors"]:
        print("⚠️ Linter 发现错误,请先修复")
        return False  # 阻止审查
    return True

def after_review(task, review_report):
    """审查后触发"""
    # 如果审查不通过,自动回滚
    if review_report["status"] == "rejected":
        rollback_to_checkpoint(task["id"] - 1)

第四章:Reviewer Agent——代码审查的自动化

4.1 传统代码审查的痛点

传统流程:
1. 开发者写代码
2. 提交 PR
3. 等待同事审查(可能等 1-2 天)
4. 审查反馈 → 修改 → 再审查(可能反复多轮)
5. 合并代码

问题:
- 审查周期长(等待同事有空)
- 审查质量参差不齐(同事可能不熟悉这个模块)
- 容易遗漏安全问题(SQL 注入、XSS、依赖漏洞)

4.2 OmX Reviewer Agent 的工作方式

class ReviewerAgent:
    """OmX 的审查 Agent"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
        self.review_prompt = self.load_review_prompt()
        
    def review(self, task: dict, code_files: list) -> dict:
        """
        审查代码
        Args:
            task: 任务描述
            code_files: 代码文件列表
        Returns:
            report: 审查报告(通过/拒绝 + 修改建议 + 风险评估)
        """
        # 1. 静态分析(AST 解析)
        static_issues = self.static_analysis(code_files)
        
        # 2. 调用 LLM 进行语义审查
        semantic_issues = self.semantic_review(task, code_files)
        
        # 3. 安全检查(调用专业安全工具)
        security_issues = self.security_check(code_files)
        
        # 4. 生成审查报告
        report = self.generate_report(static_issues, semantic_issues, security_issues)
        
        return report
    
    def static_analysis(self, code_files: list) -> list:
        """静态分析(AST 解析)"""
        issues = []
        
        for file in code_files:
            if file["language"] == "python":
                # 使用 ast 模块解析 Python 代码
                import ast
                tree = ast.parse(file["content"])
                
                # 检查:是否有 bare except
                for node in ast.walk(tree):
                    if isinstance(node, ast.ExceptHandler):
                        if node.type is None:  # bare except
                            issues.append({
                                "file": file["path"],
                                "line": node.lineno,
                                "type": "bare-except",
                                "severity": "warning",
                                "message": "避免使用 bare except,会捕获所有异常(包括 KeyboardInterrupt)"
                            })
        
        return issues
    
    def semantic_review(self, task: dict, code_files: list) -> list:
        """语义审查(调用 LLM)"""
        prompt = self.review_prompt.format(
            task=task,
            code=self.format_code(code_files)
        )
        response = self.llm.complete(prompt)
        issues = self.parse_review(response)
        return issues
    
    def security_check(self, code_files: list) -> list:
        """安全检查(调用 bandit、semgrep 等工具)"""
        issues = []
        
        for file in code_files:
            if file["language"] == "python":
                # 调用 bandit(Python 安全扫描工具)
                import subprocess
                result = subprocess.run(
                    ["bandit", "-f", "json", file["path"]],
                    capture_output=True,
                    text=True
                )
                bandit_issues = json.loads(result.stdout)
                issues.extend(bandit_issues["results"])
        
        return issues
    
    def generate_report(self, static_issues, semantic_issues, security_issues) -> dict:
        """生成审查报告"""
        all_issues = static_issues + semantic_issues + security_issues
        
        # 按严重程度排序
        severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
        all_issues.sort(key=lambda x: severity_order.get(x["severity"], 99))
        
        # 判断通过/拒绝
        critical_issues = [i for i in all_issues if i["severity"] == "critical"]
        status = "rejected" if critical_issues else "approved"
        
        return {
            "status": status,
            "issues": all_issues,
            "summary": f"发现 {len(all_issues)} 个问题({len(critical_issues)} 个严重)"
        }

4.3 审查报告示例

# 审查报告:任务 3(实现 RPC 服务端框架)

## 状态:⚠️ 需要修改

## 严重问题(必须修改)
1. **[critical] 内存泄漏** - `src/server.py:45`
   - 问题:创建了 goroutine 但没有 defer cancel(),可能导致上下文泄漏
   - 建议:在每个 goroutine 开头添加 `defer cancel()`

2. **[critical] SQL 注入风险** - `src/auth.py:78`
   - 问题:直接拼接 SQL 字符串 `query = f"SELECT * FROM users WHERE name = '{username}'"`
   - 建议:使用参数化查询 `cursor.execute("SELECT * FROM users WHERE name = ?", (username,))`

## 警告问题(建议修改)
1. **[high] 缺少 TLS 支持** - `src/server.py`
   - 问题:RPC 服务是明文传输,容易被中间人攻击
   - 建议:添加 TLS 支持(可以用 Let's Encrypt 自动证书)

2. **[medium] 没有优雅关闭** - `src/server.py:120`
   - 问题:直接 `os.exit()` 会丢失正在处理的请求
   - 建议:使用 `signal.Notify` 捕获 SIGTERM,等待现有请求完成后再退出

## 低级问题(可选修改)
1. **[low] 日志格式不统一** - 多个文件
   - 建议:使用统一的日志库(如 logrus、zap)

## 总结
- 严重问题:2 个(必须修改后才能合并)
- 警告问题:2 个(建议修改)
- 低级问题:1 个(可选)
- 预估修改时间:1-2 小时

第五章:HUD(Heads-Up Display)——实时可视化工作流

5.1 传统 CLI 工具的问题

传统 CLI 输出:
$ codex "帮我重构代码"
正在生成...
(等待 30 秒,没有任何反馈)
(突然输出一大堆代码)

问题:
- 不知道当前进度(0%?50%?90%?)
- 不知道卡在哪一步(网络慢?LLM 慢?代码生成慢?)
- 不知道是否可以取消/暂停/恢复

5.2 OmX HUD 的解决方案

class HUD:
    """Heads-Up Display:实时可视化工作流"""
    
    def __init__(self):
        self.layout = self.create_layout()
        
    def create_layout(self):
        """创建终端 UI 布局"""
        from rich.layout import Layout
        from rich.panel import Panel
        from rich.table import Table
        
        layout = Layout()
        
        # 上半部分:任务列表
        layout.split_column(
            Layout(name="header", size=3),
            Layout(name="main"),
            Layout(name="footer", size=3)
        )
        
        # 主区域:分为左右两栏
        layout["main"].split_row(
            Layout(name="task-list", ratio=2),
            Layout(name="agent-status", ratio=1)
        )
        
        return layout
    
    def update(self, plan: dict, current_task: int, agent_status: dict):
        """更新 HUD 显示"""
        from rich.table import Table
        
        # 任务列表
        table = Table(title="任务列表")
        table.add_column("ID", style="cyan")
        table.add_column("任务名", style="white")
        table.add_column("状态", style="green")
        table.add_column("进度", style="yellow")
        
        for task in plan["tasks"]:
            status = self.get_task_status(task, current_task)
            progress = self.get_task_progress(task)
            table.add_row(
                str(task["id"]),
                task["name"],
                status,
                progress
            )
        
        self.layout["task-list"].update(Panel(table))
        
        # Agent 状态
        agent_table = Table(title="Agent 状态")
        agent_table.add_column("Agent", style="cyan")
        agent_table.add_column("状态", style="green")
        agent_table.add_column("当前任务", style="white")
        
        for agent, status in agent_status.items():
            agent_table.add_row(agent, status["status"], status["current_task"])
        
        self.layout["agent-status"].update(Panel(agent_table))
        
        # 刷新终端
        self.refresh()

HUD 的实际效果:

┌─────────────────────────────────────────────────────────┐
│  OmX - RPC 框架开发项目                                │
│  总体进度:████████████████░░░░ 65%                     │
├────────────────────────────────┬────────────────────────┤
│ 任务列表                      │ Agent 状态              │
│                              │                        │
│ ✅ 1. 设计序列化协议       │  Planner: 空闲         │
│ ✅ 2. 实现传输层           │  Executor: 执行中       │
│ 🔄 3. 实现服务端框架       │  Reviewer: 等待中      │
│ ⏳ 4. 实现客户端框架       │  HUD: 运行中           │
│ ⏳ 5. 实现中间件           │                        │
│ ⏳ 6. 编写单元测试         │                        │
│ ⏳ 7. 编写集成测试         │                        │
│ ⏳ 8. 编写文档             │                        │
├────────────────────────────────┴────────────────────────┤
│  当前任务:实现服务端框架(预计剩余 15 分钟)           │
│  提示:按 'p' 暂停,按 'r' 恢复,按 'q' 退出        │
└─────────────────────────────────────────────────────────┘

第六章:Checkpoints——中间状态快照与回滚

6.1 传统开发的问题

# 场景:你让 AI 重构代码,它改了 50 个文件
# 运行后发现:改错了,想回到重构前的状态

# 传统方式:
# 1. 如果你用了 Git → git reset --hard HEAD~10
# 2. 如果没用 Git → 手动改回来(噩梦)

# 问题:
# - Git 的 commit 是手动的,AI 生成代码的过程没有细粒度快照
# - 一旦改错,很难精确定位到「哪个步骤」开始出错

6.2 OmX Checkpoints 的解决方案

class CheckpointManager:
    """检查点管理:支持回滚到任意步骤"""
    
    def __init__(self, workspace: str):
        self.workspace = workspace
        self.checkpoints_dir = os.path.join(workspace, ".omx", "checkpoints")
        os.makedirs(self.checkpoints_dir, exist_ok=True)
        
    def create_checkpoint(self, task_id: int, description: str):
        """创建检查点"""
        import shutil
        
        # 1. 创建检查点目录
        checkpoint_dir = os.path.join(self.checkpoints_dir, f"task_{task_id}")
        os.makedirs(checkpoint_dir, exist_ok=True)
        
        # 2. 快照当前工作区(复制所有文件)
        for root, dirs, files in os.walk(self.workspace):
            # 跳过 .git 和 .omx 目录
            if ".git" in root or ".omx" in root:
                continue
            
            for file in files:
                src_path = os.path.join(root, file)
                rel_path = os.path.relpath(src_path, self.workspace)
                dst_path = os.path.join(checkpoint_dir, rel_path)
                
                os.makedirs(os.path.dirname(dst_path), exist_ok=True)
                shutil.copy2(src_path, dst_path)
        
        # 3. 记录检查点元数据
        metadata = {
            "task_id": task_id,
            "description": description,
            "timestamp": time.time(),
            "files_count": len(list(os.walk(checkpoint_dir)))
        }
        
        with open(os.path.join(checkpoint_dir, "metadata.json"), "w") as f:
            json.dump(metadata, f, indent=2)
        
        print(f"✅ 检查点已创建:task_{task_id}")
    
    def rollback(self, task_id: int):
        """回滚到指定检查点"""
        import shutil
        
        checkpoint_dir = os.path.join(self.checkpoints_dir, f"task_{task_id}")
        if not os.path.exists(checkpoint_dir):
            raise Exception(f"检查点 task_{task_id} 不存在")
        
        # 1. 删除当前工作区的文件(保留 .git 和 .omx)
        for root, dirs, files in os.walk(self.workspace):
            if ".git" in root or ".omx" in root:
                continue
            for file in files:
                os.remove(os.path.join(root, file))
        
        # 2. 从检查点恢复文件
        for root, dirs, files in os.walk(checkpoint_dir):
            for file in files:
                if file == "metadata.json":
                    continue
                src_path = os.path.join(root, file)
                rel_path = os.path.relpath(src_path, checkpoint_dir)
                dst_path = os.path.join(self.workspace, rel_path)
                
                os.makedirs(os.path.dirname(dst_path), exist_ok=True)
                shutil.copy2(src_path, dst_path)
        
        print(f"✅ 已回滚到检查点:task_{task_id}")
    
    def list_checkpoints(self):
        """列出所有检查点"""
        checkpoints = []
        for dir_name in os.listdir(self.checkpoints_dir):
            if dir_name.startswith("task_"):
                metadata_path = os.path.join(self.checkpoints_dir, dir_name, "metadata.json")
                with open(metadata_path) as f:
                    metadata = json.load(f)
                    checkpoints.append(metadata)
        
        return sorted(checkpoints, key=lambda x: x["task_id"])

6.3 Checkpoints 的实际使用

# 场景:让 OmX 开发 RPC 框架,执行到任务 5 时发现任务 3 有 bug

# 查看检查点列表
$ omx checkpoint list
ID  描述                  时间戳                文件数
1   完成序列化协议设计     2026-05-13 10:00:00   12
2   完成传输层实现          2026-05-13 10:30:00   45
3   完成服务端框架          2026-05-13 11:00:00   78
4   完成客户端框架          2026-05-13 11:30:00   120

# 回滚到任务 3(重新实现服务端框架)
$ omx checkpoint rollback 3
✅ 已回滚到检查点:task_3

# 修改任务 3 的要求,重新执行
$ omx resume --from-task 3
▶ 重新执行任务 3:实现服务端框架
(这次会避免之前的 bug)

第七章:代码实战——用 OmX 开发一个完整的 REST API

7.1 安装 OmX

# 安装 OmX
npm install -g oh-my-codex

# 或者从源码安装
git clone https://github.com/Yechan-Heo/oh-my-codex.git
cd oh-my-codex
npm install
npm run build
npm link

# 验证安装
omx --version

7.2 初始化项目

# 创建项目目录
mkdir my-rest-api && cd my-rest-api

# 初始化 OmX 项目
omx init

# 生成的目录结构:
# .omx/
#   config.yaml      # OmX 配置文件
#   hooks.py        # 生命周期钩子
#   checkpoints/    # 检查点目录
#   memory/         # 共享记忆目录

7.3 配置文件详解

# .omx/config.yaml
project:
  name: "my-rest-api"
  description: "一个完整的 REST API 项目"
  language: "python"
  framework: "fastapi"

agents:
  planner:
    model: "gpt-4o"          # 规划用 GPT-4o(推理能力强)
    temperature: 0.2
    max_tokens: 4096
  
  executor:
    model: "codex"            # 执行用 Codex CLI
    cli_path: "/usr/local/bin/codex"
    timeout: 300              # 5 分钟超时
  
  reviewer:
    model: "gpt-4o"          # 审查用 GPT-4o
    check_security: true      # 启用安全检查
    check_performance: true   # 启用性能检查

hud:
  enabled: true              # 启用 HUD 实时可视化
  refresh_rate: 1000         # 每秒刷新 1 次

checkpoints:
  enabled: true              # 启用检查点
  auto_create: true          # 每个任务完成后自动创建检查点
  max_checkpoints: 10        # 最多保留 10 个检查点

memory:
  enabled: true              # 启用共享记忆
  persist: true              # 持久化到磁盘

7.4 启动 OmX 工作流

# 启动 OmX(交互模式)
$ omx start "帮我开发一个完整的 REST API,包含用户注册、登录、CRUD 操作,使用 FastAPI + PostgreSQL"

▶ OmX 工作流启动...
▶ 规划中...
✅ 规划完成,共 8 个任务:

1. 设计数据库模型(User 表)
2. 实现用户注册接口(POST /register)
3. 实现用户登录接口(POST /login,JWT)
4. 实现 CRUD 接口(GET/POST/PUT/DELETE /items)
5. 实现权限控制(只有本人可以修改自己的 items)
6. 编写单元测试(pytest)
7. 编写 API 文档(Swagger)
8. 部署到 Docker + Kubernetes

▶ 开始执行任务 1/8...
(HUD 实时显示进度)

✅ 任务 1 完成
▶ 审查中...
✅ 审查通过

▶ 开始执行任务 2/8...
...

7.5 查看共享记忆

# OmX 会自动记录开发过程中的经验教训
$ cat .omx/memory/memory.json

{
  "tasks": [
    {
      "id": 1,
      "name": "设计数据库模型",
      "lessons_learned": [
        "User 表必须包含 created_at 和 updated_at 字段(审计需求)",
        "密码必须用 bcrypt 哈希存储,不能明文或 MD5",
        "需要添加 unique 约束防止重复注册"
      ]
    },
    {
      "id": 2,
      "name": "实现用户注册接口",
      "lessons_learned": [
        "必须做输入验证(email 格式、密码强度)",
        "必须防止 SQL 注入(用 ORM 或参数化查询)",
        "必须做速率限制(防止恶意注册)"
      ]
    }
  ],
  
  "security_tips": [
    "JWT secret 必须足够复杂(>32 字符随机字符串)",
    "JWT 过期时间不能太长(建议 < 24 小时)",
    "必须使用 HTTPS(生产环境)"
  ],
  
  "performance_tips": [
    "数据库连接必须用连接池(SQLAlchemy 默认支持)",
    "热数据可以用 Redis 缓存",
    "慢查询必须加索引"
  ]
}

第八章:与竞品对比

8.1 OmX vs 其他 AI 编程助手

维度OmXClaude CodeCursorGitHub Copilot
工作流编排✅ 原生支持❌ 不支持❌ 不支持❌ 不支持
多 Agent 协作✅ Planner + Executor + Reviewer
代码审查自动化✅ Reviewer Agent❌(需 GitHub PR)
检查点/回滚✅ 原生支持
实时可视化✅ HUD
共享记忆✅ 跨任务记忆
学习曲线⭐⭐⭐(需要理解工作流概念)⭐ 最简单⭐⭐⭐ 最简单
适合场景复杂项目、多模块系统简单脚本、单文件修改代码补全、重构建议代码补全

8.2 选型建议

用 OmX 的场景:

  • ✅ 开发复杂系统(> 5 个模块)
  • ✅ 需要代码审查和安全检查
  • ✅ 需要回滚能力(避免 AI 改错代码)
  • ✅ 团队协作(共享记忆、审查报告)

用 Claude Code / Cursor 的场景:

  • 简单脚本(< 200 行)
  • 单文件修改
  • 快速原型开发

总结:AI 编程助手的「团队化」时代

OmX 的出现,标志着 AI 编程助手从「单兵作战」进入「团队协作」时代:

1. 工作流编排:从「直接写代码」到「先规划再执行」
Planner Agent 的引入,让 AI 助手学会了「思考」——不再是被动响应用户请求,而是主动规划执行路径。

2. 多 Agent 协作:让专业的人做专业的事
Planner(规划)、Executor(执行)、Reviewer(审查)各司其职,比「一个 Agent 做所有事」质量更高。

3. 检查点机制:让 AI 生成代码变得「可撤销」
这是传统 AI 助手严重缺失的能力——代码生成失败后的回滚成本极高,OmX 通过检查点解决了这个问题。

4. 共享记忆:让 AI 助手「积累经验」
传统 AI 助手是「金鱼记忆」——7 秒后就忘了。OmX 的共享记忆让 AI 能够跨任务积累经验,越用越聪明。

适用场景推荐:

  • ✅ 复杂后端系统开发(REST API、微服务、消息队列)
  • ✅ 需要高代码质量的场景(金融、医疗、基础设施)
  • ✅ 团队协作开发(共享记忆、审查报告)
  • ❌ 简单脚本开发(用 Cursor 更快)
  • ❌ 一次性原型(用 Claude Code 更快)

参考资源

  1. OmX GitHub 仓库:https://github.com/Yechan-Heo/oh-my-codex
  2. OmX 文档:https://oh-my-codex.dev/docs
  3. OpenAI Codex CLI 文档:https://platform.openai.com/docs/guides/codex
  4. Anthropic Claude Code 文档:https://docs.anthropic.com/claude-code
  5. FastAPI 官方文档:https://fastapi.tiangolo.com/

文章字数统计:约 19, 000 字

推荐文章

Go中使用依赖注入的实用技巧
2024-11-19 00:24:20 +0800 CST
Vue3中哪些API被废弃了?
2024-11-17 04:17:22 +0800 CST
前端代码规范 - Commit 提交规范
2024-11-18 10:18:08 +0800 CST
# 解决 MySQL 经常断开重连的问题
2024-11-19 04:50:20 +0800 CST
一些高质量的Mac软件资源网站
2024-11-19 08:16:01 +0800 CST
黑客帝国代码雨效果
2024-11-19 01:49:31 +0800 CST
程序员茄子在线接单