feat: 审批流程跑通

app/controller/add_approve_route.py

@@ -1,6 +1,7 @@
 from fastapi import FastAPI
 from langchain_core.runnables import RunnableConfig
-from langgraph.func import entrypoint
+from langgraph.constants import START
+from langgraph.graph import StateGraph
 from langgraph.types import Command
 from pydantic import BaseModel, Field
 
@@ -8,7 +9,7 @@ from app.model.ApproveModel import ApproveService
 from app.model.ReimburseModel import ReimburseService, ReimburseModel
 from app.utils.db_utils import AsyncSessionDep
 from app.utils.postgres_checkpointer import AsyncPostgresSaverDep
-from app.workflow.approve import create_approve_workflow, ApproveWorkflowInputs, ApproveResult
+from app.workflow.approve import ApproveResult, create_approve_graph, ApproveGraphSchema
 
 
 def add_approve_route(app: FastAPI):
@@ -54,14 +55,14 @@ def add_approve_route(app: FastAPI):
     # 将审批单与报销单管理，设置报销单的approve_id为审批单的id
     await ReimburseService.item_update(session=session, row_dict={"id": reimburse_id, "approve_id": insert_approve_cls.id}, )
 
-    reimburse_workflow = create_reimburse_workflow(session=session, checkpointer=checkpointer)
+    reimburse_graph = create_reimburse_graph(session=session, checkpointer=checkpointer)
 
-    await reimburse_workflow.ainvoke(
+    graph_state = await reimburse_graph.ainvoke(
-      {"reimburse_user_id": reimburse_cls.user.id, "approve_id": insert_approve_cls.id},
+      {"input_user_id": reimburse_cls.user.id, "input_approve_id": insert_approve_cls.id},
       config={"configurable": {"thread_id": insert_approve_cls.id}}
     )
 
-    return {"message": "报销单提交成功！"}
+    return {"message": "报销单提交成功！", "graph_state": graph_state}
 
   class ProcessApproveClass(BaseModel):
     flag: bool = Field(..., description="审批标识，是审批通过还是审批驳回")
@@ -76,26 +77,40 @@ def add_approve_route(app: FastAPI):
     session: AsyncSessionDep,
     checkpointer: AsyncPostgresSaverDep,
   ):
-    reimburse_workflow = create_reimburse_workflow(session=session, checkpointer=checkpointer)
+    reimburse_graph = create_reimburse_graph(session=session, checkpointer=checkpointer)
     approve_result: ApproveResult = {
       "flag": body.flag,
       "reason": body.reason,
       "user_id": body.user_id,
     }
-    await reimburse_workflow.ainvoke(Command(resume=approve_result), config={"configurable": {"thread_id": body.approve_id}})
+
+    print('\n\n', ":::::::::::process_approve start::::::::::::", '\n\n')
+    print(approve_result)
+    print('\n\n', ":::::::::::process_approve end::::::::::::", '\n\n')
+
+    return await reimburse_graph.ainvoke(Command(resume=approve_result), config={"configurable": {"thread_id": body.approve_id}})
 
 
-def create_reimburse_workflow(
+def create_reimburse_graph(
   checkpointer: AsyncPostgresSaverDep,
   session: AsyncSessionDep,
 ):
-  approve_workflow = create_approve_workflow(checkpointer, session)
+  approve_graph = create_approve_graph(checkpointer, session)
 
-  @entrypoint(checkpointer=checkpointer)
+  builder = StateGraph(ApproveGraphSchema)
-  async def reimburse_workflow(input_dict: ApproveWorkflowInputs, config: RunnableConfig):
+
-    approve_flag = await approve_workflow.ainvoke(input_dict, config=config)
+  async def node(state: ApproveGraphSchema, config: RunnableConfig):
+    print('\n\n', ":::::::::::node start::::::::::::", '\n\n')
+    print(state)
+    print('\n\n', ":::::::::::node end::::::::::::", '\n\n')
+
+    graph_state = await approve_graph.ainvoke(state, config=config)
+    approve_flag = graph_state.get('approve_flag')
     # 这里审批结束之后可以做一些事情，比如发送邮件、短信、微信消息等通知用户，因为没有实现对应的模块，这里就不做任何处理
     print("报销单审批流执行结束::::::::::>>>>>>>>>>>", approve_flag)
-    return None
+    return {}
 
-  return reimburse_workflow
+  builder.add_node("node", node)
+  builder.add_edge(START, "node")
+
+  return builder.compile(checkpointer=checkpointer)

app/workflow/approve.py

@@ -3,8 +3,9 @@ import json
 from typing import List
 from typing import TypedDict, Union
 
-from langgraph.func import entrypoint, task
+from langgraph.constants import START
-from langgraph.types import interrupt
+from langgraph.graph import StateGraph
+from langgraph.types import interrupt, Command
 
 from app.model.ApproveModel import ApproveService
 from app.model.UserModel import UserService, UserServiceModel
@@ -13,89 +14,128 @@ from app.utils.postgres_checkpointer import AsyncPostgresSaverDep
 
 
 # 创建审批工作流
-def create_approve_workflow(
+def create_approve_graph(
   checkpointer: AsyncPostgresSaverDep,
   session: AsyncSessionDep,
 ):
-  @entrypoint(checkpointer=checkpointer)
+  builder = StateGraph(ApproveGraphSchema)
-  async def approve_workflow(input_dict: ApproveWorkflowInputs) -> bool:
-    # 申请人id
-    reimburse_user_id = input_dict.get('reimburse_user_id')
-    # 审批单id
-    approve_id = input_dict.get('approve_id')
 
-    # 立即拟造一条审批结果，审批人为申请人自己，审批结果为通过
+  async def node_first(state: ApproveGraphSchema):
-    approve_result: ApproveResult = {"flag": True, "user_id": reimburse_user_id, }
+    supervisor_user = await aget_supervisor_user(user_id=state.get('input_user_id'), session=session)
 
-    # 循环判断审批结果，如果为通过，则继续触发下一个上级主管审批
+    print('\n\n', ":::::::::::node_first start::::::::::::", '\n\n')
-    while approve_result.get('flag'):
+    print(supervisor_user)
-      # 审批通过
+    print('\n\n', ":::::::::::node_first end::::::::::::", '\n\n')
-      # 继续触发下一个上级主管审批
-      has_supervisor_approve = await task_supervisor_approve(
-        user_id=approve_result.get('user_id'),
-        approve_id=approve_id
-      )
-      if has_supervisor_approve:
-        # 触发中断，等待上级审批恢复中断，拿到审批结果
-        approve_result = interrupt({})
-        # 中断恢复之后进入下一个循环
-      else:
-        # 已经没有上级主管了，退出循环
-        break
-
-    # 退出循环之后，如果审批结果是驳回，则结束审批流程
-    if not approve_result.get('flag'):
-      # 审批驳回
-      await update_approve(
-        approve_id=approve_id,
-        session=session,
-        log_content=approve_result.get('reason'),
-        # 不再需要审批人
-        # 审批状态为驳回
-        approve_dict={"user_id": "", "status": "rejected"},
-      )
-      return False
-    else:
-      # 审批通过
-      await update_approve(
-        approve_id=approve_id,
-        session=session,
-        log_content=approve_result.get('reason'),
-        # 不再需要审批人
-        # 审批状态为通过
-        approve_dict={"user_id": "", "status": "approved"},
-      )
-      return True
-
-  # 执行上级主管审批，返回结果为布尔值，意思是是否触发了主管审批
-  @task
-  async def task_supervisor_approve(user_id: str, approve_id: str) -> bool:
-    # 直接找审批人的上级来审批
-    supervisor_user = await aget_supervisor_user(user_id=user_id, session=session)
-
-    print('\n\n', ":::::::::::task_supervisor_approve start::::::::::::", '\n\n')
-    print(supervisor_user.full_name, user_id, approve_id)
-    print('\n\n', ":::::::::::task_supervisor_approve end::::::::::::", '\n\n')
 
     if not supervisor_user:
-      return False
+      # 审批通过
+      return Command(goto="node_accept", update={})
+    else:
+      # 更新update_approve，指派申请人的上级主管审批
+      return Command(
+        goto="node_update_approve",
+        update={
+          "update_approve": {
+            "log_content": f"「{supervisor_user.full_name}」处理审批",
+            "approve_dict": {
+              "user_id": supervisor_user.id,  # 审批人为上级主管
+              "status": "approving",  # 审批状态为处理中
+            },
+          }
+        }
+      )
 
-    await update_approve(
+  # 节点用来更新审批单信息，实际上就是指派给哪个用户审批
-      approve_id=approve_id,
+  async def node_update_approve(state: ApproveGraphSchema):
+
+    update_approve = state.get('update_approve')
+
+    print('\n\n', ":::::::::::node_update_approve start::::::::::::", '\n\n')
+    print(update_approve)
+    print('\n\n', ":::::::::::node_update_approve end::::::::::::", '\n\n')
+
+    await update_approve_cls(
+      approve_id=state.get('input_approve_id'),
       session=session,
-      log_content=f"「{supervisor_user.full_name}」处理审批",
+      log_content=update_approve.get('log_content'),
-      approve_dict={
+      approve_dict=update_approve.get('approve_dict'),
-        "user_id": supervisor_user.id,  # 审批人为上级主管
-        "status": "approving",  # 审批状态为处理中
-      },
     )
+    return Command(goto="node_supervisor_approve", update={})
 
-    return True
+  async def node_supervisor_approve(state: ApproveGraphSchema):
 
-  return approve_workflow
+    # 上一个节点已经指派给用户审批了，这里直接触发中断，等待审批回执
+    approve_result = interrupt({})
+
+    if approve_result.get('flag'):
+      # 审批回执标识为通过
+
+      user_id = approve_result.get('user_id')
+      approve_id = state.get('input_approve_id')
+
+      # 下一个准备要审批的上级主管用户
+      supervisor_user = await aget_supervisor_user(user_id=user_id, session=session)
+
+      print('\n\n', ":::::::::::node_supervisor_approve start::::::::::::", '\n\n')
+      print(supervisor_user.full_name if supervisor_user else '无上级主管', user_id, approve_id)
+      print('\n\n', ":::::::::::node_supervisor_approve end::::::::::::", '\n\n')
+
+      # 已经没有上级主管了，结束审批流程
+      if not supervisor_user:
+        # 审批通过
+        return Command(goto="node_accept", update={})
+      else:
+        # 还有上级主管，继续触发下一个上级主管审批
+        return Command(
+          update={"update_approve": {
+            "log_content": f"「{supervisor_user.full_name}」处理审批",
+            "approve_dict": {
+              "user_id": supervisor_user.id,  # 审批人为上级主管
+              "status": "approving",  # 审批状态为处理中
+            },
+          }},
+          goto="node_update_approve",
+        )
+    else:
+      return Command(goto="node_reject", update={"reject_reason": approve_result.get("reason")})
+
+  async def node_accept(state: ApproveGraphSchema):
+    await update_approve_cls(
+      approve_id=state.get('input_approve_id'),
+      session=session,
+      log_content="审批已经通过",
+      # 不再需要审批人
+      # 审批状态为通过
+      approve_dict={"user_id": "", "status": "approved"},
+    )
+    return {"approve_flag": True}
+
+  async def node_reject(state: ApproveGraphSchema):
+    # 审批驳回
+    await update_approve_cls(
+      approve_id=state.get('input_approve_id'),
+      session=session,
+      log_content=state.get('reject_reason'),
+      # 不再需要审批人
+      # 审批状态为驳回
+      approve_dict={"user_id": "", "status": "rejected"},
+    )
+    return {"approve_flag": False}
+
+  builder.add_node(node_first)
+  builder.add_node(node_update_approve)
+  builder.add_node(node_supervisor_approve)
+  builder.add_node(node_accept)
+  builder.add_node(node_reject)
+
+  builder.add_edge(START, 'node_first')
+
+  graph = builder.compile(checkpointer=checkpointer)
+
+  return graph
 
 
-async def update_approve(
+async def update_approve_cls(
   approve_id: str,
   session: AsyncSessionDep,
   log_content: str,
@@ -127,12 +167,6 @@ async def aget_supervisor_user(user_id: str, session: AsyncSessionDep) -> Union[
   return await UserService.query_item(session=session, row_dict={"pos_code": parent_code})
 
 
-# approve workflow审批流程的输入参数类型
-class ApproveWorkflowInputs(TypedDict):
-  reimburse_user_id: str
-  approve_id: str
-
-
 # 审批回执数据类型
 class ApproveResult(TypedDict):
   # 审批标识，是审批通过还是审批驳回
@@ -147,3 +181,21 @@ class ApproveResult(TypedDict):
 class ApproveLog(TypedDict):
   content: str
   datetime: str
+
+
+class NodeUpdateApproveSchema(TypedDict):
+  log_content: str
+  approve_dict: dict
+
+
+# 审批流程的输入参数类型
+class ApproveGraphSchema(TypedDict):
+  input_user_id: str
+  input_approve_id: str
+
+  # 更新审批信息数据，用来执行节点：node_update_approve（入参），实际就是指派哪个用户来审批
+  update_approve: NodeUpdateApproveSchema
+  # 最后的审批结果标识，审批通过还是驳回
+  approve_flag: bool
+  # 驳回原因
+  reject_reason: str