> ## Documentation Index
> Fetch the complete documentation index at: https://docs.shannon.run/llms.txt
> Use this file to discover all available pages before exploring further.
# 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 相同）
**响应体**：
```json theme={null}
{
"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` - 高级推理模式
## 示例
### 检查任务状态
```bash theme={null}
curl -X GET "http://localhost:8080/api/v1/tasks/task_01HQZX3Y9K8M2P4N5S7T9W2V" \
-H "X-API-Key: sk_test_123456"
```
**响应（已排队）**：
```json theme={null}
{
"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"
}
```
**响应（运行中）**：
```json theme={null}
{
"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"
}
```
**响应（已完成）**：
```json theme={null}
{
"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
}
}
```
**响应（失败）**：
```json theme={null}
{
"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 | 提取事实的摘要统计 |
### 示例：深度研究完成响应
```json theme={null}
{
"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` 时：
```json theme={null}
{
"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 未授权
```json theme={null}
{
"error": "Unauthorized"
}
```
### 404 未找到
```json theme={null}
{
"error": "Task not found"
}
```
### 500 内部服务器错误
```json theme={null}
{
"error": "Failed to get task status: database error"
}
```
## 代码示例
### Python - 简单状态检查
```python theme={null}
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 - 轮询直到完成
```python theme={null}
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
```javascript theme={null}
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 轮询
```javascript theme={null}
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
```go theme={null}
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 - 监控任务进度
```bash theme={null}
#!/bin/bash
API_KEY="sk_test_123456"
TASK_ID="$1"
if [ -z "$TASK_ID" ]; then
echo "用法：$0 "
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. 提交并等待模式
```python theme={null}
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. 仪表板状态部件
```python theme={null}
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. 批量状态检查
```python theme={null}
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 流式传输而不是轮询：
```python theme={null}
# ❌ 不好 - 每 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. 处理所有状态
```python theme={null}
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. 实施指数退避
```python theme={null}
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. 缓存状态响应
```python theme={null}
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. 提取元数据
```python theme={null}
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
实时监控
使用 client.get\_status()
## 注意
**勿在生产环境轮询**：对于长时间运行的任务，使用流式端点而不是轮询状态。轮询会增加不必要的负载并增加延迟。
**会话追踪**：`session_id` 字段允许跟踪任务属于哪个会话，这对多轮对话和成本归属很有用。
***