GET /api/v1/tasks/{id}

端点

GET http://localhost:8080/api/v1/tasks/{id}

描述

获取指定任务的当前状态、结果和元数据。使用此端点检查任务进度或获取最终结果。

认证

必需：是在请求头中包含 API 密钥：

X-API-Key: sk_test_123456

请求

路径参数

参数	类型	必需	描述
`id`	string	是	任务 ID（也用作工作流 ID）

请求头

请求头	必需	描述
`X-API-Key`	是	认证密钥

响应

成功响应

状态：200 OK 响应头：

X-Workflow-ID：Temporal 工作流标识符（与任务 ID 相同）

响应体：

{
  "task_id": "string",
  "workflow_id": "string",
  "status": "string",
  "result": "string",
  "response": {},
  "metadata": {},
  "error": "string",
  "created_at": "timestamp",
  "updated_at": "timestamp",
  "query": "string",
  "session_id": "string",
  "context": {},
  "unified_response": {},
  "mode": "string",
  "model_used": "string",
  "provider": "string",
  "usage": {
    "total_tokens": 0,
    "input_tokens": 0,
    "output_tokens": 0,
    "estimated_cost": 0.0
  }
}

响应字段

字段	类型	描述
`task_id`	string	唯一任务标识符
`workflow_id`	string	工作流标识符（与 task_id 相同）
`status`	string	当前任务状态
`result`	string	原始 LLM 输出（纯文本或 JSON 字符串）
`response`	object	解析后的 JSON（仅当 `result` 可解析为 JSON 时出现）
`metadata`	object	任务元数据（引用、验证结果、统一响应、提取事实、model_breakdown 等）
`error`	string	错误消息（无错误时为空）
`created_at`	timestamp	任务创建时间
`updated_at`	timestamp	最后更新时间
`query`	string	原始任务查询
`session_id`	string	会话标识符
`context`	object	任务上下文（预算信息、日期、参数）
`unified_response`	object	结构化统一响应，包含解析结果、性能指标和元数据
`mode`	string	使用的执行模式
`model_used`	string	使用的主要模型（例如 `gpt-5-mini-2025-08-07`）
`provider`	string	提供商名称（例如 `openai`、`anthropic`）
`usage`	object	Token 使用量和成本：`{ total_tokens, input_tokens?, output_tokens?, estimated_cost? }`

状态值

TASK_STATUS_UNSPECIFIED - 状态未知
TASK_STATUS_QUEUED - 等待执行
TASK_STATUS_RUNNING - 正在执行
TASK_STATUS_COMPLETED - 成功完成
TASK_STATUS_FAILED - 执行失败
TASK_STATUS_PAUSED - 用户暂停或 HITL 审查中
TASK_STATUS_CANCELLED - 用户已取消
TASK_STATUS_TIMEOUT - 超出超时限制

执行模式值

EXECUTION_MODE_SIMPLE - 单个 LLM 调用，无工具
EXECUTION_MODE_STANDARD - 多步骤执行，包含工具
EXECUTION_MODE_COMPLEX - 高级推理模式

示例

检查任务状态

curl -X GET "http://localhost:8080/api/v1/tasks/task_01HQZX3Y9K8M2P4N5S7T9W2V" \
  -H "X-API-Key: sk_test_123456"

响应（已排队）：

{
  "task_id": "task_01HQZX3Y9K8M2P4N5S7T9W2V",
  "status": "TASK_STATUS_QUEUED",
  "response": null,
  "error": "",
  "created_at": "2025-10-22T10:30:00Z",
  "updated_at": "2025-10-22T10:30:00Z",
  "query": "法国的首都是什么？",
  "session_id": "user-123-session",
  "mode": "EXECUTION_MODE_SIMPLE"
}

响应（运行中）：

{
  "task_id": "task_01HQZX3Y9K8M2P4N5S7T9W2V",
  "status": "TASK_STATUS_RUNNING",
  "response": null,
  "error": "",
  "created_at": "2025-10-22T10:30:00Z",
  "updated_at": "2025-10-22T10:30:02Z",
  "query": "法国的首都是什么？",
  "session_id": "user-123-session",
  "mode": "EXECUTION_MODE_SIMPLE"
}

响应（已完成）：

{
  "task_id": "task_01HQZX3Y9K8M2P4N5S7T9W2V",
  "status": "TASK_STATUS_COMPLETED",
  "result": "法国的首都是巴黎。巴黎自 987 年以来一直是首都，位于法国的中北部。",
  "error": "",
  "created_at": "2025-10-22T10:30:00Z",
  "updated_at": "2025-10-22T10:30:05Z",
  "query": "法国的首都是什么？",
  "session_id": "user-123-session",
  "mode": "EXECUTION_MODE_SIMPLE",
  "model_used": "gpt-5-mini-2025-08-07",
  "provider": "openai",
  "usage": {
    "total_tokens": 300,
    "input_tokens": 200,
    "output_tokens": 100,
    "estimated_cost": 0.006
  }
}

响应（失败）：

{
  "task_id": "task_01HQZX3Y9K8M2P4N5S7T9W2V",
  "status": "TASK_STATUS_FAILED",
  "response": null,
  "error": "LLM 服务不可用：连接超时",
  "created_at": "2025-10-22T10:30:00Z",
  "updated_at": "2025-10-22T10:30:10Z",
  "query": "法国的首都是什么？",
  "session_id": "user-123-session",
  "mode": "EXECUTION_MODE_SIMPLE"
}

深度研究响应载荷

当任务以 force_research: true 提交时，完成的响应会包含带有结构化研究数据的额外元数据字段。

深度研究元数据字段

对于深度研究任务，metadata 对象包含：

字段	类型	描述
`unified_response`	object	包含所有元数据的统一响应
`citations`	array	研究中的结构化引用数据
`verification`	object	声明验证结果（如果设置了 `enable_verification: true`）
`extracted_facts`	array	结构化事实（如果设置了 `enable_fact_extraction: true`）
`fact_summary`	object	提取事实的摘要统计

示例：深度研究完成响应

{
  "task_id": "task-abc123-research",
  "status": "TASK_STATUS_COMPLETED",
  "result": "# 2025 年 AI 趋势\n\n2025 年的人工智能格局特征是...[1][2]\n\n## 来源\n[1] MIT Technology Review (https://...) - 2025-01-15\n[2] Nature AI (https://...) - 2025-01-10",
  "metadata": {
    "unified_response": {
      "task_id": "task-abc123-research",
      "session_id": "session-xyz",
      "status": "completed",
      "result": "2025 年的人工智能格局...",
      "metadata": {
        "model": "claude-sonnet-4-5-20250929",
        "execution_mode": "EXECUTION_MODE_COMPLEX",
        "complexity_score": 0.72,
        "agents_used": 3
      },
      "usage": {
        "input_tokens": 12100,
        "output_tokens": 3320,
        "total_tokens": 15420,
        "cost_usd": 0.0462
      },
      "performance": {
        "execution_time_ms": 45200
      },
      "stop_reason": "completed",
      "error": null,
      "timestamp": "2025-01-20T10:00:00Z"
    },
    "citations": [
      {
        "title": "AI Breakthrough Report 2025",
        "url": "https://www.technologyreview.com/2025/01/ai-report",
        "source": "MIT Technology Review",
        "credibility_score": 0.92,
        "quality_score": 0.88
      },
      {
        "title": "Nature AI Special Issue",
        "url": "https://www.nature.com/articles/ai-2025",
        "source": "Nature",
        "credibility_score": 0.95,
        "quality_score": 0.91
      }
    ],
    "verification": {
      "overall_confidence": 0.87,
      "total_claims": 12,
      "supported_claims": 10,
      "unsupported_claims": ["关于未来预测的声明"],
      "conflicts": [],
      "claim_details": [
        {
          "claim": "AI 模型在基准测试 X 上达到 95% 准确率",
          "supporting_citations": [1, 2],
          "conflicting_citations": [],
          "confidence": 0.92
        }
      ]
    }
  },
  "model_used": "claude-sonnet-4-5-20250929",
  "provider": "anthropic",
  "usage": {
    "total_tokens": 15420,
    "input_tokens": 12100,
    "output_tokens": 3320,
    "estimated_cost": 0.0462
  }
}

提取的事实（可选）

当请求上下文中设置了 enable_fact_extraction: true 时：

{
  "metadata": {
    "extracted_facts": [
      {
        "statement": "GPT-5 在 MMLU 基准测试上达到 95% 准确率",
        "confidence": 0.92,
        "source_citation": [1, 3],
        "category": "performance",
        "entity_mentions": ["GPT-5", "MMLU"],
        "temporal_marker": "2025",
        "is_quantitative": true
      }
    ],
    "fact_summary": {
      "total_facts": 24,
      "high_confidence": 18,
      "categorized_facts": {
        "performance": 8,
        "market": 6,
        "research": 10
      },
      "contradiction_count": 1
    }
  }
}

引用对象模式

字段	类型	描述
`url`	string	来源 URL
`title`	string	文章/页面标题
`source`	string	出版商/域名
`credibility_score`	float	来源可信度（0.0-1.0）
`quality_score`	float	内容质量（0.0-1.0）

在 metadata.citations 数组中的位置与正文中使用的 [n] 引用编号一一对应。

验证对象模式

字段	类型	描述
`overall_confidence`	float	总体验证置信度（0.0-1.0）
`total_claims`	integer	提取的事实声明数量
`supported_claims`	integer	有引用支持的声明数量
`unsupported_claims`	array	没有引用支持的声明文本列表
`conflicts`	array	检测到的冲突信息
`claim_details`	array	每个声明的验证详情

访问深度研究数据：metadata.citations 数组和 metadata.verification 对象仅对研究工作流（force_research: true）填充。对于简单任务，这些字段将不会出现在响应中。

错误响应

401 未授权

{
  "error": "Unauthorized"
}

404 未找到

{
  "error": "Task not found"
}

500 内部服务器错误

{
  "error": "Failed to get task status: database error"
}

代码示例

Python - 简单状态检查

import httpx

def get_task_status(task_id: str, api_key: str):
    """获取任务状态。"""
    response = httpx.get(
        f"http://localhost:8080/api/v1/tasks/{task_id}",
        headers={"X-API-Key": api_key}
    )

    if response.status_code == 404:
        return None

    return response.json()

status = get_task_status("task_abc123", "sk_test_123456")
if status:
    print(f"状态：{status['status']}")
    if status['status'] == 'TASK_STATUS_COMPLETED':
        print(f"结果：{status['result']}")

Python - 轮询直到完成

import httpx
import time

def wait_for_completion(task_id: str, api_key: str, timeout: int = 300):
    """轮询任务状态直到完成或超时。"""
    start_time = time.time()

    while True:
        response = httpx.get(
            f"http://localhost:8080/api/v1/tasks/{task_id}",
            headers={"X-API-Key": api_key}
        )

        status = response.json()
        current_status = status["status"]

        # 检查终止状态
        if current_status == "TASK_STATUS_COMPLETED":
            return status.get("result", "")
        elif current_status == "TASK_STATUS_FAILED":
            raise Exception(f"任务失败：{status['error']}")
        elif current_status == "TASK_STATUS_TIMEOUT":
            raise Exception("任务超时")
        elif current_status == "TASK_STATUS_CANCELLED":
            raise Exception("任务已被取消")

        # 检查轮询超时
        if time.time() - start_time > timeout:
            raise TimeoutError(f"轮询超时，已等待 {timeout}s")

        # 等待后再轮询
        time.sleep(2)

# 用法
try:
    result = wait_for_completion("task_abc123", "sk_test_123456")
    print("结果：", result)
except Exception as e:
    print("错误：", e)

JavaScript/Node.js

const axios = require('axios');

async function getTaskStatus(taskId) {
  try {
    const response = await axios.get(
      `http://localhost:8080/api/v1/tasks/${taskId}`,
      {
        headers: {
          'X-API-Key': 'sk_test_123456'
        }
      }
    );

    const status = response.data;

    console.log('状态：', status.status);

    if (status.status === 'TASK_STATUS_COMPLETED') {
      console.log('结果：', status.result);
    } else if (status.status === 'TASK_STATUS_FAILED') {
      console.error('错误：', status.error);
    }

    return status;
  } catch (error) {
    if (error.response?.status === 404) {
      console.error('任务未找到');
    } else {
      console.error('错误：', error.response?.data || error.message);
    }
    throw error;
  }
}

getTaskStatus('task_abc123');

JavaScript - 使用 Async/Await 轮询

async function waitForCompletion(taskId, timeout = 300000) {
  const startTime = Date.now();
  const pollInterval = 2000; // 2 秒

  while (true) {
    const response = await axios.get(
      `http://localhost:8080/api/v1/tasks/${taskId}`,
      { headers: { 'X-API-Key': 'sk_test_123456' } }
    );

    const { status, result, error } = response.data;

    if (status === 'TASK_STATUS_COMPLETED') {
      return result;
    } else if (status === 'TASK_STATUS_FAILED') {
      throw new Error(`任务失败：${error}`);
    } else if (status === 'TASK_STATUS_TIMEOUT') {
      throw new Error('任务超时');
    }

    // 检查超时
    if (Date.now() - startTime > timeout) {
      throw new Error(`轮询超时，已等待 ${timeout}ms`);
    }

    // 等待后再轮询
    await new Promise(resolve => setTimeout(resolve, pollInterval));
  }
}

// 用法
waitForCompletion('task_abc123')
  .then(result => console.log('结果：', result))
  .catch(error => console.error('错误：', error.message));

Go

package main

import (
    "encoding/json"
    "fmt"
    "net/http"
    "time"
)

type TaskStatusResponse struct {
    TaskID          string                 `json:"task_id"`
    WorkflowID      string                 `json:"workflow_id"`
    Status          string                 `json:"status"`
    Result          string                 `json:"result"`
    Response        map[string]interface{} `json:"response"`
    Error           string                 `json:"error"`
    Query           string                 `json:"query"`
    SessionID       string                 `json:"session_id"`
    Mode            string                 `json:"mode"`
    Context         map[string]interface{} `json:"context"`
    ModelUsed       string                 `json:"model_used"`
    Provider        string                 `json:"provider"`
    CreatedAt       string                 `json:"created_at"`
    UpdatedAt       string                 `json:"updated_at"`
    Usage           map[string]interface{} `json:"usage"`
    Metadata        map[string]interface{} `json:"metadata"`
    UnifiedResponse map[string]interface{} `json:"unified_response"`
}

func getTaskStatus(taskID string) (*TaskStatusResponse, error) {
    url := fmt.Sprintf("http://localhost:8080/api/v1/tasks/%s", taskID)

    req, _ := http.NewRequest("GET", url, nil)
    req.Header.Set("X-API-Key", "sk_test_123456")

    client := &http.Client{}
    resp, err := client.Do(req)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()

    if resp.StatusCode == 404 {
        return nil, fmt.Errorf("任务未找到")
    }

    var status TaskStatusResponse
    json.NewDecoder(resp.Body).Decode(&status)

    return &status, nil
}

func waitForCompletion(taskID string, timeout time.Duration) (map[string]interface{}, error) {
    start := time.Now()

    for {
        status, err := getTaskStatus(taskID)
        if err != nil {
            return nil, err
        }

        switch status.Status {
        case "TASK_STATUS_COMPLETED":
            return status.Response, nil
        case "TASK_STATUS_FAILED":
            return nil, fmt.Errorf("任务失败：%s", status.Error)
        case "TASK_STATUS_TIMEOUT":
            return nil, fmt.Errorf("任务超时")
        }

        if time.Since(start) > timeout {
            return nil, fmt.Errorf("轮询超时")
        }

        time.Sleep(2 * time.Second)
    }
}

func main() {
    result, err := waitForCompletion("task_abc123", 5*time.Minute)
    if err != nil {
        fmt.Println("错误：", err)
        return
    }

    fmt.Println("结果：", result)
}

Bash - 监控任务进度

#!/bin/bash

API_KEY="sk_test_123456"
TASK_ID="$1"

if [ -z "$TASK_ID" ]; then
  echo "用法：$0 <task_id>"
  exit 1
fi

echo "监控任务：$TASK_ID"
echo ""

while true; do
  RESPONSE=$(curl -s "http://localhost:8080/api/v1/tasks/$TASK_ID" \
    -H "X-API-Key: $API_KEY")

  STATUS=$(echo $RESPONSE | jq -r '.status')

  echo "[$(date +%T)] 状态：$STATUS"

  case $STATUS in
    "TASK_STATUS_COMPLETED")
      echo ""
      echo "✓ 任务完成！"
      echo ""
      echo $RESPONSE | jq -r '.result'
      exit 0
      ;;
    "TASK_STATUS_FAILED")
      echo ""
      echo "✗ 任务失败！"
      ERROR=$(echo $RESPONSE | jq -r '.error')
      echo "错误：$ERROR"
      exit 1
      ;;
    "TASK_STATUS_TIMEOUT")
      echo ""
      echo "✗ 任务超时！"
      exit 1
      ;;
    "TASK_STATUS_CANCELLED")
      echo ""
      echo "✗ 任务已取消！"
      exit 1
      ;;
  esac

  sleep 2
done

用例

1. 提交并等待模式

import httpx
import time

def submit_and_wait(query: str, api_key: str):
    """提交任务并等待结果。"""
    # 提交
    submit_response = httpx.post(
        "http://localhost:8080/api/v1/tasks",
        headers={"X-API-Key": api_key},
        json={"query": query}
    )
    task_id = submit_response.json()["task_id"]
    print(f"任务已提交：{task_id}")

    # 等待
    while True:
        status_response = httpx.get(
            f"http://localhost:8080/api/v1/tasks/{task_id}",
            headers={"X-API-Key": api_key}
        )
        status = status_response.json()

        if status["status"] == "TASK_STATUS_COMPLETED":
            return status["result"]
        elif status["status"] == "TASK_STATUS_FAILED":
            raise Exception(status["error"])

        time.sleep(2)

result = submit_and_wait("Python 是什么？", "sk_test_123456")
print(result)

2. 仪表板状态部件

def get_task_summary(task_id: str, api_key: str):
    """获取用于仪表板的任务摘要。"""
    response = httpx.get(
        f"http://localhost:8080/api/v1/tasks/{task_id}",
        headers={"X-API-Key": api_key}
    )

    status = response.json()

    return {
        "id": task_id,
        "query": status["query"][:50] + "...",
        "status": status["status"].replace("TASK_STATUS_", ""),
        "mode": status["mode"].replace("EXECUTION_MODE_", ""),
        "created": status["created_at"]
    }

# 在 UI 中显示
summary = get_task_summary("task_abc123", "sk_test_123456")
print(f"{summary['status']}：{summary['query']}")

3. 批量状态检查

def check_multiple_tasks(task_ids: list, api_key: str):
    """检查多个任务的状态。"""
    results = {}

    for task_id in task_ids:
        try:
            response = httpx.get(
                f"http://localhost:8080/api/v1/tasks/{task_id}",
                headers={"X-API-Key": api_key},
                timeout=5.0
            )
            results[task_id] = response.json()["status"]
        except Exception as e:
            results[task_id] = f"错误：{e}"

    return results

# 检查 5 个任务
task_ids = ["task_1", "task_2", "task_3", "task_4", "task_5"]
statuses = check_multiple_tasks(task_ids, "sk_test_123456")

for task_id, status in statuses.items():
    print(f"{task_id}：{status}")

最佳实践

1. 使用流式传输而不是轮询

对于长时间运行的任务，使用 SSE 流式传输而不是轮询：

# ❌ 不好 - 每 2 秒轮询一次
while True:
    status = httpx.get(f".../{task_id}")
    if status["status"] == "COMPLETED":
        break
    time.sleep(2)

# ✅ 好 - 使用流式传输
for event in client.stream(task_id):
    print(event.type, event.message)
    if event.type == "TASK_COMPLETED":
        break

2. 处理所有状态

status = get_task_status(task_id, api_key)

match status["status"]:
    case "TASK_STATUS_QUEUED":
        print("任务已排队...")
    case "TASK_STATUS_RUNNING":
        print("任务正在运行...")
    case "TASK_STATUS_COMPLETED":
        result = status["result"]
        print(f"结果：{result}")
    case "TASK_STATUS_FAILED":
        print(f"失败：{status['error']}")
    case "TASK_STATUS_TIMEOUT":
        print("任务超时")
    case "TASK_STATUS_CANCELLED":
        print("任务已被取消")

3. 实施指数退避

import time

def poll_with_backoff(task_id, api_key, max_wait=60):
    """使用指数退避轮询。"""
    wait_time = 1

    while True:
        status = get_task_status(task_id, api_key)

        if status["status"] in ["TASK_STATUS_COMPLETED", "TASK_STATUS_FAILED"]:
            return status

        time.sleep(wait_time)
        wait_time = min(wait_time * 2, max_wait)  # 上限为 60 秒

4. 缓存状态响应

from functools import lru_cache
import time

@lru_cache(maxsize=1000)
def get_cached_status(task_id: str, api_key: str, timestamp: int):
    """缓存状态 5 秒。"""
    return get_task_status(task_id, api_key)

# 用法
current_time = int(time.time() / 5)  # 5 秒单位
status = get_cached_status("task_abc123", "sk_test_123456", current_time)

5. 提取元数据

def extract_task_info(task_id: str, api_key: str):
    """提取有用的元数据。"""
    status = get_task_status(task_id, api_key)

    return {
        "task_id": status["task_id"],
        "query": status["query"],
        "status": status["status"],
        "mode": status["mode"],
        "session_id": status["session_id"],
        "has_result": status["response"] is not None,
        "has_error": bool(status["error"]),
        "workflow_url": f"http://localhost:8088/workflows/{task_id}"
    }

提交任务

POST /api/v1/tasks

流式事件

实时监控

Python SDK

使用 client.get_status()

注意

勿在生产环境轮询：对于长时间运行的任务，使用流式端点而不是轮询状态。轮询会增加不必要的负载并增加延迟。

会话追踪：session_id 字段允许跟踪任务属于哪个会话，这对多轮对话和成本归属很有用。

​端点

​描述

​认证

​请求

​路径参数

​请求头

​响应

​成功响应

​响应字段

​状态值

​执行模式值

​示例

​检查任务状态

​深度研究响应载荷

​深度研究元数据字段

​示例：深度研究完成响应

​提取的事实（可选）

​引用对象模式

​验证对象模式

​错误响应

​401 未授权

​404 未找到

​500 内部服务器错误

​代码示例

​Python - 简单状态检查

​Python - 轮询直到完成

​JavaScript/Node.js

​JavaScript - 使用 Async/Await 轮询

​Go

​Bash - 监控任务进度

​用例

​1. 提交并等待模式

​2. 仪表板状态部件

​3. 批量状态检查

​最佳实践

​1. 使用流式传输而不是轮询

​2. 处理所有状态

​3. 实施指数退避

​4. 缓存状态响应

​5. 提取元数据

​相关端点

提交任务

流式事件

Python SDK

​注意

端点

描述

认证

请求

路径参数

请求头

响应

成功响应

响应字段

状态值

执行模式值

示例

检查任务状态

深度研究响应载荷

深度研究元数据字段

示例：深度研究完成响应

提取的事实（可选）

引用对象模式

验证对象模式

错误响应

401 未授权

404 未找到

500 内部服务器错误

代码示例

Python - 简单状态检查

Python - 轮询直到完成

JavaScript/Node.js

JavaScript - 使用 Async/Await 轮询

Go

Bash - 监控任务进度

用例

1. 提交并等待模式

2. 仪表板状态部件

3. 批量状态检查

最佳实践

1. 使用流式传输而不是轮询

2. 处理所有状态

3. 实施指数退避

4. 缓存状态响应

5. 提取元数据

相关端点

注意