任务管理器 {#guide-task-manager}¶

什么时候进入本页 {#guide-task-manager-when-to-read}¶

当你需要：

统一管理多个量子云平台的任务
使用本地缓存保存任务状态和结果
批量提交多个量子线路
在开发阶段使用本地模拟代替真实云平台

就应该进入本页。

本页介绍的是 统一任务管理接口，提供跨平台的一致 API，简化云平台任务的提交、查询和结果管理。

概述 {#guide-task-manager-overview}¶

task_manager 模块提供了高层次的 API 来管理量子计算任务：

统一接口：支持 OriginQ、IBM Quantum 和本地 Dummy 模拟器
本地缓存：自动保存任务状态和结果，支持离线查询
批量操作：支持批量提交和批量查询
Dummy 模式：开发测试时可使用本地模拟代替真实平台

快速开始 {#guide-task-manager-quickstart}¶

安装依赖¶

# 基础安装（包含 OriginQ / IBM Qiskit 的核心依赖）
pip install unified-quantum

# 本地模拟（Dummy 模式）
pip install unified-quantum[simulation]

# QuarkStudio / Quark 云平台（要求 Python ≥ 3.12）
pip install unified-quantum[quark]

# 全部维护中的可选依赖
pip install unified-quantum[all]

[all] 会安装当前维护的可选功能依赖。

配置云平台凭据¶

uniqc config set originq.token "your-api-key"
uniqc config set ibm.token "your-ibm-token"
uniqc config set quark.host "your-quark-host"

基本用法¶

from uniqc import Circuit
from uniqc import submit_task, wait_for_result

# 创建量子线路
circuit = Circuit()
circuit.h(0)
circuit.cnot(0, 1)
circuit.measure(0, 1)

# 提交到云平台（直接传 Circuit 对象，也接受 OriginIR/QASM 字符串）
task_id = submit_task(
    circuit,
    backend='originq:WK_C180',
    shots=1000
)

# 等待结果（阻塞直到完成）
result = wait_for_result(task_id, timeout=300)
print(result.counts)

Bitstring 约定（永久不变）：所有平台返回的 result.counts 字典 key 都满足 c[0] = 最右字符（LSB）、c[N-1] = 最左字符；c[i] 是第 i 次 circuit.measure(q) 调用的结果。详见 submit_task §结果处理与 platform_conventions §2.6。由 uniqc/test/test_endianness_convention.py 在所有 dummy 平台上回归保护。

核心 API {#guide-task-manager-core-api}¶

任务提交¶

submit_task()¶

提交单个量子线路到云平台：

from uniqc import submit_task

task_id = submit_task(
    circuit,              # Circuit 对象、OriginIR 字符串、QASM 字符串或 qiskit.QuantumCircuit
    backend='originq:WK_C180',  # 平台选择：'originq:chip', 'ibm:chip', 'dummy:local:simulator'
    shots=1000,           # 测量次数
    **kwargs              # 其他平台相关参数
)

参数说明：

参数	类型	说明
`circuit`	Circuit \| str \| qiskit.QuantumCircuit	量子线路（自动检测并转换为内部 Circuit）
`backend`	str	平台标识：`'originq:WK_C180'`, `'ibm:ibm_brisbane'`, `'dummy:local:simulator'`
`shots`	int	测量次数，默认 1000
`chip_id`	str	目标芯片 ID（可选）

submit_batch()¶

批量提交多个量子线路：

from uniqc import submit_batch

task_ids = submit_batch(
    circuits,             # 线路列表
    backend='originq:WK_C180',
    shots=1000
)

任务查询¶

query_task()¶

查询单个任务状态：

from uniqc import query_task

info = query_task(task_id)
print(info.status)  # TaskStatus.RUNNING / SUCCESS / FAILED

wait_for_result()¶

阻塞等待任务完成并返回结果：

from uniqc import wait_for_result

result = wait_for_result(
    task_id,
    timeout=300,      # 超时时间（秒）
    poll_interval=5   # 轮询间隔（秒）
)

get_result()¶

阻塞获取结果（wait_for_result 的别名，强调”我要结果”的语义）：

from uniqc import get_result

result = get_result(task_id, timeout=300, poll_interval=5)

poll_result()¶

非阻塞状态查询（立即返回当前状态，适合在循环中轮询）：

from uniqc import poll_result, TaskStatus

info = poll_result(task_id)
if info.status == TaskStatus.SUCCESS:
    print(info.result)

任务管理¶

list_tasks()¶

列出所有缓存的任务：

from uniqc import list_tasks

tasks = list_tasks()
for task in tasks:
    print(f"Task {task['task_id']}: {task['status']}")

get_task()¶

获取特定任务的详细信息：

from uniqc import get_task

task_info = get_task(task_id)

clear_completed_tasks()¶

清理已完成的任务：

from uniqc import clear_completed_tasks

cleared = clear_completed_tasks()
print(f"Cleared {cleared} tasks")

clear_cache()¶

清空所有缓存：

from uniqc import clear_cache

clear_cache()

Dummy 模式 {#guide-task-manager-dummy-mode}¶

通过 backend 名称前缀 dummy 激活本地模拟：

from uniqc import submit_task

# 默认 dummy 模拟
task_id = submit_task(circuit, backend='dummy:local:simulator')

# 带 chip 特征的 dummy 模拟
task_id = submit_task(circuit, backend='dummy:originq:WK_C180')

使用 DummyBackend¶

新代码推荐通过 submit_task(..., backend=...) 选择 dummy backend id：

from uniqc import submit_task, wait_for_result

# 无约束、无噪声
task_id = submit_task(circuit, backend="dummy:local:simulator", shots=1000)

# 指定虚拟拓扑，仍然无噪声
line_task = submit_task(circuit, backend="dummy:local:virtual-line-3", shots=1000)

# 复用真实 backend 拓扑和标定数据，先 compile/transpile，再本地含噪执行
noisy_task = submit_task(circuit, backend="dummy:originq:WK_C180", shots=1000)

result = wait_for_result(task_id)

dummy:<platform>:<backend> 是规则型写法，不会作为独立 backend 出现在后端列表或 Gateway WebUI 中。

底层测试仍可直接使用 DummyAdapter 进行本地模拟：

from uniqc.backend_adapter.task.adapters.dummy_adapter import DummyAdapter

adapter = DummyAdapter()

# 支持噪声模拟
adapter_with_noise = DummyAdapter(
    noise_model={'depol': 0.01}  # 1% 去极化噪声
)

# 提交任务
task_id = adapter.submit(circuit, shots=1000)
result = adapter.query(task_id)

任务持久化 {#guide-task-manager-persistence}¶

TaskPersistence 类¶

使用 TaskPersistence 进行更细粒度的任务存储管理：

from uniqc.backend_adapter.task.persistence import TaskPersistence

# 创建持久化实例
persistence = TaskPersistence()

# 保存任务
persistence.save(task_id, {
    'status': 'running',
    'backend': 'originq',
    'circuit': circuit
})

# 加载任务
task = persistence.load(task_id)

# 列出所有任务
all_tasks = persistence.list_all()

# 按平台筛选
originq_tasks = persistence.list_by_platform('originq')

# 列出待处理任务
pending = persistence.list_pending()

# 清理已完成任务
persistence.clear_completed()

存储位置¶

任务数据默认存储在 ~/.uniqc/cache/tasks.sqlite 单一 SQLite 数据库中。数据库使用 PRAGMA application_id（常量值 "UNIC"）校验文件归属，并通过 PRAGMA user_version 维护 schema 版本；打开时如发现新版本会自动按 uniqc.backend_adapter.task.store.MIGRATIONS 顺序迁移。可通过向 TaskPersistence(cache_dir=...) 或 TaskStore(cache_dir=...) 传入其他路径覆盖默认位置。

错误处理 {#guide-task-manager-error-handling}¶

MissingDependencyError¶

当缺少平台依赖时抛出：

from uniqc.backend_adapter.task.optional_deps import MissingDependencyError

try:
    from uniqc.backend_adapter.task.adapters.qiskit_adapter import QiskitAdapter
    adapter = QiskitAdapter()
except MissingDependencyError as e:
    print(f"Missing dependency: {e.package}")
    print(f"Install with: pip install unified-quantum[{e.extra}]")

任务状态¶

任务可能有以下状态：

状态	说明
`pending`	任务已提交但尚未开始执行
`running`	任务正在执行中
`success`	任务成功完成
`failed`	任务执行失败

平台对比 {#guide-task-manager-platform-comparison}¶

特性	OriginQ	IBM	Dummy
输入格式	OriginIR string	qiskit.QuantumCircuit	OriginIR string
结果格式	`{"00": 512}`	`[{"00": 512}, ...]`	`{"00": 512}`
真机支持	✓	✓	✗
噪声模拟	✗	✗	✓
提交模式	异步	同步	同步
网络要求	需要	需要	不需要
适用场景	生产环境	国际平台	开发测试

下一步 {#guide-task-manager-next-steps}¶

了解云平台适配器架构
查看具体的任务提交指南
参考 API 文档：uniqc.backend_adapter.task_manager