Warp 的 AI 编码代理新布局：GPT-5.5 如何打通本地、云端与开源工作流

"""
最小多代理协调原型：模拟 Warp 式的本地/云端/开源三层代理分工
运行前请设置: export OPENAI_API_KEY="sk-..."
依赖: pip install openai pydantic
"""

import json
from openai import OpenAI
from pydantic import BaseModel

client = OpenAI()

# ---- 定义代理角色与任务类型 ----

class SubTask(BaseModel):
    task_id: str
    description: str
    agent_role: str       # "local" | "cloud" | "opensource"
    status: str           # "pending" | "done"
    result: str | None    = None

class TaskPlan(BaseModel):
    original_intent: str
    subtasks: list[SubTask]

# ---- Step 1: 意图分解 ----

def decompose_intent(intent: str) -> TaskPlan:
    """用 GPT 模型把开发意图拆成子任务，并标注执行角色"""
    resp = client.chat.completions.create(
        model="gpt-4o",  # 实际可用时替换为 gpt-5.5 系列模型
        response_format={"type": "json_object"},
        messages=[
            {
                "role": "system",
                "content": (
                    "你是一个编码任务调度器。把用户的开发意图拆成 2-5 个子任务，"
                    "每个子任务标注应该由哪种代理执行：\n"
                    "- local: 需要读本地文件、改本地代码\n"
                    "- cloud: 需要跑测试、构建、长耗时计算\n"
                    "- opensource: 需要查上游仓库、处理 PR/issue\n"
                    "输出 JSON，包含 original_intent 和 subtasks 数组，"
                    "每个 subtask 有 task_id, description, agent_role, status(pending), result(null)"
                ),
            },
            {"role": "user", "content": intent},
        ],
    )
    data = json.loads(resp.choices[0].message.content)
    return TaskPlan(**data)

# ---- Step 2: 各代理执行（模拟） ----

AGENT_PROMPTS = {
    "local": "你是本地编码代理。直接给出代码修改建议，以 diff 格式输出。",
    "cloud": "你是云端构建代理。给出测试命令和预期结果摘要。",
    "opensource": "你是开源协作代理。给出 PR 描述模板和 review checklist。",
}

def execute_subtask(subtask: SubTask) -> str:
    """让对应角色代理执行单个子任务"""
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": AGENT_PROMPTS[subtask.agent_role]},
            {"role": "user", "content": subtask.description},
        ],
    )
    return resp.choices[0].message.content

# ---- Step 3: 汇总结果 ----

def summarize_results(plan: TaskPlan) -> str:
    """把所有子任务结果汇总为可操作的执行清单"""
    results = "\n".join(
        f"[{t.agent_role}] {t.task_id}: {t.result}" for t in plan.subtasks
    )
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": "把以下子任务结果整理成开发者可直接执行的步骤清单，标注优先级。",
            },
            {"role": "user", "content": f"原始意图: {plan.original_intent}\n\n{results}"},
        ],
    )
    return resp.choices[0].message.content

# ---- 主流程 ----

if __name__ == "__main__":
    intent = "把一个 Flask 项目迁移到 FastAPI，保持所有路由语义不变，确保测试通过，并向上游开源仓库提交 PR"

    print("=== 意图分解 ===")
    plan = decompose_intent(intent)
    for t in plan.subtasks:
        print(f"  {t.agent_role:>10} | {t.task_id}: {t.description}")

    print("\n=== 代理执行 ===")
    for t in plan.subtasks:
        t.status = "done"
        t.result = execute_subtask(t)
        print(f"  [{t.agent_role}] {t.task_id} 完成")

    print("\n=== 汇总清单 ===")
    print(summarize_results(plan))

export OPENAI_API_KEY="sk-你的密钥"
pip install openai pydantic
python multi_agent_coordinator.py