diff --git a/app/controller/add_approve_route.py b/app/controller/add_approve_route.py index 1cc76c7..0c902cc 100644 --- a/app/controller/add_approve_route.py +++ b/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( - {"reimburse_user_id": reimburse_cls.user.id, "approve_id": insert_approve_cls.id}, + graph_state = await reimburse_graph.ainvoke( + {"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) - async def reimburse_workflow(input_dict: ApproveWorkflowInputs, config: RunnableConfig): - approve_flag = await approve_workflow.ainvoke(input_dict, config=config) + builder = StateGraph(ApproveGraphSchema) + + 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) diff --git a/app/workflow/approve.py b/app/workflow/approve.py index 688fabf..ecd5d65 100644 --- a/app/workflow/approve.py +++ b/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.types import interrupt +from langgraph.constants import START +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) - 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') + builder = StateGraph(ApproveGraphSchema) - # 立即拟造一条审批结果,审批人为申请人自己,审批结果为通过 - approve_result: ApproveResult = {"flag": True, "user_id": reimburse_user_id, } + async def node_first(state: ApproveGraphSchema): + supervisor_user = await aget_supervisor_user(user_id=state.get('input_user_id'), session=session) - # 循环判断审批结果,如果为通过,则继续触发下一个上级主管审批 - while approve_result.get('flag'): - # 审批通过 - # 继续触发下一个上级主管审批 - 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') + print('\n\n', ":::::::::::node_first start::::::::::::", '\n\n') + print(supervisor_user) + print('\n\n', ":::::::::::node_first 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}」处理审批", - approve_dict={ - "user_id": supervisor_user.id, # 审批人为上级主管 - "status": "approving", # 审批状态为处理中 - }, + log_content=update_approve.get('log_content'), + approve_dict=update_approve.get('approve_dict'), ) + 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