langgraph之智能体工作流

目录

前言

一、langgraph是什么？

1.相关概念

二、使用步骤

1.引入库

2.假设需求

3.定义AgentState

4.定义LLM

5.定义节点方法

6.节点、边添加

7.编译执行

8.优化

9.条件边

总结

---

前言

随着大模型能力的不断提升，大模型应用已经逐步从提示词到RAG再到智能体Agent过渡。Agent属于大模型高阶应用，但是受限于模型能力（思考规划等），智能体的稳定性存在很大的考验。为了解决这个问题，一个方向是提高大模型能力，再一个方向是预先将智能体执行过程进行编排，即工作流。

本文就是要介绍智能体工作流框架langgraph，如何通过langgraph实现一个不错的智能体应用。

---

一、langgraph是什么？

langgraph是langchain官方推出的智能体工作流框架，该框架基于langchain、LECE，这样即可利用langchain的基础功能，从而可以更加高效的搭建出智能体工作流

1.相关概念

节点：可以理解为是一个独立的操作，每个节点可以和一个方法关联起来

边：节点的连线，通过边将节点连接起来，以便将多个操作串联起来

条件：有些节点到节点之间是有条件的

状态信息：在多个节点之间执行操作，需要保持状态，就用到了状态信息

编译：对工作流进行编译

执行：传参、执行工作流，触发工作流按预定义顺序执行相关操作

二、使用步骤

1.引入库

from typing import TypedDict, List
import colorama
import os
 
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
from langchain_core.messages import HumanMessage
from langchain_core.runnables import RunnableConfig
 
from langgraph.graph import StateGraph, END
from langgraph.pregel import GraphRecursionError

2.假设需求

给我写一封关于AI应用的邮件，然后发给老李

需求分析：

1、写邮件，参数：主题（AI应用）

2、找邮箱，参数：姓名

3、发送邮件，参数：主题、收件人邮箱、正文

3.定义AgentState

代码如下（示例）：

class AgentState(TypedDict):  # 定义AgentState类型，用于存储代理的状态
    receiver: str
    title: str
    content: str
    sendState:str
    theme: str
    user: str

这里AgentState类会在流程图每个节点共享数据，这里定义的字段主要用途：

1）传参，以便第一个节点可以获取到需要的信息

2）过程数据传递，以便后续步骤可以使用上一步骤的信息

4.定义LLM

代码如下（示例）：

from __init__ import load_env
 
load_env()
 
base_url = os.environ.get("Yi_AI_BASE_URL")
api_key = os.environ.get("Yi_AI_API_KEY")
 
 
llm = ChatOpenAI(base_url=base_url, api_key=api_key, model="yi-large")

这里定义了LLM，后续节点方法中即可调用大模型

5.定义节点方法

代码如下（示例）：

def writeEmail(state: AgentState):
    user=state["user"]
    theme=state["theme"]
 
    prompt = """
    ##角色
    你是一个邮件编写专家，可以根据用户需求编写高质量的邮件，要求字数控制在500字以内
    
    ##用户需求
    接收人：{user}
    主题：{theme}
    """
    prompt = PromptTemplate.from_template(prompt)
    prompt = prompt.format(user=user, theme=theme)
    
    content = llm.invoke(prompt).content
    
    state["content"]=content
    state["title"]="测试主题"
 
    return state
 
 
def getEmail(state: AgentState):
    state["receiver"]= "laoli@xx.com"
    return state
 
 
def sendEmail(state: AgentState):
    print(f"已发送：{state['receiver']}-{state['title']}-{state['content']}")
 
    state["sendState"]="ok"

这里分别实现了写邮件、获取邮件地址、发送邮件功能

注意：相关代码主要为了举例，不作为实际参考代码

6.节点、边添加

代码如下（示例）：

# 创建StateGraph
workflow = StateGraph(AgentState)
 
 
# 将节点添加到工作流中
# 节点名称  # 节点对应的函数
workflow.add_node("writeEmail", writeEmail)  
 
# 将获取邮箱节点添加到工作流中
workflow.add_node("getEmail", getEmail)
 
# 将发送邮件节点添加到工作流中
workflow.add_node("sendEmail", sendEmail)
 
# 定义入口点，这是流程开始的地方
workflow.set_entry_point("writeEmail")
 
# 总是从writeEmail跳转到getEmail
workflow.add_edge("writeEmail", "getEmail")
 
# 总是从getEmail跳转到sendEmail
workflow.add_edge("getEmail", "sendEmail")
 
# 总是从sendEmail跳转到END
workflow.add_edge("sendEmail", END)

先添加节点，然后安装from-to格式添加边

7.编译执行

代码如下（示例）：

app = workflow.compile()  # 编译工作流
 
# app.get_graph().print_ascii()
 
inputs = {"user": "老李","theme":"AI应用"}  # 输入参数
config = RunnableConfig(recursion_limit=10)  # 设置递归限制
try:
    result = app.invoke(inputs, config)  # 运行应用
    # print(result)  # 打印结果
    print("finish")
except GraphRecursionError:  # 如果达到递归限制
    print("达到图递归限制。")  # 打印错误信息

注意：inputs即初始参数，这里键值对要和AgentState字段对应

8.优化

如上传参，其实可以优化

通过输入“给老李发一封关于AI应用的邮件”，然后先调用大模型识别出参数user（老李）、theme（AI应用），这样使用更加自然一些。

9.条件边

# 判断是否完成
def should_continue(state: AgentState):
    if len(state["class_methods"]) == 0:  # 如果没有更多的方法要测试
        return "end"  # 结束流程
    else:
        return "continue"  # 继续流程
 
 
# 添加条件边
workflow.add_conditional_edges(
    "write_tests",  # 条件边的起始节点
    should_continue,  # 条件函数
    {
        "continue": "write_tests",  # 如果应该继续，则再次执行write_tests节点
        "end": "write_file",  # 如果结束，则跳转到write_file节点
    },
)

这里的条件判断，也是基于AgentState数据

如上代码的思路可以说下，这里class_methods是一个list对象，在前面节点方法中每次处理掉一条数据则去除一条数据，当 class_methods的len==0说明数据处理完了，即可到下一步骤，否则需要回到前面节点重复执行。

---

总结

如上将langgraph相关概念，基本使用做了介绍，希望可以帮助到有需要的小伙伴。

欢迎搭建多多点赞、收藏、评论，一起学习进步。