Devin 年化营收冲到 4.45 亿美元——自主编程智能体的爆发意味着什么

import os
import time
import requests

# 从环境变量读取 API Key，不要硬编码到代码里
DEVIN_API_KEY = os.environ.get("DEVIN_API_KEY")
DEVIN_BASE_URL = "https://api.devin.ai/v1"

headers = {
    "Authorization": f"Bearer {DEVIN_API_KEY}",
    "Content-Type": "application/json",
}

def create_session(task_prompt: str, repo_url: str = None) -> dict:
    """创建一个 Devin 自主编码会话"""
    payload = {
        "prompt": task_prompt,
        # 如果需要 Devin 在指定仓库里工作，传入 repo_url
    }
    if repo_url:
        payload["repo_url"] = repo_url

    resp = requests.post(
        f"{DEVIN_BASE_URL}/sessions",
        headers=headers,
        json=payload,
    )
    resp.raise_for_status()
    return resp.json()

def get_session_status(session_id: str) -> dict:
    """查询会话状态和输出"""
    resp = requests.get(
        f"{DEVIN_BASE_URL}/sessions/{session_id}",
        headers=headers,
    )
    resp.raise_for_status()
    return resp.json()

def run_task_and_wait(task_prompt: str, repo_url: str = None, timeout: int = 600) -> dict:
    """提交任务并等待完成，最长等待 timeout 秒"""
    session = create_session(task_prompt, repo_url)
    session_id = session["session_id"]
    print(f"会话已创建: {session_id}")

    elapsed = 0
    interval = 15  # 每 15 秒查询一次
    while elapsed < timeout:
        status = get_session_status(session_id)
        state = status.get("status_enum", status.get("status", "unknown"))
        print(f"  [{elapsed}s] 状态: {state}")

        if state in ("completed", "failed", "cancelled"):
            return status

        time.sleep(interval)
        elapsed += interval

    raise TimeoutError(f"任务在 {timeout} 秒内未完成")

# ---- 实际使用示例 ----
if __name__ == "__main__":
    # 示例 1：让 Devin 在指定仓库里添加一个新功能
    result = run_task_and_wait(
        task_prompt=(
            "在 src/api/ 目录下新增一个 health_check.py 模块，"
            "实现一个 /health GET 端点，返回 {'status': 'ok', 'timestamp': <当前UTC时间>}。"
            "同时补上对应的单元测试放在 tests/test_health_check.py。"
            "确保所有测试通过后再提交。"
        ),
        repo_url="https://github.com/your-org/your-repo",
        timeout=600,
    )
    print("任务结果:", result.get("output", result))