数据查询
这些函数通过 TASK_JWT 调用 TaskFlow 内部 API，用于在脚本内查询项目数据。
list_tasks()
列出当前项目的任务实例，按创建时间降序排列。
python
def list_tasks(
status: str | None = None,
limit: int = 50,
offset: int = 0,
) -> list[dict]| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
status | str | None | None | 按状态过滤：pending、running、success、failed、pending_approval、rejected |
limit | int | 50 | 最多返回条数（最大 200） |
offset | int | 0 | 分页偏移 |
返回值： 任务 dict 列表，每项包含 id、name、status、progress、task_type 等字段。
python
running = list_tasks(status="running")
for t in running:
print(f"{t['name']} — {t['progress']}%")get_task()
获取单个任务实例的当前状态（非阻塞快照，不轮询）。
python
def get_task(task_id: str) -> dict成功/已审批的任务会自动从执行日志中附加 result 字段。
python
task = get_task("550e8400-e29b-41d4-a716-446655440000")
if task["status"] == "success":
print(task.get("result"))list_agents()
列出当前项目的所有 Agent。
python
def list_agents() -> list[dict]返回值： 每项包含 id、name、description、is_default、agent_type。
list_workspaces()
列出当前项目的所有工作区。
python
def list_workspaces() -> list[dict]返回值： 每项包含 id、name、slug、type、is_default、artifact_type。
python
kb_ws = next(
(w for w in list_workspaces() if w["artifact_type"] == "knowledge_base"),
None,
)list_task_type_memories()
查询元任务的历史记忆。支持分页列举和语义相似度检索两种模式。
python
def list_task_type_memories(
task_type: str,
query: str | None = None,
k: int = 20,
order: str = "most", # "most" | "least"
limit: int = 20,
offset: int = 0,
) -> list[dict]| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
task_type | str | 必填 | 元任务 identifier 或 UUID |
query | str | None | None | 提供时切换为语义检索模式，limit/offset 忽略 |
k | int | 20 | 语义检索返回条数（最大 100） |
order | str | "most" | "most" = 最相关，"least" = 最不相关 |
limit | int | 20 | 分页模式页大小 |
offset | int | 0 | 分页偏移 |
python
# 分页列举
for m in list_task_type_memories("summariser", limit=10):
print(m["memory"])
# 语义检索——最相关 5 条
hits = list_task_type_memories("summariser", query="认证流程", k=5)
# 最不相关 5 条（探索差异最大的记忆）
novel = list_task_type_memories("summariser", query="认证流程", k=5, order="least")get_task_memories()
查询单次任务运行贡献的记忆。
python
def get_task_memories(
task_id: str,
limit: int = 20,
offset: int = 0,
) -> list[dict]python
for m in get_task_memories(TASK_ID):
print(m["memory"])okf_search()
在项目知识库工作区中搜索 OKF 概念。
python
def okf_search(query: str, workspace_id: str | None = None) -> list[dict]| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
query | str | 必填 | 搜索词 |
workspace_id | str | None | None | 知识库工作区 UUID；默认使用项目的知识库工作区 |
python
hits = okf_search("身份验证流程")
for h in hits:
print(h["title"], "-", h.get("description", ""))okf_lint()
对项目知识库运行健康检查（断链、孤儿概念、缺失类型/引用、重复标题）。
python
def okf_lint(workspace_id: str | None = None) -> list[dict]python
issues = okf_lint()
for issue in issues:
print(f"[{issue['issue_type']}] {issue['concept_id']}: {issue['message']}")