workflow.py 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508
  1. import json
  2. import logging
  3. from flask import abort, request
  4. from flask_restful import Resource, marshal_with, reqparse
  5. from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
  6. import services
  7. from controllers.console import api
  8. from controllers.console.app.error import ConversationCompletedError, DraftWorkflowNotExist, DraftWorkflowNotSync
  9. from controllers.console.app.wraps import get_app_model
  10. from controllers.console.setup import setup_required
  11. from controllers.console.wraps import account_initialization_required
  12. from core.app.apps.base_app_queue_manager import AppQueueManager
  13. from core.app.entities.app_invoke_entities import InvokeFrom
  14. from core.app.segments import factory
  15. from core.errors.error import AppInvokeQuotaExceededError
  16. from fields.workflow_fields import workflow_fields
  17. from fields.workflow_run_fields import workflow_run_node_execution_fields
  18. from libs import helper
  19. from libs.helper import TimestampField, uuid_value
  20. from libs.login import current_user, login_required
  21. from models.account import Account
  22. from models.model import App, AppMode
  23. from services.app_dsl_service import AppDslService
  24. from services.app_generate_service import AppGenerateService
  25. from services.errors.app import WorkflowHashNotEqualError
  26. from services.workflow_service import WorkflowService
  27. logger = logging.getLogger(__name__)
  28. class DraftWorkflowApi(Resource):
  29. @setup_required
  30. @login_required
  31. @account_initialization_required
  32. @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
  33. @marshal_with(workflow_fields)
  34. def get(self, app_model: App):
  35. """
  36. Get draft workflow
  37. """
  38. # The role of the current user in the ta table must be admin, owner, or editor
  39. if not current_user.is_editor:
  40. raise Forbidden()
  41. # fetch draft workflow by app_model
  42. workflow_service = WorkflowService()
  43. workflow = workflow_service.get_draft_workflow(app_model=app_model)
  44. if not workflow:
  45. raise DraftWorkflowNotExist()
  46. # return workflow, if not found, return None (initiate graph by frontend)
  47. return workflow
  48. @setup_required
  49. @login_required
  50. @account_initialization_required
  51. @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
  52. def post(self, app_model: App):
  53. """
  54. Sync draft workflow
  55. """
  56. # The role of the current user in the ta table must be admin, owner, or editor
  57. if not current_user.is_editor:
  58. raise Forbidden()
  59. content_type = request.headers.get("Content-Type", "")
  60. if "application/json" in content_type:
  61. parser = reqparse.RequestParser()
  62. parser.add_argument("graph", type=dict, required=True, nullable=False, location="json")
  63. parser.add_argument("features", type=dict, required=True, nullable=False, location="json")
  64. parser.add_argument("hash", type=str, required=False, location="json")
  65. # TODO: set this to required=True after frontend is updated
  66. parser.add_argument("environment_variables", type=list, required=False, location="json")
  67. parser.add_argument("conversation_variables", type=list, required=False, location="json")
  68. args = parser.parse_args()
  69. elif "text/plain" in content_type:
  70. try:
  71. data = json.loads(request.data.decode("utf-8"))
  72. if "graph" not in data or "features" not in data:
  73. raise ValueError("graph or features not found in data")
  74. if not isinstance(data.get("graph"), dict) or not isinstance(data.get("features"), dict):
  75. raise ValueError("graph or features is not a dict")
  76. args = {
  77. "graph": data.get("graph"),
  78. "features": data.get("features"),
  79. "hash": data.get("hash"),
  80. "environment_variables": data.get("environment_variables"),
  81. "conversation_variables": data.get("conversation_variables"),
  82. }
  83. except json.JSONDecodeError:
  84. return {"message": "Invalid JSON data"}, 400
  85. else:
  86. abort(415)
  87. if not isinstance(current_user, Account):
  88. raise Forbidden()
  89. workflow_service = WorkflowService()
  90. try:
  91. environment_variables_list = args.get("environment_variables") or []
  92. environment_variables = [factory.build_variable_from_mapping(obj) for obj in environment_variables_list]
  93. conversation_variables_list = args.get("conversation_variables") or []
  94. conversation_variables = [factory.build_variable_from_mapping(obj) for obj in conversation_variables_list]
  95. workflow = workflow_service.sync_draft_workflow(
  96. app_model=app_model,
  97. graph=args["graph"],
  98. features=args["features"],
  99. unique_hash=args.get("hash"),
  100. account=current_user,
  101. environment_variables=environment_variables,
  102. conversation_variables=conversation_variables,
  103. )
  104. except WorkflowHashNotEqualError:
  105. raise DraftWorkflowNotSync()
  106. return {
  107. "result": "success",
  108. "hash": workflow.unique_hash,
  109. "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
  110. }
  111. class DraftWorkflowImportApi(Resource):
  112. @setup_required
  113. @login_required
  114. @account_initialization_required
  115. @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
  116. @marshal_with(workflow_fields)
  117. def post(self, app_model: App):
  118. """
  119. Import draft workflow
  120. """
  121. # The role of the current user in the ta table must be admin, owner, or editor
  122. if not current_user.is_editor:
  123. raise Forbidden()
  124. if not isinstance(current_user, Account):
  125. raise Forbidden()
  126. parser = reqparse.RequestParser()
  127. parser.add_argument("data", type=str, required=True, nullable=False, location="json")
  128. args = parser.parse_args()
  129. workflow = AppDslService.import_and_overwrite_workflow(
  130. app_model=app_model, data=args["data"], account=current_user
  131. )
  132. return workflow
  133. class AdvancedChatDraftWorkflowRunApi(Resource):
  134. @setup_required
  135. @login_required
  136. @account_initialization_required
  137. @get_app_model(mode=[AppMode.ADVANCED_CHAT])
  138. def post(self, app_model: App):
  139. """
  140. Run draft workflow
  141. """
  142. # The role of the current user in the ta table must be admin, owner, or editor
  143. if not current_user.is_editor:
  144. raise Forbidden()
  145. if not isinstance(current_user, Account):
  146. raise Forbidden()
  147. parser = reqparse.RequestParser()
  148. parser.add_argument("inputs", type=dict, location="json")
  149. parser.add_argument("query", type=str, required=True, location="json", default="")
  150. parser.add_argument("files", type=list, location="json")
  151. parser.add_argument("conversation_id", type=uuid_value, location="json")
  152. args = parser.parse_args()
  153. try:
  154. response = AppGenerateService.generate(
  155. app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.DEBUGGER, streaming=True
  156. )
  157. return helper.compact_generate_response(response)
  158. except services.errors.conversation.ConversationNotExistsError:
  159. raise NotFound("Conversation Not Exists.")
  160. except services.errors.conversation.ConversationCompletedError:
  161. raise ConversationCompletedError()
  162. except ValueError as e:
  163. raise e
  164. except Exception as e:
  165. logging.exception("internal server error.")
  166. raise InternalServerError()
  167. class AdvancedChatDraftRunIterationNodeApi(Resource):
  168. @setup_required
  169. @login_required
  170. @account_initialization_required
  171. @get_app_model(mode=[AppMode.ADVANCED_CHAT])
  172. def post(self, app_model: App, node_id: str):
  173. """
  174. Run draft workflow iteration node
  175. """
  176. # The role of the current user in the ta table must be admin, owner, or editor
  177. if not current_user.is_editor:
  178. raise Forbidden()
  179. if not isinstance(current_user, Account):
  180. raise Forbidden()
  181. parser = reqparse.RequestParser()
  182. parser.add_argument("inputs", type=dict, location="json")
  183. args = parser.parse_args()
  184. try:
  185. response = AppGenerateService.generate_single_iteration(
  186. app_model=app_model, user=current_user, node_id=node_id, args=args, streaming=True
  187. )
  188. return helper.compact_generate_response(response)
  189. except services.errors.conversation.ConversationNotExistsError:
  190. raise NotFound("Conversation Not Exists.")
  191. except services.errors.conversation.ConversationCompletedError:
  192. raise ConversationCompletedError()
  193. except ValueError as e:
  194. raise e
  195. except Exception as e:
  196. logging.exception("internal server error.")
  197. raise InternalServerError()
  198. class WorkflowDraftRunIterationNodeApi(Resource):
  199. @setup_required
  200. @login_required
  201. @account_initialization_required
  202. @get_app_model(mode=[AppMode.WORKFLOW])
  203. def post(self, app_model: App, node_id: str):
  204. """
  205. Run draft workflow iteration node
  206. """
  207. # The role of the current user in the ta table must be admin, owner, or editor
  208. if not current_user.is_editor:
  209. raise Forbidden()
  210. if not isinstance(current_user, Account):
  211. raise Forbidden()
  212. parser = reqparse.RequestParser()
  213. parser.add_argument("inputs", type=dict, location="json")
  214. args = parser.parse_args()
  215. try:
  216. response = AppGenerateService.generate_single_iteration(
  217. app_model=app_model, user=current_user, node_id=node_id, args=args, streaming=True
  218. )
  219. return helper.compact_generate_response(response)
  220. except services.errors.conversation.ConversationNotExistsError:
  221. raise NotFound("Conversation Not Exists.")
  222. except services.errors.conversation.ConversationCompletedError:
  223. raise ConversationCompletedError()
  224. except ValueError as e:
  225. raise e
  226. except Exception as e:
  227. logging.exception("internal server error.")
  228. raise InternalServerError()
  229. class DraftWorkflowRunApi(Resource):
  230. @setup_required
  231. @login_required
  232. @account_initialization_required
  233. @get_app_model(mode=[AppMode.WORKFLOW])
  234. def post(self, app_model: App):
  235. """
  236. Run draft workflow
  237. """
  238. # The role of the current user in the ta table must be admin, owner, or editor
  239. if not current_user.is_editor:
  240. raise Forbidden()
  241. if not isinstance(current_user, Account):
  242. raise Forbidden()
  243. parser = reqparse.RequestParser()
  244. parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
  245. parser.add_argument("files", type=list, required=False, location="json")
  246. args = parser.parse_args()
  247. try:
  248. response = AppGenerateService.generate(
  249. app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.DEBUGGER, streaming=True
  250. )
  251. return helper.compact_generate_response(response)
  252. except (ValueError, AppInvokeQuotaExceededError) as e:
  253. raise e
  254. except Exception as e:
  255. logging.exception("internal server error.")
  256. raise InternalServerError()
  257. class WorkflowTaskStopApi(Resource):
  258. @setup_required
  259. @login_required
  260. @account_initialization_required
  261. @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
  262. def post(self, app_model: App, task_id: str):
  263. """
  264. Stop workflow task
  265. """
  266. # The role of the current user in the ta table must be admin, owner, or editor
  267. if not current_user.is_editor:
  268. raise Forbidden()
  269. AppQueueManager.set_stop_flag(task_id, InvokeFrom.DEBUGGER, current_user.id)
  270. return {"result": "success"}
  271. class DraftWorkflowNodeRunApi(Resource):
  272. @setup_required
  273. @login_required
  274. @account_initialization_required
  275. @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
  276. @marshal_with(workflow_run_node_execution_fields)
  277. def post(self, app_model: App, node_id: str):
  278. """
  279. Run draft workflow node
  280. """
  281. # The role of the current user in the ta table must be admin, owner, or editor
  282. if not current_user.is_editor:
  283. raise Forbidden()
  284. if not isinstance(current_user, Account):
  285. raise Forbidden()
  286. parser = reqparse.RequestParser()
  287. parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
  288. args = parser.parse_args()
  289. inputs = args.get("inputs")
  290. if inputs == None:
  291. raise ValueError("missing inputs")
  292. workflow_service = WorkflowService()
  293. workflow_node_execution = workflow_service.run_draft_workflow_node(
  294. app_model=app_model, node_id=node_id, user_inputs=inputs, account=current_user
  295. )
  296. return workflow_node_execution
  297. class PublishedWorkflowApi(Resource):
  298. @setup_required
  299. @login_required
  300. @account_initialization_required
  301. @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
  302. @marshal_with(workflow_fields)
  303. def get(self, app_model: App):
  304. """
  305. Get published workflow
  306. """
  307. # The role of the current user in the ta table must be admin, owner, or editor
  308. if not current_user.is_editor:
  309. raise Forbidden()
  310. # fetch published workflow by app_model
  311. workflow_service = WorkflowService()
  312. workflow = workflow_service.get_published_workflow(app_model=app_model)
  313. # return workflow, if not found, return None
  314. return workflow
  315. @setup_required
  316. @login_required
  317. @account_initialization_required
  318. @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
  319. def post(self, app_model: App):
  320. """
  321. Publish workflow
  322. """
  323. # The role of the current user in the ta table must be admin, owner, or editor
  324. if not current_user.is_editor:
  325. raise Forbidden()
  326. if not isinstance(current_user, Account):
  327. raise Forbidden()
  328. workflow_service = WorkflowService()
  329. workflow = workflow_service.publish_workflow(app_model=app_model, account=current_user)
  330. return {"result": "success", "created_at": TimestampField().format(workflow.created_at)}
  331. class DefaultBlockConfigsApi(Resource):
  332. @setup_required
  333. @login_required
  334. @account_initialization_required
  335. @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
  336. def get(self, app_model: App):
  337. """
  338. Get default block config
  339. """
  340. # The role of the current user in the ta table must be admin, owner, or editor
  341. if not current_user.is_editor:
  342. raise Forbidden()
  343. # Get default block configs
  344. workflow_service = WorkflowService()
  345. return workflow_service.get_default_block_configs()
  346. class DefaultBlockConfigApi(Resource):
  347. @setup_required
  348. @login_required
  349. @account_initialization_required
  350. @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
  351. def get(self, app_model: App, block_type: str):
  352. """
  353. Get default block config
  354. """
  355. # The role of the current user in the ta table must be admin, owner, or editor
  356. if not current_user.is_editor:
  357. raise Forbidden()
  358. if not isinstance(current_user, Account):
  359. raise Forbidden()
  360. parser = reqparse.RequestParser()
  361. parser.add_argument("q", type=str, location="args")
  362. args = parser.parse_args()
  363. q = args.get("q")
  364. filters = None
  365. if q:
  366. try:
  367. filters = json.loads(q)
  368. except json.JSONDecodeError:
  369. raise ValueError("Invalid filters")
  370. # Get default block configs
  371. workflow_service = WorkflowService()
  372. return workflow_service.get_default_block_config(node_type=block_type, filters=filters)
  373. class ConvertToWorkflowApi(Resource):
  374. @setup_required
  375. @login_required
  376. @account_initialization_required
  377. @get_app_model(mode=[AppMode.CHAT, AppMode.COMPLETION])
  378. def post(self, app_model: App):
  379. """
  380. Convert basic mode of chatbot app to workflow mode
  381. Convert expert mode of chatbot app to workflow mode
  382. Convert Completion App to Workflow App
  383. """
  384. # The role of the current user in the ta table must be admin, owner, or editor
  385. if not current_user.is_editor:
  386. raise Forbidden()
  387. if not isinstance(current_user, Account):
  388. raise Forbidden()
  389. if request.data:
  390. parser = reqparse.RequestParser()
  391. parser.add_argument("name", type=str, required=False, nullable=True, location="json")
  392. parser.add_argument("icon_type", type=str, required=False, nullable=True, location="json")
  393. parser.add_argument("icon", type=str, required=False, nullable=True, location="json")
  394. parser.add_argument("icon_background", type=str, required=False, nullable=True, location="json")
  395. args = parser.parse_args()
  396. else:
  397. args = {}
  398. # convert to workflow mode
  399. workflow_service = WorkflowService()
  400. new_app_model = workflow_service.convert_to_workflow(app_model=app_model, account=current_user, args=args)
  401. # return app id
  402. return {
  403. "new_app_id": new_app_model.id,
  404. }
  405. api.add_resource(DraftWorkflowApi, "/apps/<uuid:app_id>/workflows/draft")
  406. api.add_resource(DraftWorkflowImportApi, "/apps/<uuid:app_id>/workflows/draft/import")
  407. api.add_resource(AdvancedChatDraftWorkflowRunApi, "/apps/<uuid:app_id>/advanced-chat/workflows/draft/run")
  408. api.add_resource(DraftWorkflowRunApi, "/apps/<uuid:app_id>/workflows/draft/run")
  409. api.add_resource(WorkflowTaskStopApi, "/apps/<uuid:app_id>/workflow-runs/tasks/<string:task_id>/stop")
  410. api.add_resource(DraftWorkflowNodeRunApi, "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/run")
  411. api.add_resource(
  412. AdvancedChatDraftRunIterationNodeApi,
  413. "/apps/<uuid:app_id>/advanced-chat/workflows/draft/iteration/nodes/<string:node_id>/run",
  414. )
  415. api.add_resource(
  416. WorkflowDraftRunIterationNodeApi, "/apps/<uuid:app_id>/workflows/draft/iteration/nodes/<string:node_id>/run"
  417. )
  418. api.add_resource(PublishedWorkflowApi, "/apps/<uuid:app_id>/workflows/publish")
  419. api.add_resource(DefaultBlockConfigsApi, "/apps/<uuid:app_id>/workflows/default-workflow-block-configs")
  420. api.add_resource(
  421. DefaultBlockConfigApi, "/apps/<uuid:app_id>/workflows/default-workflow-block-configs/<string:block_type>"
  422. )
  423. api.add_resource(ConvertToWorkflowApi, "/apps/<uuid:app_id>/convert-to-workflow")