前面创建的langgraph服务都是agent服务，本文创建一个基于工作流的langgraph服务。该工作流用于电子印章系统中，支持用户通过AI实现电子印章的申请、管理，也可以直接使用聊天工程。

     1.创建工作流服务

        仍基于系统提供的模板创建，具体步骤如下：

      2.工作流说明

        工作流图如下图所示：

        用户发起会话后，大模型对用户的意图进行识别。判断用户是想制作新的印章，还是对已有印章进行管理，还是只是想跟大模型闲聊。如果用户想制作印章，则工作流进入premake_node节点，如果用户想管理印章，则工作流进入premanage_ndoe节点，否则直接进入chatbot节点。

        1）制作印章

        premake_node检查用户的提供的信息中是否包含了制作印章必须的信息，如果没有，则提示用户补充完善，并进入__end__节点；如果信息齐备，则进入make_seal_bot节点。

        make_seal_bot节点中使用绑定了制作印章工具的大模型，使用大模型的function calling获取调用的工具和参数，如果成功，则进入make_seal_tools，在make_seal_tools中调用外部接口制作印章，然后再从make_seal_tools节点回到make_seal_bot节点，在make_seal_bot节点中由大模型根据接口调用结果给用户返回应答。

        2）管理印章

        premanage_node检查用户的提供的信息中是否包含了管理印章必须的信息，如果没有，则提示用户补充完善，并进入__end__节点；如果信息齐备，则进入manage_seal_bot节点。

        manage_seal_bot节点中使用绑定了管理印章工具的大模型，使用大模型的function calling获取调用的工具和参数，如果成功，则进入make_seal_tools，在make_seal_tools中调用外部接口管理印章，然后再从manage_seal_tools节点回到manage_seal_bot节点，在manage_seal_bot节点中由大模型根据接口调用结果给用户返回应答。

     3）在以上两个流程中，在调用外部工具前，均通过用户回环提示用户确认调用工具的参数。

     4）如果用户的意图并非制作印章和管理印章，则进入chatbot节点，chatbot节点中使用一个绑定了搜索引擎工具的大模型与用户交互。

     3.工作流实现

    3.1全局变量

os.environ["TAVILY_API_KEY"] = "tvly-……"
search_tool = TavilySearch(max_results=2)

llm = ChatOpenAI(
    model = 'qwen-plus',
    api_key = "sk-……",
    base_url = "https://dashscope.aliyuncs.com/compatible-mode/v1")
system_prompt_1: str = """#制作印章时，请求参数检查提示词
                    请根据用户最近的输入信息，检查是否包含了\
                    企业名称信息和印章类型信息,\
                    其中印章类型信息包括：\
                    法人名章\
                    发票章\
                    财务章\
                    合同章\
                    公章\
                    如果用户输入中包含了企业名称和印章类型信息,\
                    则答复:我将马上帮您制作印章\
                    否则提醒用户提供对应的缺失信息
                    其中印章类型必须是以上所列5种,\
                    不需要参考电子印章相关法律法规和材料\
                    不需要额外扩展，也不需要follow up
                    """
system_prompt_2: str = """管理印章时，请求参数检查提示词
                    请对用户用户最近的输入信息，检查其中是否包含了
                    企业名称信息,印章类型信息和管理类型信息\
                    其中印章类型信息包括：\
                    法人名章,发票章,财务章,合同章和公章.\
                    管理类型信息包括:\
                    停用印章,启用印章,注销印章和对印章续期\
                    如果用户输入中包含了企业信息,印章类型信息和\
                    管理类型信息,\
                    则答复:我将马上帮您对您的印章进行管理\
                    否则提示用户请提供对应的缺失信息,\
                    其中印章类型必须是以上所列5种,\
                    管理类型比如是以上列四种\
                    不需要参考电子印章相关法律法规和材料\
                    不需要额外扩展，也不需要follow up
                    """

@dataclass
class State:
    messages: Annotated[list, add_messages]
 

       3.2工具包装器

        调用工具前人工介入相关代码：

#所有需要对参数经用户确认的工具，均需要用本方法来包装。

def add_human_in_the_loop(
    tool: Callable | BaseTool,
    *,
    interrupt_config: HumanInterruptConfig = None,
) -> BaseTool:
    """Wrap a tool to support human-in-the-loop review."""
    if not isinstance(tool, BaseTool):
        tool = create_tool(tool)

    @create_tool(
        tool.name,
        description=tool.description,
        args_schema=tool.args_schema
    )
    def call_tool_with_interrupt(config: RunnableConfig, **tool_input):
        #HumanInterrupt中的action_request和config不能用官网中的实例代码，
        #在界面中不会出现相关的界面元素。
        request = HumanInterrupt(
            action_request=ActionRequest(
                action=tool.name, #工具名
                args=tool_input    #工具参数
            ),

            config=HumanInterruptConfig(#控制界面元素
                allow_ignore=False,    # 是否允许忽略
                allow_respond=False,   # 是否允许直接文本恢复
                allow_edit=True,     #是否允许编辑
                allow_accept=True     # 是否允许直接接收
            ),
            description="请对您的数据进行确认"
        )
        response = interrupt([request])[0]
        # 用户接受时，直接用原有参数调用工具
        if response["type"] == "accept":
            tool_response = tool.invoke(tool_input, config)
        # 用修改后的参数数据调用工具
        elif response["type"] == "edit":
            tool_input = response["args"]["args"]
            tool_response = tool.invoke(tool_input, config)
        else:
            raise ValueError(f"Unsupported interrupt response type: {response['type']}")
        return tool_response
    return call_tool_with_interrupt

      3.3意图识别

        意图识别代码entry_router，用大模型判断用户意图，该函数作为增加条件边的函数，具体代码如下：

def entry_router(
    state: State,
):
    prompt = f"""
    根据用户输入分析用户的意图类型，返回以下之一：
    - make: 用户想制作电子印章
    - manage: 用户想对印章进行管理，比如对印章冻结、解冻、注销、续期和变更
    - grant: 用户想把印章授权给其他用户管理管理或使用
    — consult: 用户想咨询印章相关的业务知识或技术知识
    — chat: 其他
    用户输入{state.messages}
    """
    response = llm.predict(prompt)
    match response:
        case "make":
            return "premake_node"
        case "manage":
            return "premanage_node"
        case _:
            return "chatbot"

        3.4制作印章

        制作印章相关的节点包括premake_node节点、make_seal_bot节点和make_seal_tools节点。

        premake_node节点代码如下：

def premake_node(state: State, runtime: Runtime[Context]) -> Dict[str, Any]:

     #在用户输入的数据前插入检查参数提示词
     state.messages.insert(-1, {"role":"system", "content":system_prompt_1})
     response = llm.invoke(state.messages)
     del state.messages[-2] #把插入的提示词从消息中清除，减少干扰
     return {"messages": response}

       从premake_node节点是进入make_seal_bot还是END，由如下方法判断实现：

def make_seal_router(
    state: State,
):
    message = state.messages[-1].content
    prompt = f"""
    根据输入信息判断是否表达了将马上为用户制作印章的意图，返回结果如下：
    - "yes": 如果表达了将马上为用户制作印章的意图
    - "no": 如果未表达将马上为用户制作印章的意图

    输入信息：{message}
    """
    response = llm.predict(prompt)
    if  response == "yes":
        return "make_seal_bot" #如果大模型能够制作印章，则进入make_seal_bot节点
    return END
 

        make_seal_tools节点代码如下：

make_seal_tools = [
    add_human_in_the_loop(make_seal_tool),
]#工具封装
make_seal_tool_node = ToolNode(tools=make_seal_tools)#生成节点

        make_seal_bot节点代码如下：

make_seal_llm = llm.bind_tools(make_seal_tools)#只绑定制作印章工具的大模型
def make_seal_bot(state: State, runtime: Runtime[Context]) -> Dict[str, Any]:
    response = make_seal_llm.invoke(state.messages)
    return {"messages": [response]}

        从make_seal_bot节点出发，是流转至make_seal_tools节点调用工具，还是结束，由如下方法判断实现：

def route_make_seal_tools(
    state: State,
):
    """
    Use in the conditional_edge to route to the ToolNode if the last message
    has tool calls. Otherwise, route to the end.
    """
    ai_message = state.messages[-1]
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "make_seal_tools"
    return END
 

        3.5管理印章

        制作印章相关的节点包括premanage_node节点、manage_seal_bot节点和manage_seal_tools节点。

        premange_node节点代码如下：

def premanage_node(state: State, runtime: Runtime[Context]) -> Dict[str, Any]:

     #在用户输入前插入检查印章管理信息的提示词
     state.messages.insert(-1, {"role":"system", "content":system_prompt_2})
     response = llm.invoke(state.messages)
     del state.messages[-2]#把前面增加的提示词删除，减少干扰
     return {"messages": response}

         manage_seal_tools节点代码如下：

manage_seal_tools = [
    add_human_in_the_loop(manage_seal_tool),
]#工具封装
manage_seal_tool_node = ToolNode(tools=manage_seal_tools)#生成节点

      从premanage_node节点出发，是流转至manage_seal_bot节点还是结束，由如下方法判断实现：

def manage_seal_router(
    state: State,
):
    message = state.messages[-1].content
    prompt = f"""
    根据输入信息判断是否表达了将马上为用户管理印章的意图，返回结果如下：
    - "yes": 如果表达了将马上为用户管理印章的意图
    - "no": 如果未表达将马上为用户管理印章的意图

    输入信息：{message}
    """
    response = llm.predict(prompt)
    if  response == "yes":
        return "manage_seal_bot"
    return END
 

        manage_seal_bot节点代码如下：

manage_seal_llm = llm.bind_tools(manage_seal_tools)#只绑定管理印章工具的大模型

def manage_seal_bot(state: State, runtime: Runtime[Context]) -> Dict[str, Any]:
    response = manage_seal_llm.invoke(state.messages)
    return {"messages": [response]}

        从manage_seal_tools节点出发，是流转至manage_seal_tools节点调用工具还是结束，由如下代码实现：

def route_manage_seal_tools(
    state: State,
):
    """
    Use in the conditional_edge to route to the ToolNode if the last message
    has tool calls. Otherwise, route to the end.
    """
    ai_message = state.messages[-1]
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "manage_seal_tools"
    return END
 

      3.5通用聊天

        聊天工具节点chat_tool_node代码如下：

chat_tools = [
    search_tool,
]
chat_tool_node = ToolNode(tools=chat_tools)
 

      chatbot节点代码如下：

chat_llm = llm.bind_tools(chat_tools) #仅绑定搜索引擎工具的大模型

def chatbot(state: State):
    return {"messages": [chat_llm.invoke(state.messages)]}

        路由条件如下：

def route_chat_tools(
    state: State,
):
    """
    如果大模型返回的消息中含有tool_calls信息，则流转到chat_tools节点，否则结束
    """
    ai_message = state.messages[-1]
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "chat_tools"
    return END
 

       3.6工具代码

  工具代码如下：

#制作印章工具，实际需要调用电子印章系统的接口，这里仅模拟各种成功失败的输出结果

def make_seal_tool(name: str, type: str):
    """to make seal"""
    rnd = random.randint(1, 4)
    match rnd:
        case 1|2:
            return f"{name}的{type} 制作成功, 印章图片的base64编码为"
        case 3:
            return f"{name}的{type} 制作失败,请求第三方数字证书系统失败."
        case 4:
            return f"{name}的{type} 制作失败, 系统内部错误"
        case _:
            return f"{name}的{type} 制作成功, 印章图片的base64编码为"

#仅返回管理成功，可根据具体的业务需求增加业务逻辑

def manage_seal_tool(name: str, seal_type: str, manage_type: str):
    """to manage seal"""
    return f"{name}的{seal_type} 成功 {manage_ty

        3.7 生成工作流图

graph_builder = StateGraph(State)

#以下代码是增加前面定义的所有节点

#聊天相关节点
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("chat_tools", chat_tool_node)

#制作印章相关节点
graph_builder.add_node("premake_node", premake_node)
graph_builder.add_node("make_seal_bot", make_seal_bot)
graph_builder.add_node("make_seal_tools", make_seal_tool_node)

#管理印章相关节点
graph_builder.add_node("premanage_node", premanage_node)
graph_builder.add_node("manage_seal_bot", manage_seal_bot)
graph_builder.add_node("manage_seal_tools", manage_seal_tool_node)
#以下代码是增加边

#从START出发的三个边，分别指向premake_node,premanage_node和chatbot

graph_builder.add_conditional_edges(
    START,
    entry_router,
    {"premake_node": "premake_node", "premanage_node": "premanage_node","chatbot": "chatbot"}
)

#从premake_node出发的两条边，分别指向make_seal_bot和END
graph_builder.add_conditional_edges(
    "premake_node",
    make_seal_router,
    {"make_seal_bot": "make_seal_bot", END: END}
)

#增加从make_seal_tools—>make_seal_bot的边
graph_builder.add_edge("make_seal_tools", "make_seal_bot")

 

#增加从make_seal_bot出发的两条边，一条指向make_seal_tools，一条指向END节点
graph_builder.add_conditional_edges(
    "make_seal_bot",
    route_make_seal_tools,
    {"make_seal_tools": "make_seal_tools", END: END},
)

#增加从premanage_node出发的两条边，一条边指向manage_seal_bot节点，一条指向END

graph_builder.add_conditional_edges(
    "premanage_node",
    manage_seal_router,
    {"manage_seal_bot": "manage_seal_bot", END: END}
)

#增加从manage_seal_tools—>manage_seal_bot的边
graph_builder.add_edge("manage_seal_tools", "manage_seal_bot")

#增加从manage_seal_bot出发的两条边，一条指向manage_seal_tools节点，一条指向END

graph_builder.add_conditional_edges(
    "manage_seal_bot",
    route_manage_seal_tools,
    {"manage_seal_tools": "manage_seal_tools", END: END},
)

#增加从chat_tools—>chatbot的边
graph_builder.add_edge("chat_tools", "chatbot")

#增加从chatbot出发的两条边，一条指向chat_tools节点，一条指向END节点
graph_builder.add_conditional_edges(
    "chatbot",
    route_chat_tools,
    {"chat_tools": "chat_tools", END: END},
)

graph = graph_builder.compile()