import uuid import time import json import hashlib import secrets from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, Request, Query from fastapi.responses import StreamingResponse from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from database import get_db from models import FlowDefinition, FlowVersion, FlowApiKey, FlowExecution, User from schemas import ( FlowDefinitionCreate, FlowDefinitionUpdate, FlowDefinitionOut, FlowVersionOut, FlowApiKeyCreate, FlowApiKeyOut, FlowExecuteRequest, FlowChatMessageRequest, ) from modules.flow_engine.engine import FlowEngine, ToolNodeAgent from agentscope.message import Msg from dependencies import get_current_user import logging logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/flow", tags=["flow"]) def _build_flow_out(f) -> FlowDefinitionOut: return FlowDefinitionOut( id=f.id, name=f.name, description=f.description, version=f.version, status=f.status, definition_json=f.definition_json, published_version_id=f.published_version_id, published_to_wecom=f.published_to_wecom, published_to_web=f.published_to_web, created_at=f.created_at, updated_at=f.updated_at, ) # ============================== CRUD ============================== @router.get("/definitions", response_model=list[FlowDefinitionOut]) async def list_flows(request: Request, db: AsyncSession = Depends(get_db)): result = await db.execute( select(FlowDefinition).order_by(FlowDefinition.updated_at.desc()) ) return [_build_flow_out(f) for f in result.scalars().all()] @router.get("/definitions/{flow_id}", response_model=FlowDefinitionOut) async def get_flow(flow_id: uuid.UUID, request: Request, db: AsyncSession = Depends(get_db)): f = await db.get(FlowDefinition, flow_id) if not f: raise HTTPException(404, "流定义不存在") return _build_flow_out(f) @router.post("/definitions", response_model=FlowDefinitionOut) async def create_flow(req: FlowDefinitionCreate, request: Request, db: AsyncSession = Depends(get_db)): user_ctx = request.state.user definition_json = { "nodes": [n.model_dump() for n in req.nodes], "edges": [e.model_dump() for e in req.edges], "trigger": req.trigger, } flow = FlowDefinition( name=req.name, description=req.description, definition_json=definition_json, draft_definition_json=definition_json, creator_id=uuid.UUID(user_ctx["id"]), ) db.add(flow) await db.flush() return _build_flow_out(flow) @router.put("/definitions/{flow_id}", response_model=FlowDefinitionOut) async def update_flow(flow_id: uuid.UUID, req: FlowDefinitionUpdate, request: Request, db: AsyncSession = Depends(get_db)): f = await db.get(FlowDefinition, flow_id) if not f: raise HTTPException(404, "流定义不存在") if req.name is not None: f.name = req.name if req.description is not None: f.description = req.description if req.nodes is not None and req.edges is not None: new_def = { "nodes": [n.model_dump() for n in req.nodes], "edges": [e.model_dump() for e in req.edges], "trigger": req.trigger or f.definition_json.get("trigger", {}), } f.version += 1 f.draft_definition_json = new_def f.definition_json = new_def return _build_flow_out(f) @router.delete("/definitions/{flow_id}") async def delete_flow(flow_id: uuid.UUID, db: AsyncSession = Depends(get_db)): f = await db.get(FlowDefinition, flow_id) if not f: raise HTTPException(404, "流定义不存在") await db.delete(f) return {"code": 200, "message": "已删除"} # ============================== 发布 (创建快照) ============================== async def _snapshot_publish(flow: FlowDefinition, db: AsyncSession, user_id: str, publish_wecom: bool = False, publish_web: bool = False, changelog: str = ""): new_version = FlowVersion( flow_id=flow.id, version=flow.version, definition_json=json.loads(json.dumps(flow.definition_json)), changelog=changelog, published_by=uuid.UUID(user_id), published_to_wecom=publish_wecom, published_to_web=publish_web, ) db.add(new_version) await db.flush() flow.published_version_id = new_version.id flow.status = "published" if publish_wecom: flow.published_to_wecom = True if publish_web: flow.published_to_web = True return new_version @router.post("/definitions/{flow_id}/publish") async def publish_flow(flow_id: uuid.UUID, request: Request, db: AsyncSession = Depends(get_db)): f = await db.get(FlowDefinition, flow_id) if not f: raise HTTPException(404, "流定义不存在") nodes = f.definition_json.get("nodes", []) if not nodes: raise HTTPException(400, "流定义中没有节点") user_ctx = request.state.user await _snapshot_publish(f, db, user_ctx["id"], publish_wecom=True) return {"code": 200, "message": "流已上架到企微", "data": {"status": "published", "version": f.version}} @router.post("/definitions/{flow_id}/publish-web") async def publish_flow_to_web(flow_id: uuid.UUID, request: Request, db: AsyncSession = Depends(get_db)): f = await db.get(FlowDefinition, flow_id) if not f: raise HTTPException(404, "流定义不存在") user_ctx = request.state.user prev_version_id = f.published_version_id await _snapshot_publish(f, db, user_ctx["id"], publish_web=True) return {"code": 200, "message": "流已上架到网页", "data": {"status": "published", "version": f.version}} @router.post("/definitions/{flow_id}/unpublish") async def unpublish_flow(flow_id: uuid.UUID, db: AsyncSession = Depends(get_db)): f = await db.get(FlowDefinition, flow_id) if not f: raise HTTPException(404, "流定义不存在") f.status = "draft" f.published_to_wecom = False f.published_version_id = None return {"code": 200, "message": "流已下架"} @router.post("/definitions/{flow_id}/unpublish-web") async def unpublish_flow_from_web(flow_id: uuid.UUID, db: AsyncSession = Depends(get_db)): f = await db.get(FlowDefinition, flow_id) if not f: raise HTTPException(404, "流定义不存在") f.published_to_web = False if not f.published_to_wecom: f.status = "draft" f.published_version_id = None return {"code": 200, "message": "流已从网页下架"} # ============================== 版本管理 ============================== @router.get("/definitions/{flow_id}/versions", response_model=list[FlowVersionOut]) async def list_versions(flow_id: uuid.UUID, db: AsyncSession = Depends(get_db)): result = await db.execute( select(FlowVersion) .where(FlowVersion.flow_id == flow_id) .order_by(FlowVersion.version.desc()) .limit(50) ) return [FlowVersionOut( id=v.id, flow_id=v.flow_id, version=v.version, definition_json=v.definition_json, changelog=v.changelog or "", published_to_wecom=v.published_to_wecom, published_to_web=v.published_to_web, published_by=v.published_by, created_at=v.created_at, ) for v in result.scalars().all()] @router.post("/definitions/{flow_id}/rollback/{version_id}") async def rollback_flow(flow_id: uuid.UUID, version_id: uuid.UUID, db: AsyncSession = Depends(get_db)): f = await db.get(FlowDefinition, flow_id) if not f: raise HTTPException(404, "流定义不存在") target = await db.get(FlowVersion, version_id) if not target or str(target.flow_id) != str(flow_id): raise HTTPException(404, "版本不存在") f.definition_json = json.loads(json.dumps(target.definition_json)) f.draft_definition_json = f.definition_json f.published_version_id = target.id f.published_to_wecom = target.published_to_wecom f.published_to_web = target.published_to_web f.status = "published" if (target.published_to_wecom or target.published_to_web) else "draft" f.version = target.version return {"code": 200, "message": f"已回滚到版本 v{target.version}", "data": {"version": target.version}} # ============================== 执行 (加版本快照) ============================== def _get_definition_json(flow: FlowDefinition, db_session) -> dict: """优先加载 published_version 快照，不存在则使用当前 definition_json""" return flow.definition_json async def _get_published_definition(flow: FlowDefinition, db: AsyncSession) -> dict: if flow.published_version_id: result = await db.execute(select(FlowVersion).where(FlowVersion.id == flow.published_version_id)) published = result.scalar_one_or_none() if published: return json.loads(json.dumps(published.definition_json)) return flow.definition_json @router.post("/definitions/{flow_id}/execute") async def execute_flow(flow_id: uuid.UUID, request: Request, payload: dict, db: AsyncSession = Depends(get_db)): f = await db.get(FlowDefinition, flow_id) if not f: raise HTTPException(404, "流定义不存在") user_ctx = request.state.user input_text = payload.get("input", payload.get("message", "")) definition = await _get_published_definition(f, db) await ToolNodeAgent.load_custom_tools(db) engine = FlowEngine(definition) input_msg = Msg(name="user", content=input_text, role="user") context = { "user_id": user_ctx["id"], "username": user_ctx.get("username", ""), "trigger_data": payload.get("trigger", {}), "_node_results": {}, } start_time = time.time() try: result_msg = await engine.execute(input_msg, context) elapsed_ms = int((time.time() - start_time) * 1000) output_text = result_msg.get_text_content() if hasattr(result_msg, 'get_text_content') else str(result_msg) execution = FlowExecution( flow_id=f.id, version=_get_published_version_number(f), trigger_type=payload.get("trigger_type", "manual"), trigger_user_id=uuid.UUID(user_ctx["id"]), input_data={"input": input_text}, output_data={"output": output_text}, status="completed", latency_ms=elapsed_ms, finished_at=datetime.utcnow(), ) db.add(execution) return { "code": 200, "data": { "output": output_text, "node_results": context.get("_node_results", {}), "execution_id": str(execution.id), "latency_ms": elapsed_ms, }, } except Exception as e: elapsed_ms = int((time.time() - start_time) * 1000) execution = FlowExecution( flow_id=f.id, version=_get_published_version_number(f), trigger_type="manual", trigger_user_id=uuid.UUID(user_ctx["id"]), input_data={"input": input_text}, status="failed", latency_ms=elapsed_ms, error_message=str(e)[:2000], finished_at=datetime.utcnow(), ) db.add(execution) raise HTTPException(500, f"流执行失败: {str(e)}") def _get_published_version_number(flow: FlowDefinition) -> int | None: return flow.version @router.post("/definitions/{flow_id}/test") async def test_flow(flow_id: uuid.UUID, request: Request, db: AsyncSession = Depends(get_db)): f = await db.get(FlowDefinition, flow_id) if not f: raise HTTPException(404, "流定义不存在") definition = await _get_published_definition(f, db) nodes = definition.get("nodes", []) edges = definition.get("edges", []) validation = {"valid": True, "node_count": len(nodes), "edge_count": len(edges), "node_types": list(set(n.get("type", "unknown") for n in nodes)), "issues": []} node_ids = {n["id"] for n in nodes} for edge in edges: s = edge.get("source") or edge.get("from") t = edge.get("target") or edge.get("to") if s and s not in node_ids: validation["issues"].append(f"边源节点 {s} 不存在") if t and t not in node_ids: validation["issues"].append(f"边目标节点 {t} 不存在") if validation["issues"]: validation["valid"] = False if not any(n.get("type") == "trigger" for n in nodes): validation["issues"].append("流缺少触发节点") return {"code": 200, "data": validation} # ============================== SSE 流式执行 ============================== @router.post("/definitions/{flow_id}/stream") async def execute_flow_stream(flow_id: uuid.UUID, request: Request, db: AsyncSession = Depends(get_db)): body = await request.json() input_text = body.get("input", body.get("message", "")) user_ctx = request.state.user f = await db.get(FlowDefinition, flow_id) if not f: raise HTTPException(404, "流定义不存在") definition = await _get_published_definition(f, db) await ToolNodeAgent.load_custom_tools(db) async def event_generator(): import asyncio engine = FlowEngine(definition) context = { "user_id": user_ctx["id"], "username": user_ctx.get("username", ""), "trigger_data": body.get("trigger", {}), "_node_results": {}, "_stream_callback": None, } input_msg = Msg(name="user", content=input_text, role="user") start_time = time.time() # bind stream callback stream_chunks = [] async def stream_callback(event_type: str, data: dict): chunk_data = json.dumps({"event": event_type, "data": data}, ensure_ascii=False) stream_chunks.append(f"data: {chunk_data}\n\n") context["_stream_callback"] = stream_callback try: yield f"data: {json.dumps({'event': 'workflow_started', 'data': {'flow_id': str(flow_id)}}, ensure_ascii=False)}\n\n" # get execution order graph = engine._build_graph() start = engine._find_start_nodes(graph) if start: yield f"data: {json.dumps({'event': 'node_started', 'data': {'node_id': start[0], 'node_type': definition.get('nodes', [{}])[0].get('type', 'unknown'), 'label': definition.get('nodes', [{}])[0].get('label', '开始')}}, ensure_ascii=False)}\n\n" result_msg = await asyncio.wait_for( engine.execute(input_msg, context), timeout=engine.FLOW_TIMEOUT_SECONDS, ) output_text = result_msg.get_text_content() if hasattr(result_msg, 'get_text_content') else str(result_msg) elapsed_ms = int((time.time() - start_time) * 1000) yield f"data: {json.dumps({'event': 'text_chunk', 'data': {'content': output_text}}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'event': 'workflow_finished', 'data': {'output': output_text, 'node_results': {k: str(v)[:200] for k, v in context.get('_node_results', {}).items()}, 'latency_ms': elapsed_ms}}, ensure_ascii=False)}\n\n" execution = FlowExecution( flow_id=f.id, version=_get_published_version_number(f), trigger_type=body.get("trigger_type", "manual"), trigger_user_id=uuid.UUID(user_ctx["id"]), input_data={"input": input_text}, output_data={"output": output_text}, status="completed", latency_ms=elapsed_ms, finished_at=datetime.utcnow(), ) db.add(execution) except asyncio.TimeoutError: elapsed_ms = int((time.time() - start_time) * 1000) yield f"data: {json.dumps({'event': 'error', 'data': {'message': '执行超时'}}, ensure_ascii=False)}\n\n" execution = FlowExecution( flow_id=f.id, version=_get_published_version_number(f), trigger_type="manual", trigger_user_id=uuid.UUID(user_ctx["id"]), input_data={"input": input_text}, status="failed", latency_ms=elapsed_ms, error_message="执行超时", finished_at=datetime.utcnow(), ) db.add(execution) except Exception as e: elapsed_ms = int((time.time() - start_time) * 1000) yield f"data: {json.dumps({'event': 'error', 'data': {'message': str(e)}}, ensure_ascii=False)}\n\n" execution = FlowExecution( flow_id=f.id, version=_get_published_version_number(f), trigger_type="manual", trigger_user_id=uuid.UUID(user_ctx["id"]), input_data={"input": input_text}, status="failed", latency_ms=elapsed_ms, error_message=str(e)[:2000], finished_at=datetime.utcnow(), ) db.add(execution) finally: yield "data: [DONE]\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", }, ) # ============================== API Key 管理 ============================== def _generate_api_key() -> tuple[str, str, str]: raw = "flow-" + secrets.token_urlsafe(24) key_hash = hashlib.sha256(raw.encode()).hexdigest() return raw, key_hash, raw[:14] @router.post("/definitions/{flow_id}/api-keys", response_model=dict) async def create_api_key(flow_id: uuid.UUID, body: FlowApiKeyCreate, request: Request, db: AsyncSession = Depends(get_db)): f = await db.get(FlowDefinition, flow_id) if not f: raise HTTPException(404, "流定义不存在") user_ctx = request.state.user raw, key_hash, key_prefix = _generate_api_key() api_key = FlowApiKey( flow_id=flow_id, name=body.name, key_hash=key_hash, key_prefix=key_prefix, created_by=uuid.UUID(user_ctx["id"]), ) db.add(api_key) await db.flush() return { "code": 200, "data": { "id": str(api_key.id), "name": body.name, "key_prefix": key_prefix, "api_key": raw, "created_at": str(api_key.created_at), }, } @router.get("/definitions/{flow_id}/api-keys", response_model=list[FlowApiKeyOut]) async def list_api_keys(flow_id: uuid.UUID, db: AsyncSession = Depends(get_db)): result = await db.execute( select(FlowApiKey).where(FlowApiKey.flow_id == flow_id).order_by(FlowApiKey.created_at.desc()) ) return [FlowApiKeyOut( id=k.id, flow_id=k.flow_id, name=k.name, key_prefix=k.key_prefix, last_used_at=k.last_used_at, created_at=k.created_at, ) for k in result.scalars().all()] @router.delete("/api-keys/{key_id}") async def delete_api_key(key_id: uuid.UUID, db: AsyncSession = Depends(get_db)): k = await db.get(FlowApiKey, key_id) if not k: raise HTTPException(404, "API Key不存在") await db.delete(k) return {"code": 200, "message": "API Key已删除"} # ============================== 执行历史 ============================== @router.get("/executions") async def list_executions( db: AsyncSession = Depends(get_db), flow_id: str | None = Query(None), page: int = Query(1), page_size: int = Query(20), ): query = select(FlowExecution).order_by(FlowExecution.started_at.desc()) if flow_id: query = query.where(FlowExecution.flow_id == uuid.UUID(flow_id)) total_result = await db.execute(query) total = len(total_result.scalars().all()) result = await db.execute(query.offset((page - 1) * page_size).limit(page_size)) executions = result.scalars().all() return { "code": 200, "data": [{ "id": str(e.id), "flow_id": str(e.flow_id), "version": e.version, "trigger_type": e.trigger_type, "status": e.status, "latency_ms": e.latency_ms, "token_usage": e.token_usage, "error_message": e.error_message, "started_at": str(e.started_at), "finished_at": str(e.finished_at) if e.finished_at else None, } for e in executions], "total": total, "page": page, "page_size": page_size, } # ============================== 模板 ============================== FLOW_TEMPLATES = [ { "id": "tpl_doc_process", "name": "文档处理流", "description": "自动解析文档内容，提取关键信息并生成摘要", "icon": "Document", "nodes": [ {"id": "n1", "type": "trigger", "label": "文档上传", "config": {"event_type": "document_upload"}, "position": {"x": 100, "y": 100}}, {"id": "n2", "type": "tool", "label": "解析文档", "config": {"tool_name": "parse_document"}, "position": {"x": 400, "y": 100}}, {"id": "n3", "type": "llm", "label": "生成摘要", "config": {"system_prompt": "请为以下文档内容生成简洁摘要", "model": "gpt-4o-mini", "temperature": 0.5}, "position": {"x": 700, "y": 100}}, {"id": "n4", "type": "output", "label": "输出结果", "config": {"format": "text"}, "position": {"x": 1000, "y": 100}}, ], "edges": [ {"source": "n1", "target": "n2", "sourceHandle": "source"}, {"source": "n2", "target": "n3", "sourceHandle": "source"}, {"source": "n3", "target": "n4", "sourceHandle": "source"}, ], }, { "id": "tpl_wecom_notify", "name": "企微通知流", "description": "接收触发后查询数据并推送企微通知", "icon": "Bell", "nodes": [ {"id": "n1", "type": "trigger", "label": "定时触发", "config": {"event_type": "scheduled"}, "position": {"x": 100, "y": 100}}, {"id": "n2", "type": "tool", "label": "查询任务", "config": {"tool_name": "list_tasks"}, "position": {"x": 400, "y": 100}}, {"id": "n3", "type": "condition", "label": "有待办任务?", "config": {"condition": "tasks.length > 0"}, "position": {"x": 700, "y": 100}}, {"id": "n4", "type": "wecom_notify", "label": "推送通知", "config": {"message_template": "您有{{tasks.length}}条待办任务", "target": "@all"}, "position": {"x": 1000, "y": 50}}, {"id": "n5", "type": "output", "label": "无任务", "config": {"format": "text"}, "position": {"x": 1000, "y": 200}}, ], "edges": [ {"source": "n1", "target": "n2", "sourceHandle": "source"}, {"source": "n2", "target": "n3", "sourceHandle": "source"}, {"source": "n3", "target": "n4", "sourceHandle": "true"}, {"source": "n3", "target": "n5", "sourceHandle": "false"}, ], }, { "id": "tpl_data_analysis", "name": "数据分析流", "description": "查询员工数据并生成效率分析报告", "icon": "DataAnalysis", "nodes": [ {"id": "n1", "type": "trigger", "label": "分析请求", "config": {"event_type": "button_click"}, "position": {"x": 100, "y": 100}}, {"id": "n2", "type": "tool", "label": "查询下属", "config": {"tool_name": "list_subordinates"}, "position": {"x": 400, "y": 100}}, {"id": "n3", "type": "tool", "label": "统计数据", "config": {"tool_name": "get_task_statistics"}, "position": {"x": 700, "y": 100}}, {"id": "n4", "type": "llm", "label": "生成报告", "config": {"system_prompt": "基于以下数据生成团队效率分析报告", "model": "gpt-4o", "temperature": 0.7}, "position": {"x": 1000, "y": 100}}, {"id": "n5", "type": "output", "label": "报告输出", "config": {"format": "json"}, "position": {"x": 1300, "y": 100}}, ], "edges": [ {"source": "n1", "target": "n2", "sourceHandle": "source"}, {"source": "n2", "target": "n3", "sourceHandle": "source"}, {"source": "n3", "target": "n4", "sourceHandle": "source"}, {"source": "n4", "target": "n5", "sourceHandle": "source"}, ], }, { "id": "tpl_rag_qa", "name": "知识库问答流", "description": "从知识库检索信息后由LLM回答", "icon": "Search", "nodes": [ {"id": "n1", "type": "trigger", "label": "问题触发", "config": {"event_type": "text_message"}, "position": {"x": 100, "y": 100}}, {"id": "n2", "type": "rag", "label": "知识检索", "config": {"knowledge_base": "default", "top_k": 5}, "position": {"x": 400, "y": 100}}, {"id": "n3", "type": "llm", "label": "生成回答", "config": {"system_prompt": "基于知识库检索结果回答用户问题", "model": "gpt-4o-mini", "temperature": 0.3}, "position": {"x": 700, "y": 100}}, {"id": "n4", "type": "output", "label": "输出答案", "config": {"format": "text"}, "position": {"x": 1000, "y": 100}}, ], "edges": [ {"source": "n1", "target": "n2", "sourceHandle": "source"}, {"source": "n2", "target": "n3", "sourceHandle": "source"}, {"source": "n3", "target": "n4", "sourceHandle": "source"}, ], }, { "id": "tpl_task_auto", "name": "任务自动分配流", "description": "根据描述自动创建任务并分派给合适人员", "icon": "Tools", "nodes": [ {"id": "n1", "type": "trigger", "label": "任务描述", "config": {"event_type": "text_message"}, "position": {"x": 100, "y": 100}}, {"id": "n2", "type": "llm", "label": "分析任务", "config": {"system_prompt": "分析以下任务描述，提取标题、优先级、负责人", "model": "gpt-4o-mini", "temperature": 0.5}, "position": {"x": 400, "y": 100}}, {"id": "n3", "type": "tool", "label": "创建任务", "config": {"tool_name": "create_task"}, "position": {"x": 700, "y": 100}}, {"id": "n4", "type": "wecom_notify", "label": "通知负责人", "config": {"message_template": "您有新任务: {{task_title}}", "target": "@all"}, "position": {"x": 1000, "y": 100}}, {"id": "n5", "type": "output", "label": "完成", "config": {"format": "text"}, "position": {"x": 1300, "y": 100}}, ], "edges": [ {"source": "n1", "target": "n2", "sourceHandle": "source"}, {"source": "n2", "target": "n3", "sourceHandle": "source"}, {"source": "n3", "target": "n4", "sourceHandle": "source"}, {"source": "n4", "target": "n5", "sourceHandle": "source"}, ], }, ] @router.get("/market", response_model=list[FlowDefinitionOut]) async def flow_market(db: AsyncSession = Depends(get_db)): result = await db.execute( select(FlowDefinition).where(FlowDefinition.status == "published").order_by(FlowDefinition.updated_at.desc()) ) return [_build_flow_out(f) for f in result.scalars().all()] @router.get("/templates") async def get_flow_templates(request: Request): return {"code": 200, "data": FLOW_TEMPLATES} @router.post("/templates/{template_id}/use") async def use_flow_template(template_id: str, request: Request, db: AsyncSession = Depends(get_db)): template = next((t for t in FLOW_TEMPLATES if t["id"] == template_id), None) if not template: raise HTTPException(404, "模板不存在") user_ctx = request.state.user definition_json = {"nodes": template["nodes"], "edges": template["edges"], "trigger": {}} flow = FlowDefinition( name=template["name"] + " (副本)", description=template["description"], definition_json=definition_json, draft_definition_json=definition_json, creator_id=uuid.UUID(user_ctx["id"]), ) db.add(flow) await db.flush() return _build_flow_out(flow)