为灵活函数调用定义节点与边

使用 LangChain 设计 Agentic 系统

Dilini K. Sumanapala, PhD

Founder & AI Engineer, Genverv, Ltd.

构建多工具工作流

   

  • 多个可用工具

    • 回文
    • 历史事件
    • 维基百科

完整的聊天机器人工作流

使用 LangChain 设计 Agentic 系统

定义工作流函数

   

  • 创建停止函数

聊天机器人工作流的聊天机器人与工具节点。

使用 LangChain 设计 Agentic 系统

定义工作流函数

   

  • 创建停止函数

    • 检查是否有工具调用

高亮的停止函数规则。

使用 LangChain 设计 Agentic 系统

定义工作流函数

   

  • 创建停止函数

    • 检查是否有工具调用
    • 无则结束对话

工作流中高亮的 END 规则

使用 LangChain 设计 Agentic 系统

定义工作流函数

   

  • 创建停止函数

    • 检查是否有工具调用
    • 无则结束对话    

  • 创建动态工具调用器

    • 如有工具调用则返回工具响应

高亮的 Call_model 规则。

使用 LangChain 设计 Agentic 系统

定义工作流函数

   

  • 创建停止函数

    • 检查是否有工具调用
    • 无则结束对话    

  • 创建动态工具调用器

    • 如有工具调用则返回工具响应
    • 无工具调用时，仅用聊天机器人节点调用 LLM

高亮的 Call_model 规则。

使用 LangChain 设计 Agentic 系统

定义工作流函数

   

  • 创建停止函数

    • 检查是否有工具调用
    • 无则结束对话    

  • 创建动态工具调用器

    • 如有工具调用则返回工具响应
    • 无工具调用时，仅用聊天机器人节点调用 LLM

  • 编译完整图

高亮的 Call_model 规则。

使用 LangChain 设计 Agentic 系统

创建停止条件函数

from langgraph.graph import MessagesState, START, END


# 使用 MessagesState 定义停止函数的状态
def should_continue(state: MessagesState):


    # 从状态中获取最后一条消息
    last_message = state["messages"][-1]


    # 检查最后一条消息是否包含工具调用
    if last_message.tool_calls:
        return "tools"


    # 若无工具调用则结束对话
    return END
使用 LangChain 设计 Agentic 系统

创建动态工具调用器

# 从历史中取出最后一条消息
def call_model(state: MessagesState):

    last_message = state["messages"][-1]


    # 若最后一条消息含工具调用，则返回工具的响应
    if isinstance(last_message, AIMessage) and last_message.tool_calls:


        # 返回工具调用产生的消息
        return {"messages": [AIMessage(content=last_message.tool_calls[0]["response"])]}


    # 否则，走常规 LLM 响应
    return {"messages": [model_with_tools.invoke(state["messages"])]}
使用 LangChain 设计 Agentic 系统

创建图

workflow = StateGraph(MessagesState)
使用 LangChain 设计 Agentic 系统

创建图

workflow = StateGraph(MessagesState)


# 为聊天机器人和工具添加节点
workflow.add_node("chatbot", call_model)
workflow.add_node("tools", tool_node)

聊天机器人和工具节点。

使用 LangChain 设计 Agentic 系统

创建图

workflow = StateGraph(MessagesState)


# 为聊天机器人和工具添加节点
workflow.add_node("chatbot", call_model)
workflow.add_node("tools", tool_node)


# 将 START 节点连接到聊天机器人
workflow.add_edge(START, "chatbot")

将 START 节点连接到聊天机器人。

使用 LangChain 设计 Agentic 系统

创建图

workflow = StateGraph(MessagesState)


# 为聊天机器人和工具添加节点
workflow.add_node("chatbot", call_model)
workflow.add_node("tools", tool_node)


# 将 START 节点连接到聊天机器人
workflow.add_edge(START, "chatbot")


# 定义条件，然后回到聊天机器人
workflow.add_conditional_edges("chatbot", 
         should_continue, ["tools", END])

已添加条件的图。

使用 LangChain 设计 Agentic 系统

创建图

workflow = StateGraph(MessagesState)


# 为聊天机器人和工具添加节点
workflow.add_node("chatbot", call_model)
workflow.add_node("tools", tool_node)


# 将 START 节点连接到聊天机器人
workflow.add_edge(START, "chatbot")


# 定义条件，然后回到聊天机器人
workflow.add_conditional_edges("chatbot", 
         should_continue, ["tools", END])

workflow.add_edge("tools", "chatbot")

完整图工作流。

使用 LangChain 设计 Agentic 系统

添加内存

# 设置内存并编译工作流
memory = MemorySaver()

app = workflow.compile(
               checkpointer=memory)


display(Image(app.get_graph()
        .draw_mermaid_png()))

完整图工作流。

使用 LangChain 设计 Agentic 系统

Passons à la pratique !

使用 LangChain 设计 Agentic 系统

Preparing Video For Download...