feat: 恢复中断的时候,要根据不同的审批来源执行不同的图

This commit is contained in:
martsforever
2025-08-24 23:15:42 +08:00
parent aa303a6799
commit 79b60e7075
+12 -83
View File
@@ -1,69 +1,16 @@
from fastapi import FastAPI from fastapi import FastAPI
from langchain_core.runnables import RunnableConfig
from langgraph.constants import START
from langgraph.graph import StateGraph
from langgraph.types import Command from langgraph.types import Command
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from app.controller.add_hotel_route import create_book_hotel_graph
from app.controller.add_reimburse_route import create_reimburse_graph
from app.model.ApproveModel import ApproveService, ApproveModel from app.model.ApproveModel import ApproveService, ApproveModel
from app.model.ReimburseModel import ReimburseService, ReimburseModel
from app.utils.db_utils import AsyncSessionDep from app.utils.db_utils import AsyncSessionDep
from app.utils.postgres_checkpointer import AsyncPostgresSaverDep from app.utils.postgres_checkpointer import AsyncPostgresSaverDep
from app.workflow.approve import ApproveResult, create_approve_graph, ApproveGraphSchema from app.workflow.approve import ApproveResult
def add_approve_route(app: FastAPI): def add_approve_route(app: FastAPI):
# 提交报销单接口
@app.post('/submit_reimburse')
async def submit_reimburse(
reimburse: dict,
session: AsyncSessionDep,
checkpointer: AsyncPostgresSaverDep,
):
reimburse_id = reimburse.get('id')
# 先查一遍报销单信息
reimburse_cls: ReimburseModel = await ReimburseService.query_item(session=session, row_dict={"id": reimburse_id})
# 1. 没有审批单——
# 2. 审批单为审批中状态——不处理
# 3. 审批单为已通过状态——不处理
# 4. 审批单为已驳回状态——
# 5. 审批单为已撤回状态——
# 2,3 的情况下不再提交报销单
if reimburse_cls.approve:
if reimburse_cls.approve.status == 'approving':
return {"message": "报销单已提交审批,请勿重复提交!"}
if reimburse_cls.approve.status == 'approved':
return {"message": "报销单已通过审批,请勿重复提交!"}
# 1,4,5 的情况下新建审批单
# 创建审批单,使用新的审批单走审批流程
new_approve_dict = {
"title": f"{reimburse_cls.user.full_name} - 报销单审批",
"description": reimburse_cls.title + (f" / {reimburse_cls.remarks}" if reimburse_cls.remarks else ""),
"status": "approving",
"amount": reimburse_cls.amount,
"logs": "[]",
"user_id": reimburse_cls.user_id, # 先把审批人的id设置为申请人
"proj_id": reimburse_cls.proj_id,
}
insert_approve_cls: ApproveModel = await ApproveService.item_insert(session=session, row_dict=new_approve_dict)
# 将审批单与报销单管理,设置报销单的approve_id为审批单的id
await ReimburseService.item_update(session=session, row_dict={"id": reimburse_id, "approve_id": insert_approve_cls.id}, )
reimburse_graph = create_reimburse_graph(session=session, checkpointer=checkpointer)
graph_state = await reimburse_graph.ainvoke(
{"input_user_id": reimburse_cls.user.id, "input_approve_id": insert_approve_cls.id, "input_amount": str(insert_approve_cls.amount)},
config={"configurable": {"thread_id": insert_approve_cls.id}}
)
return {"message": "报销单提交成功!", "graph_state": graph_state}
class ProcessApproveClass(BaseModel): class ProcessApproveClass(BaseModel):
flag: bool = Field(..., description="审批标识,是审批通过还是审批驳回") flag: bool = Field(..., description="审批标识,是审批通过还是审批驳回")
reason: str = Field(default=None, description="审批驳回原因") reason: str = Field(default=None, description="审批驳回原因")
@@ -77,7 +24,14 @@ def add_approve_route(app: FastAPI):
session: AsyncSessionDep, session: AsyncSessionDep,
checkpointer: AsyncPostgresSaverDep, checkpointer: AsyncPostgresSaverDep,
): ):
reimburse_graph = create_reimburse_graph(session=session, checkpointer=checkpointer) approve_cls: ApproveModel = await ApproveService.query_item(session, row_dict={"id": body.approve_id})
if approve_cls.approve_from == 'reimburse':
graph = create_reimburse_graph(session=session, checkpointer=checkpointer)
elif approve_cls.approve_from == 'book_hotel':
graph = create_book_hotel_graph(session=session, checkpointer=checkpointer)
else:
raise Exception("未知的审批来源:" + approve_cls.approve_from)
approve_result: ApproveResult = { approve_result: ApproveResult = {
"flag": body.flag, "flag": body.flag,
"reason": body.reason, "reason": body.reason,
@@ -88,29 +42,4 @@ def add_approve_route(app: FastAPI):
print(approve_result) print(approve_result)
print('\n\n', ":::::::::::process_approve end::::::::::::", '\n\n') print('\n\n', ":::::::::::process_approve end::::::::::::", '\n\n')
return await reimburse_graph.ainvoke(Command(resume=approve_result), config={"configurable": {"thread_id": body.approve_id}}) return await graph.ainvoke(Command(resume=approve_result), config={"configurable": {"thread_id": body.approve_id}})
def create_reimburse_graph(
checkpointer: AsyncPostgresSaverDep,
session: AsyncSessionDep,
):
approve_graph = create_approve_graph(checkpointer, session)
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 {}
builder.add_node("node", node)
builder.add_edge(START, "node")
return builder.compile(checkpointer=checkpointer)