Codex
721 lines
26 KiB
Python
721 lines
26 KiB
Python
"""
|
||
AI 工作流 API
|
||
"""
|
||
import json
|
||
import logging
|
||
from typing import Any, List, Optional
|
||
from datetime import datetime
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from fastapi.responses import StreamingResponse
|
||
from sqlalchemy import select, func, or_, and_
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from pydantic import BaseModel, Field
|
||
|
||
from app.database import get_db
|
||
from app.base_schema import PaginatedResponse, ResponseModel
|
||
from ai_platform.models import AIWorkflow, AIWorkflowVersion, AIWorkflowRun
|
||
from core.application.model import Application
|
||
from ai_platform.schemas.workflow_schema import (
|
||
WorkflowCreate,
|
||
WorkflowUpdate,
|
||
WorkflowResponse,
|
||
WorkflowListResponse,
|
||
WorkflowRunInput,
|
||
WorkflowRunResponse,
|
||
WorkflowRunListResponse,
|
||
WorkflowImportCheckIn,
|
||
WorkflowImportCheckOut,
|
||
WorkflowImportIn,
|
||
NodeSchemaResponse,
|
||
)
|
||
from ai_platform.services.workflow_import_export import (
|
||
WorkflowImportExportException,
|
||
export_config as export_workflow_config,
|
||
check_import as check_workflow_import,
|
||
import_config as import_workflow_config,
|
||
)
|
||
from ai_platform.nodes.registry import NodeRegistry
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/workflow", tags=["AI-工作流"])
|
||
|
||
|
||
# ============ 节点 Schema API ============
|
||
|
||
@router.get("/nodes/schemas", response_model=List[NodeSchemaResponse], summary="获取节点 Schema 列表")
|
||
async def get_node_schemas():
|
||
"""获取所有已注册节点的 Schema"""
|
||
return NodeRegistry.get_all_schemas()
|
||
|
||
|
||
@router.get("/nodes/schemas/by-category", summary="按分类获取节点 Schema")
|
||
async def get_node_schemas_by_category():
|
||
"""按分类获取所有已注册节点的 Schema"""
|
||
return NodeRegistry.get_schemas_by_category()
|
||
|
||
|
||
# ============ 工作流 API ============
|
||
|
||
@router.get("/list", response_model=PaginatedResponse[WorkflowListResponse], summary="工作流列表")
|
||
async def list_workflows(
|
||
name: Optional[str] = Query(None, description="名称"),
|
||
status: Optional[str] = Query(None, description="状态"),
|
||
workflow_type: Optional[str] = Query(None, description="工作流类型"),
|
||
application_id: Optional[str] = Query(None, alias="applicationId", description="所属应用ID"),
|
||
page: int = Query(1, ge=1, description="页码"),
|
||
page_size: int = Query(20, ge=1, le=1000, alias="pageSize", description="每页数量"),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""获取工作流列表（自动应用数据权限）"""
|
||
from ai_platform.services.workflow_service import AIWorkflowService
|
||
from app.data_scope_utils import get_data_scope_filter, apply_data_scope_to_conditions
|
||
|
||
conditions = [AIWorkflow.is_deleted == False]
|
||
|
||
if application_id:
|
||
conditions.append(or_(
|
||
AIWorkflow.application_id == application_id,
|
||
and_(AIWorkflow.application_id.is_(None), AIWorkflow.is_global == True)
|
||
))
|
||
if name:
|
||
conditions.append(AIWorkflow.name.ilike(f"%{name}%"))
|
||
if status:
|
||
conditions.append(AIWorkflow.status == status)
|
||
if workflow_type:
|
||
conditions.append(AIWorkflow.workflow_type == workflow_type)
|
||
|
||
# 获取数据权限过滤条件并应用
|
||
from ai_platform.services.workflow_service import RESOURCE_TYPE
|
||
data_scope_filter = await get_data_scope_filter(db, RESOURCE_TYPE)
|
||
scope_conditions = apply_data_scope_to_conditions(AIWorkflow, data_scope_filter)
|
||
conditions.extend(scope_conditions)
|
||
|
||
# 获取总数
|
||
query = select(AIWorkflow).where(and_(*conditions))
|
||
count_result = await db.execute(select(func.count()).select_from(query.subquery()))
|
||
total = count_result.scalar() or 0
|
||
|
||
# 分页
|
||
offset = (page - 1) * page_size
|
||
query = query.order_by(AIWorkflow.sort.desc(), AIWorkflow.sys_create_datetime.desc())
|
||
query = query.offset(offset).limit(page_size)
|
||
|
||
result = await db.execute(query)
|
||
workflows = result.scalars().all()
|
||
|
||
# 批量查询应用名称
|
||
app_ids = list({w.application_id for w in workflows if w.application_id})
|
||
app_name_map = {}
|
||
if app_ids:
|
||
app_result = await db.execute(
|
||
select(Application.id, Application.name).where(Application.id.in_(app_ids))
|
||
)
|
||
app_name_map = {row.id: row.name for row in app_result}
|
||
|
||
items = [_build_workflow_list_response(w, app_name_map.get(w.application_id, "")) for w in workflows]
|
||
|
||
return PaginatedResponse(items=items, total=total)
|
||
|
||
|
||
# 全局运行记录路由须注册在 /{workflow_id} 之前，避免 "runs" 被当作 workflow_id
|
||
@router.get("/runs", response_model=PaginatedResponse[WorkflowRunListResponse], summary="全局工作流运行记录")
|
||
async def list_all_workflow_runs(
|
||
workflow_id: Optional[str] = Query(None, alias="workflowId", description="工作流ID"),
|
||
status: Optional[str] = Query(None, description="运行状态"),
|
||
trigger_type: Optional[str] = Query(None, alias="triggerType", description="触发来源"),
|
||
page: int = Query(1, ge=1, description="页码"),
|
||
page_size: int = Query(20, ge=1, le=100, alias="pageSize", description="每页数量"),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""获取全局工作流运行记录（可按工作流、状态、触发来源筛选）"""
|
||
conditions = [AIWorkflowRun.is_deleted == False]
|
||
if workflow_id:
|
||
conditions.append(AIWorkflowRun.workflow_id == workflow_id)
|
||
if status:
|
||
conditions.append(AIWorkflowRun.status == status)
|
||
if trigger_type:
|
||
conditions.append(AIWorkflowRun.trigger_type == trigger_type)
|
||
|
||
query = select(AIWorkflowRun).where(*conditions).order_by(
|
||
AIWorkflowRun.sys_create_datetime.desc()
|
||
)
|
||
|
||
count_result = await db.execute(select(func.count()).select_from(query.subquery()))
|
||
total = count_result.scalar() or 0
|
||
|
||
offset = (page - 1) * page_size
|
||
result = await db.execute(query.offset(offset).limit(page_size))
|
||
runs = result.scalars().all()
|
||
|
||
workflow_ids = {r.workflow_id for r in runs if r.workflow_id}
|
||
workflow_name_map: dict[str, str] = {}
|
||
if workflow_ids:
|
||
wf_result = await db.execute(
|
||
select(AIWorkflow.id, AIWorkflow.name).where(AIWorkflow.id.in_(workflow_ids))
|
||
)
|
||
workflow_name_map = {row[0]: row[1] for row in wf_result.all()}
|
||
|
||
items = [
|
||
_build_run_list_item(r, workflow_name_map.get(r.workflow_id, ""))
|
||
for r in runs
|
||
]
|
||
return PaginatedResponse(items=items, total=total)
|
||
|
||
|
||
@router.get("/runs/{run_id}", response_model=WorkflowRunResponse, summary="运行记录详情")
|
||
async def get_workflow_run(run_id: str, db: AsyncSession = Depends(get_db)):
|
||
"""获取运行记录详情"""
|
||
result = await db.execute(
|
||
select(AIWorkflowRun).where(AIWorkflowRun.id == run_id, AIWorkflowRun.is_deleted == False)
|
||
)
|
||
run = result.scalar_one_or_none()
|
||
if not run:
|
||
raise HTTPException(status_code=404, detail="运行记录不存在")
|
||
|
||
workflow_result = await db.execute(
|
||
select(AIWorkflow).where(AIWorkflow.id == run.workflow_id)
|
||
)
|
||
workflow = workflow_result.scalar_one_or_none()
|
||
workflow_name = workflow.name if workflow else ""
|
||
|
||
return _build_run_detail(run, workflow_name)
|
||
|
||
|
||
@router.post("/runs/{run_id}/stop", response_model=ResponseModel, summary="停止运行")
|
||
async def stop_workflow_run(run_id: str, db: AsyncSession = Depends(get_db)):
|
||
"""停止工作流运行"""
|
||
result = await db.execute(
|
||
select(AIWorkflowRun).where(AIWorkflowRun.id == run_id, AIWorkflowRun.is_deleted == False)
|
||
)
|
||
run = result.scalar_one_or_none()
|
||
if not run:
|
||
raise HTTPException(status_code=404, detail="运行记录不存在")
|
||
|
||
if run.status not in ("pending", "running"):
|
||
raise HTTPException(status_code=400, detail="工作流已完成，无法停止")
|
||
|
||
run.status = "stopped"
|
||
await db.commit()
|
||
|
||
return ResponseModel(message="已停止")
|
||
|
||
|
||
class ResumeWorkflowInput(BaseModel):
|
||
"""恢复工作流输入"""
|
||
user_input: Any = Field(..., description="用户输入（可以是字符串、布尔值、对象等）")
|
||
|
||
|
||
@router.post("/runs/{run_id}/resume", summary="恢复工作流运行")
|
||
async def resume_workflow_run(
|
||
run_id: str,
|
||
data: ResumeWorkflowInput,
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""恢复等待中的工作流运行（SSE）"""
|
||
import json
|
||
from ai_platform.services.workflow_service import AIWorkflowService
|
||
|
||
async def generate():
|
||
workflow_service = AIWorkflowService(db)
|
||
async for event in workflow_service.resume_workflow_stream_async(
|
||
run_id=run_id,
|
||
user_input=data.user_input,
|
||
):
|
||
yield f"data: {json.dumps(event, ensure_ascii=False, default=str)}\n\n"
|
||
yield "data: [DONE]\n\n"
|
||
|
||
return StreamingResponse(
|
||
generate(),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache",
|
||
"Connection": "keep-alive",
|
||
}
|
||
)
|
||
|
||
|
||
@router.post("/runs/{run_id}/resume/stream", summary="流式恢复工作流运行")
|
||
async def resume_workflow_run_stream(
|
||
run_id: str,
|
||
data: ResumeWorkflowInput,
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""流式恢复等待中的工作流运行（SSE）"""
|
||
import json
|
||
from ai_platform.services.workflow_service import AIWorkflowService
|
||
|
||
async def generate():
|
||
workflow_service = AIWorkflowService(db)
|
||
async for event in workflow_service.resume_workflow_stream_async(
|
||
run_id=run_id,
|
||
user_input=data.user_input,
|
||
):
|
||
yield f"data: {json.dumps(event, ensure_ascii=False, default=str)}\n\n"
|
||
yield "data: [DONE]\n\n"
|
||
|
||
return StreamingResponse(
|
||
generate(),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache",
|
||
"Connection": "keep-alive",
|
||
}
|
||
)
|
||
|
||
|
||
@router.get("/code/{code}", response_model=WorkflowResponse, summary="根据编码获取工作流")
|
||
async def get_workflow_by_code(code: str, db: AsyncSession = Depends(get_db)):
|
||
"""根据编码获取工作流"""
|
||
result = await db.execute(
|
||
select(AIWorkflow).where(AIWorkflow.code == code, AIWorkflow.is_deleted == False)
|
||
)
|
||
workflow = result.scalar_one_or_none()
|
||
if not workflow:
|
||
raise HTTPException(status_code=404, detail="工作流不存在")
|
||
return _build_workflow_response(workflow)
|
||
|
||
|
||
@router.get("/{workflow_id}", response_model=WorkflowResponse, summary="工作流详情")
|
||
async def get_workflow(workflow_id: str, db: AsyncSession = Depends(get_db)):
|
||
"""获取工作流详情"""
|
||
result = await db.execute(
|
||
select(AIWorkflow).where(AIWorkflow.id == workflow_id, AIWorkflow.is_deleted == False)
|
||
)
|
||
workflow = result.scalar_one_or_none()
|
||
if not workflow:
|
||
raise HTTPException(status_code=404, detail="工作流不存在")
|
||
return _build_workflow_response(workflow)
|
||
|
||
|
||
@router.post("", response_model=WorkflowResponse, summary="创建工作流")
|
||
async def create_workflow(data: WorkflowCreate, db: AsyncSession = Depends(get_db)):
|
||
"""创建工作流"""
|
||
# 检查编码是否重复
|
||
exists_result = await db.execute(
|
||
select(AIWorkflow).where(AIWorkflow.code == data.code, AIWorkflow.is_deleted == False)
|
||
)
|
||
if exists_result.scalar_one_or_none():
|
||
raise HTTPException(status_code=400, detail=f"工作流编码 {data.code} 已存在")
|
||
|
||
workflow = AIWorkflow(**data.model_dump())
|
||
db.add(workflow)
|
||
await db.commit()
|
||
await db.refresh(workflow)
|
||
|
||
return _build_workflow_response(workflow)
|
||
|
||
|
||
@router.put("/{workflow_id}", response_model=WorkflowResponse, summary="更新工作流")
|
||
async def update_workflow(
|
||
workflow_id: str,
|
||
data: WorkflowUpdate,
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""更新工作流"""
|
||
result = await db.execute(
|
||
select(AIWorkflow).where(AIWorkflow.id == workflow_id, AIWorkflow.is_deleted == False)
|
||
)
|
||
workflow = result.scalar_one_or_none()
|
||
if not workflow:
|
||
raise HTTPException(status_code=404, detail="工作流不存在")
|
||
|
||
# 如果更新了 code，检查编码是否重复（排除自身）
|
||
if data.code and data.code != workflow.code:
|
||
exists_result = await db.execute(
|
||
select(AIWorkflow).where(
|
||
AIWorkflow.code == data.code,
|
||
AIWorkflow.id != workflow_id,
|
||
AIWorkflow.is_deleted == False
|
||
)
|
||
)
|
||
if exists_result.scalar_one_or_none():
|
||
raise HTTPException(status_code=400, detail=f"工作流编码 {data.code} 已存在")
|
||
|
||
update_data = data.model_dump(exclude_unset=True)
|
||
for key, value in update_data.items():
|
||
setattr(workflow, key, value)
|
||
|
||
# 更新草稿版本号
|
||
workflow.version = (workflow.version or 0) + 1
|
||
|
||
await db.commit()
|
||
await db.refresh(workflow)
|
||
|
||
return _build_workflow_response(workflow)
|
||
|
||
|
||
@router.delete("/{workflow_id}", response_model=ResponseModel, summary="删除工作流")
|
||
async def delete_workflow(workflow_id: str, db: AsyncSession = Depends(get_db)):
|
||
"""删除工作流"""
|
||
result = await db.execute(
|
||
select(AIWorkflow).where(AIWorkflow.id == workflow_id, AIWorkflow.is_deleted == False)
|
||
)
|
||
workflow = result.scalar_one_or_none()
|
||
if not workflow:
|
||
raise HTTPException(status_code=404, detail="工作流不存在")
|
||
|
||
workflow.is_deleted = True
|
||
await db.commit()
|
||
|
||
return ResponseModel(message="删除成功")
|
||
|
||
|
||
@router.post("/{workflow_id}/copy", response_model=WorkflowResponse, summary="复制工作流")
|
||
async def copy_workflow(workflow_id: str, db: AsyncSession = Depends(get_db)):
|
||
"""复制工作流"""
|
||
# 获取原工作流
|
||
result = await db.execute(
|
||
select(AIWorkflow).where(AIWorkflow.id == workflow_id, AIWorkflow.is_deleted == False)
|
||
)
|
||
workflow = result.scalar_one_or_none()
|
||
if not workflow:
|
||
raise HTTPException(status_code=404, detail="工作流不存在")
|
||
|
||
# 生成新的编码（原编码 + _copy + 时间戳）
|
||
import time
|
||
timestamp = int(time.time() * 1000)
|
||
new_code = f"{workflow.code}_copy_{timestamp}"
|
||
|
||
# 检查编码是否重复（理论上不会，但保险起见）
|
||
exists_result = await db.execute(
|
||
select(AIWorkflow).where(AIWorkflow.code == new_code, AIWorkflow.is_deleted == False)
|
||
)
|
||
if exists_result.scalar_one_or_none():
|
||
new_code = f"{workflow.code}_copy_{timestamp}_{int(time.time())}"
|
||
|
||
# 创建新工作流
|
||
new_workflow = AIWorkflow(
|
||
name=f"{workflow.name} (副本)",
|
||
code=new_code,
|
||
description=workflow.description,
|
||
workflow_type=workflow.workflow_type,
|
||
definition=workflow.definition, # 复制工作流定义
|
||
input_variables=workflow.input_variables,
|
||
output_variables=workflow.output_variables,
|
||
status="draft", # 新工作流默认为草稿状态
|
||
version=1,
|
||
published_version=None,
|
||
published_at=None,
|
||
published_definition=None,
|
||
)
|
||
db.add(new_workflow)
|
||
await db.commit()
|
||
await db.refresh(new_workflow)
|
||
|
||
return _build_workflow_response(new_workflow)
|
||
|
||
|
||
@router.get("/{workflow_id}/export", summary="导出工作流配置")
|
||
async def export_workflow(
|
||
workflow_id: str,
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""导出工作流配置为 JSON（草稿 definition）"""
|
||
try:
|
||
config = await export_workflow_config(db, workflow_id)
|
||
content = json.dumps(config, ensure_ascii=False, indent=2)
|
||
return StreamingResponse(
|
||
iter([content]),
|
||
media_type="application/json",
|
||
headers={
|
||
"Content-Disposition": f'attachment; filename="{config["code"]}.json"'
|
||
},
|
||
)
|
||
except WorkflowImportExportException as e:
|
||
raise HTTPException(status_code=400, detail=str(e))
|
||
|
||
|
||
@router.post("/import/check", response_model=WorkflowImportCheckOut, summary="导入预检查")
|
||
async def check_import_workflow(
|
||
data: WorkflowImportCheckIn,
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""导入预检查：检查工作流编码是否冲突"""
|
||
try:
|
||
return await check_workflow_import(db, data.code)
|
||
except WorkflowImportExportException as e:
|
||
raise HTTPException(status_code=400, detail=str(e))
|
||
|
||
|
||
@router.post("/import", response_model=WorkflowResponse, summary="导入工作流配置")
|
||
async def import_workflow(
|
||
data: WorkflowImportIn,
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""导入工作流配置"""
|
||
try:
|
||
workflow = await import_workflow_config(db, data.model_dump())
|
||
await db.commit()
|
||
await db.refresh(workflow)
|
||
return _build_workflow_response(workflow)
|
||
except WorkflowImportExportException as e:
|
||
raise HTTPException(status_code=400, detail=str(e))
|
||
|
||
|
||
@router.post("/{workflow_id}/publish", response_model=WorkflowResponse, summary="发布工作流")
|
||
async def publish_workflow(
|
||
workflow_id: str,
|
||
description: str = Query("", description="版本说明"),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""发布工作流"""
|
||
result = await db.execute(
|
||
select(AIWorkflow).where(AIWorkflow.id == workflow_id, AIWorkflow.is_deleted == False)
|
||
)
|
||
workflow = result.scalar_one_or_none()
|
||
if not workflow:
|
||
raise HTTPException(status_code=404, detail="工作流不存在")
|
||
|
||
# 创建版本记录
|
||
new_version = (workflow.published_version or 0) + 1
|
||
version = AIWorkflowVersion(
|
||
workflow_id=workflow_id,
|
||
version=new_version,
|
||
definition=workflow.definition or {},
|
||
description=description,
|
||
published_at=datetime.now(),
|
||
)
|
||
db.add(version)
|
||
|
||
# 更新工作流
|
||
workflow.status = "published"
|
||
workflow.published_version = new_version
|
||
workflow.published_at = datetime.now()
|
||
workflow.published_definition = workflow.definition
|
||
|
||
await db.commit()
|
||
await db.refresh(workflow)
|
||
|
||
return _build_workflow_response(workflow)
|
||
|
||
|
||
@router.get("/{workflow_id}/versions", summary="获取版本历史")
|
||
async def list_workflow_versions(
|
||
workflow_id: str,
|
||
page: int = Query(1, ge=1, description="页码"),
|
||
page_size: int = Query(20, ge=1, le=100, alias="pageSize", description="每页数量"),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""获取工作流版本历史"""
|
||
query = select(AIWorkflowVersion).where(
|
||
AIWorkflowVersion.workflow_id == workflow_id,
|
||
AIWorkflowVersion.is_deleted == False
|
||
).order_by(AIWorkflowVersion.version.desc())
|
||
|
||
# 获取总数
|
||
count_result = await db.execute(select(func.count()).select_from(query.subquery()))
|
||
total = count_result.scalar() or 0
|
||
|
||
# 分页
|
||
offset = (page - 1) * page_size
|
||
query = query.offset(offset).limit(page_size)
|
||
|
||
result = await db.execute(query)
|
||
versions = result.scalars().all()
|
||
|
||
items = [
|
||
{
|
||
"id": v.id,
|
||
"version": v.version,
|
||
"description": v.description or "",
|
||
"published_at": v.published_at,
|
||
"run_count": v.run_count or 0,
|
||
"success_count": v.success_count or 0,
|
||
}
|
||
for v in versions
|
||
]
|
||
|
||
return PaginatedResponse(items=items, total=total)
|
||
|
||
|
||
@router.get("/{workflow_id}/runs", response_model=PaginatedResponse[WorkflowRunListResponse], summary="工作流运行记录")
|
||
async def list_workflow_runs(
|
||
workflow_id: str,
|
||
page: int = Query(1, ge=1, description="页码"),
|
||
page_size: int = Query(20, ge=1, le=100, alias="pageSize", description="每页数量"),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""获取工作流运行记录"""
|
||
query = select(AIWorkflowRun).where(
|
||
AIWorkflowRun.workflow_id == workflow_id,
|
||
AIWorkflowRun.is_deleted == False
|
||
).order_by(AIWorkflowRun.sys_create_datetime.desc())
|
||
|
||
# 获取总数
|
||
count_result = await db.execute(select(func.count()).select_from(query.subquery()))
|
||
total = count_result.scalar() or 0
|
||
|
||
# 分页
|
||
offset = (page - 1) * page_size
|
||
query = query.offset(offset).limit(page_size)
|
||
|
||
result = await db.execute(query)
|
||
runs = result.scalars().all()
|
||
|
||
# 获取工作流名称
|
||
workflow_result = await db.execute(
|
||
select(AIWorkflow).where(AIWorkflow.id == workflow_id)
|
||
)
|
||
workflow = workflow_result.scalar_one_or_none()
|
||
workflow_name = workflow.name if workflow else ""
|
||
|
||
items = [
|
||
_build_run_list_item(r, workflow_name)
|
||
for r in runs
|
||
]
|
||
|
||
return PaginatedResponse(items=items, total=total)
|
||
|
||
|
||
def _build_run_list_item(run: AIWorkflowRun, workflow_name: str = "") -> dict:
|
||
return {
|
||
"id": run.id,
|
||
"workflow_id": run.workflow_id,
|
||
"workflow_name": workflow_name,
|
||
"status": run.status or "pending",
|
||
"trigger_type": run.trigger_type or "api",
|
||
"total_steps": run.total_steps or 0,
|
||
"total_tokens": run.total_tokens or 0,
|
||
"elapsed_time": run.elapsed_time or 0,
|
||
"error_message": (run.error_message or "")[:200],
|
||
"started_at": run.started_at,
|
||
"completed_at": run.completed_at,
|
||
}
|
||
|
||
|
||
def _build_run_detail(run: AIWorkflowRun, workflow_name: str = "") -> dict:
|
||
return {
|
||
"id": run.id,
|
||
"workflow_id": run.workflow_id,
|
||
"workflow_name": workflow_name,
|
||
"status": run.status or "pending",
|
||
"trigger_type": run.trigger_type or "api",
|
||
"use_draft": bool(run.use_draft),
|
||
"workflow_version": run.workflow_version,
|
||
"definition_snapshot": run.definition_snapshot or {},
|
||
"inputs": run.inputs or {},
|
||
"outputs": run.outputs or {},
|
||
"execution_log": run.execution_log or [],
|
||
"current_node_id": run.current_node_id or "",
|
||
"waiting_config": run.waiting_config or {},
|
||
"error_message": run.error_message or "",
|
||
"total_tokens": run.total_tokens or 0,
|
||
"total_steps": run.total_steps or 0,
|
||
"elapsed_time": run.elapsed_time or 0,
|
||
"started_at": run.started_at,
|
||
"completed_at": run.completed_at,
|
||
"sys_create_datetime": run.sys_create_datetime,
|
||
}
|
||
|
||
|
||
class WorkflowRunInput(BaseModel):
|
||
"""工作流运行输入"""
|
||
inputs: dict = Field(default_factory=dict, description="输入变量")
|
||
use_draft: bool = Field(default=False, description="是否使用草稿版本")
|
||
|
||
|
||
@router.post("/{workflow_id}/run", summary="运行工作流")
|
||
async def run_workflow(
|
||
workflow_id: str,
|
||
data: WorkflowRunInput,
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""运行工作流（非流式）"""
|
||
from ai_platform.services.workflow_service import AIWorkflowService
|
||
|
||
workflow_service = AIWorkflowService(db)
|
||
run = await workflow_service.run_workflow(
|
||
workflow_id=workflow_id,
|
||
inputs=data.inputs,
|
||
use_draft=data.use_draft,
|
||
trigger_type='api',
|
||
)
|
||
|
||
return {
|
||
"id": run.id,
|
||
"workflow_id": run.workflow_id,
|
||
"status": run.status,
|
||
"outputs": run.outputs or {},
|
||
"execution_log": run.execution_log or [],
|
||
"total_tokens": run.total_tokens or 0,
|
||
"total_steps": run.total_steps or 0,
|
||
"elapsed_time": run.elapsed_time or 0,
|
||
}
|
||
|
||
|
||
@router.post("/{workflow_id}/run/stream", summary="流式运行工作流")
|
||
async def run_workflow_stream(
|
||
workflow_id: str,
|
||
data: WorkflowRunInput,
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""流式运行工作流（SSE）"""
|
||
import json
|
||
from ai_platform.services.workflow_service import AIWorkflowService
|
||
|
||
async def generate():
|
||
workflow_service = AIWorkflowService(db)
|
||
trigger_type = 'editor_draft' if data.use_draft else 'editor_published'
|
||
async for event in workflow_service.run_workflow_stream_async(
|
||
workflow_id=workflow_id,
|
||
inputs=data.inputs,
|
||
use_draft=data.use_draft,
|
||
trigger_type=trigger_type,
|
||
):
|
||
yield f"data: {json.dumps(event, ensure_ascii=False, default=str)}\n\n"
|
||
yield "data: [DONE]\n\n"
|
||
|
||
return StreamingResponse(
|
||
generate(),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache",
|
||
"Connection": "keep-alive",
|
||
}
|
||
)
|
||
|
||
|
||
def _build_workflow_response(workflow: AIWorkflow) -> dict:
|
||
"""构建工作流输出"""
|
||
return {
|
||
"id": workflow.id,
|
||
"name": workflow.name,
|
||
"code": workflow.code,
|
||
"workflow_type": workflow.workflow_type or "general",
|
||
"description": workflow.description or "",
|
||
"status": workflow.status or "draft",
|
||
"version": workflow.version or 1,
|
||
"published_version": workflow.published_version,
|
||
"published_at": workflow.published_at,
|
||
"published_definition": workflow.published_definition,
|
||
"definition": workflow.definition or {},
|
||
"input_variables": workflow.input_variables or [],
|
||
"output_variables": workflow.output_variables or [],
|
||
"run_count": workflow.run_count or 0,
|
||
"success_count": workflow.success_count or 0,
|
||
"sort": workflow.sort or 0,
|
||
"sys_create_datetime": workflow.sys_create_datetime,
|
||
"sys_update_datetime": workflow.sys_update_datetime,
|
||
}
|
||
|
||
|
||
def _build_workflow_list_response(workflow: AIWorkflow, application_name: str = "") -> dict:
|
||
"""构建工作流列表输出"""
|
||
return {
|
||
"id": workflow.id,
|
||
"application_id": workflow.application_id,
|
||
"application_name": application_name,
|
||
"is_global": workflow.is_global or False,
|
||
"name": workflow.name,
|
||
"code": workflow.code,
|
||
"workflow_type": workflow.workflow_type or "general",
|
||
"description": workflow.description or "",
|
||
"status": workflow.status or "draft",
|
||
"version": workflow.version or 1,
|
||
"run_count": workflow.run_count or 0,
|
||
"success_count": workflow.success_count or 0,
|
||
"sys_create_datetime": workflow.sys_create_datetime,
|
||
}
|