LangGraph 入门与实战

原文:LangGraph 入门与实战 - 知乎

大家好,我是雨飞。LangGraph 是在 LangChain 基础上的一个库,是 LangChain 的 LangChain Expression Language (LCEL)的扩展。能够利用有向无环图的方式,去协调多个LLM或者状态,使用起来比 LCEL 会复杂,但是逻辑会更清晰。

相当于一种高级的LCEL语言,值得一试。

安装也十分简单。注意,这个库需要自己去安装,默认的LangChain不会安装这个库。

pip install langgraph

由于,OpenAI访问不方便,我们统一使用智普AI的大模型进行下面的实践。

智普AI的接口和OpenAI的比较类似,因此也可以使用OpenAI的tools的接口,目前还没有发现第二家如此方便的接口。实际使用起来,还是比较丝滑的,虽然有一些小问题。

我们下面以ToolAgent的思想,利用LangGraph去实现一个可以调用工具的Agent。

定义工具以及LLM

工具的定义,可以参考这篇文章,写的比较详细了,比较方便的就是使用 tools 这个注解。

雨飞:使用智普清言的Tools功能实现ToolAgent

定义Agent的状态

LangGraph 中最基础的类型是 StatefulGraph,这种图就会在每一个Node之间传递不同的状态信息。然后每一个节点会根据自己定义的逻辑去更新这个状态信息。具体来说,可以继承 TypeDict 这个类去定义状态,下图我们就定义了有四个变量的信息。

input:这是输入字符串,代表用户的主要请求。

chat_history: 这是之前的对话信息,也作为输入信息传入.

agent_outcome: 这是来自代理的响应,可以是 AgentAction,也可以是 AgentFinish。如果是 AgentFinish,AgentExecutor 就应该结束,否则就应该调用请求的工具。

intermediate_steps: 这是代理在一段时间内采取的行动和相应观察结果的列表。每次迭代都会更新。

class AgentState(TypedDict):# The input stringinput: str# The list of previous messages in the conversationchat_history: list[BaseMessage]# The outcome of a given call to the agent# Needs `None` as a valid type, since this is what this will start asagent_outcome: Union[AgentAction, AgentFinish, None]# List of actions and corresponding observations# Here we annotate this with `operator.add` to indicate that operations to# this state should be ADDED to the existing values (not overwrite it)intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]

定义图中的节点

在LangGraph中,节点一般是一个函数或者langchain中runnable的一种类。

我们这里定义两个节点,agent和tool节点,其中agent节点就是决定执行什么样的行动,

tool节点就是当agent节点选择执行某个行动时,去调用相应的工具。

此外,还需要定义节点之间的连接,也就是边。

条件判断的边:定义图的走向,比如Agent要采取行动时,就需要接下来调用tools,如果Agent说当前的的任务已经完成了,则结束整个流程。

普通的边:调用工具后,始终需要返回到Agent,让Agent决定下一步的行动

from langchain_core.agents import AgentFinish
from langgraph.prebuilt.tool_executor import ToolExecutor# This a helper class we have that is useful for running tools
# It takes in an agent action and calls that tool and returns the result
tool_executor = ToolExecutor(tools)# Define the agent
def run_agent(data):agent_outcome = agent_runnable.invoke(data)return {"agent_outcome": agent_outcome}# Define the function to execute tools
def execute_tools(data):# Get the most recent agent_outcome - this is the key added in the `agent` aboveagent_action = data["agent_outcome"]print("agent action:{}".format(agent_action))output = tool_executor.invoke(agent_action[-1])return {"intermediate_steps": [(agent_action[-1], str(output))]}# Define logic that will be used to determine which conditional edge to go down
def should_continue(data):# If the agent outcome is an AgentFinish, then we return `exit` string# This will be used when setting up the graph to define the flowif isinstance(data["agent_outcome"], AgentFinish):return "end"# Otherwise, an AgentAction is returned# Here we return `continue` string# This will be used when setting up the graph to define the flowelse:return "continue"

定义图

然后,我们就可以定义整个图了。值得注意的是,条件判断的边和普通的边添加方式是不一样的

最后需要编译整个图,才能正常运行。

# Define a new graph
workflow = StateGraph(AgentState)# Define the two nodes we will cycle between
workflow.add_node("agent", run_agent)
workflow.add_node("action", execute_tools)# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")# We now add a conditional edge
workflow.add_conditional_edges(# First, we define the start node. We use `agent`.# This means these are the edges taken after the `agent` node is called."agent",# Next, we pass in the function that will determine which node is called next.should_continue,# Finally we pass in a mapping.# The keys are strings, and the values are other nodes.# END is a special node marking that the graph should finish.# What will happen is we will call `should_continue`, and then the output of that# will be matched against the keys in this mapping.# Based on which one it matches, that node will then be called.{# If `tools`, then we call the tool node."continue": "action",# Otherwise we finish."end": END,},
)# We now add a normal edge from `tools` to `agent`.
# This means that after `tools` is called, `agent` node is called next.
workflow.add_edge("action", "agent")# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
app = workflow.compile()

总代码

下面是所有的可执行代码,注意,需要将api_key替换为自己的api_key。

# !/usr/bin env python3
# -*- coding: utf-8 -*-
# author: yangyunlong time:2024/2/28
import datetime
import operator
from typing import TypedDict, Annotated, Union, Optional,Type,Listimport requests
from langchain import hub
from langchain.agents import create_openai_tools_agent
from langchain.pydantic_v1 import BaseModel, Field
from langchain.tools import BaseTool, tool
from langchain_core.agents import AgentAction
from langchain_core.agents import AgentFinish
from langchain_core.messages import BaseMessage
from langgraph.graph import END, StateGraph
from langgraph.prebuilt.tool_executor import ToolExecutor
from zhipu_llm import ChatZhipuAIzhipuai_api_key = ""
glm3 = "glm-3-turbo"
glm4 = "glm-4"chat_zhipu = ChatZhipuAI(temperature=0.8,api_key=zhipuai_api_key,model=glm3
)class Tagging(BaseModel):"""分析句子的情感极性,并输出句子对应的语言"""sentiment: str = Field(description="sentiment of text, should be `pos`, `neg`, or `neutral`")language: str = Field(description="language of text (should be ISO 639-1 code)")class Overview(BaseModel):"""Overview of a section of text."""summary: str = Field(description="Provide a concise summary of the content.")language: str = Field(description="Provide the language that the content is written in.")keywords: str = Field(description="Provide keywords related to the content.")@tool("tagging", args_schema=Tagging)
def tagging(s1: str, s2: str):"""分析句子的情感极性,并输出句子对应的语言"""return "The sentiment is {a}, the language is {b}".format(a=s1, b=s2)@tool("overview", args_schema=Overview)
def overview(summary: str, language: str, keywords: str):"""Overview of a section of text."""return "Summary: {a}\nLanguage: {b}\nKeywords: {c}".format(a=summary, b=language, c=keywords)@tool
def get_current_temperature(latitude: float, longitude: float):"""Fetch current temperature for given coordinates."""BASE_URL = "https://api.open-meteo.com/v1/forecast"# Parameters for the requestparams = {'latitude': latitude,'longitude': longitude,'hourly': 'temperature_2m','forecast_days': 1,}# Make the requestresponse = requests.get(BASE_URL, params=params)if response.status_code == 200:results = response.json()else:raise Exception(f"API Request failed with status code: {response.status_code}")current_utc_time = datetime.datetime.utcnow()time_list = [datetime.datetime.fromisoformat(time_str.replace('Z', '+00:00')) for time_str inresults['hourly']['time']]temperature_list = results['hourly']['temperature_2m']closest_time_index = min(range(len(time_list)), key=lambda i: abs(time_list[i] - current_utc_time))current_temperature = temperature_list[closest_time_index]return f'The current temperature is {current_temperature}°C'tools = [tagging, overview, get_current_temperature]
# Get the prompt to use - you can modify this!
prompt = hub.pull("hwchase17/openai-tools-agent")# Construct the OpenAI Functions agent
agent_runnable = create_openai_tools_agent(chat_zhipu, tools, prompt)class AgentState(TypedDict):# The input stringinput: str# The list of previous messages in the conversationchat_history: list[BaseMessage]# The outcome of a given call to the agent# Needs `None` as a valid type, since this is what this will start asagent_outcome: Union[AgentAction, AgentFinish, None]# List of actions and corresponding observations# Here we annotate this with `operator.add` to indicate that operations to# this state should be ADDED to the existing values (not overwrite it)intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]# This a helper class we have that is useful for running tools
# It takes in an agent action and calls that tool and returns the resulttool_executor = ToolExecutor(tools)# Define the agent
def run_agent(data):agent_outcome = agent_runnable.invoke(data)return {"agent_outcome": agent_outcome}# Define the function to execute tools
def execute_tools(data):# Get the most recent agent_outcome - this is the key added in the `agent` aboveagent_action = data["agent_outcome"]print("agent action:{}".format(agent_action))output = tool_executor.invoke(agent_action[-1])return {"intermediate_steps": [(agent_action[-1], str(output))]}# Define logic that will be used to determine which conditional edge to go down
def should_continue(data):# If the agent outcome is an AgentFinish, then we return `exit` string# This will be used when setting up the graph to define the flowif isinstance(data["agent_outcome"], AgentFinish):return "end"# Otherwise, an AgentAction is returned# Here we return `continue` string# This will be used when setting up the graph to define the flowelse:return "continue"# Define a new graph
workflow = StateGraph(AgentState)# Define the two nodes we will cycle between
workflow.add_node("agent", run_agent)
workflow.add_node("action", execute_tools)# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")# We now add a conditional edge
workflow.add_conditional_edges(# First, we define the start node. We use `agent`.# This means these are the edges taken after the `agent` node is called."agent",# Next, we pass in the function that will determine which node is called next.should_continue,# Finally we pass in a mapping.# The keys are strings, and the values are other nodes.# END is a special node marking that the graph should finish.# What will happen is we will call `should_continue`, and then the output of that# will be matched against the keys in this mapping.# Based on which one it matches, that node will then be called.{# If `tools`, then we call the tool node."continue": "action",# Otherwise we finish."end": END,},
)# We now add a normal edge from `tools` to `agent`.
# This means that after `tools` is called, `agent` node is called next.
workflow.add_edge("action", "agent")# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
app = workflow.compile()inputs = {"input": "what is the weather in NewYork", "chat_history": []}
result = app.invoke(inputs)
print(result["agent_outcome"].messages[0].content)

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/769351.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

【周总结】

周总结 完成项目混合版时区改造 完成相关jira问题的修改 完成老版本APP数据保存接口的兼容,手动赋值时区 2024/03/24 天气阴 一点不冷 1.Its time to go、Spring is coming! 2. Its a nice day that staying with friends in a peaceful …

CentOS DHCP服务器部署指南

title: DHCP 服务器部署以及配置 search: 2024-03-21 tags: “#DHCP 服务器部署以及配置” CentOS DHCP 服务器部署指南 背景 :因上了 Linux 的实验课程,在课程中,老师要求我们自己搭建 DHCP 服务器构建局域网,在构建的时候问题百…

2024年云服务器ECS价格表出炉——腾讯云

腾讯云服务器多少钱一年?61元一年起。2024年最新腾讯云服务器优惠价格表,腾讯云轻量2核2G3M服务器61元一年、2核2G4M服务器99元一年可买三年、2核4G5M服务器165元一年、3年756元、轻量4核8M12M服务器646元15个月、4核16G10M配置32元1个月、312元一年、8核…

用Python做一个植物大战僵尸

植物大战僵尸是一个相对复杂的游戏,涉及到图形界面、动画、游戏逻辑等多个方面。用Python实现一个完整的植物大战僵尸游戏是一个大工程,但我们可以简化一些内容,做一个基础版本。 以下是一个简化版的植物大战僵尸游戏的Python实现思路&#…

成功案例|全基因组测序+GWAS联合分析揭示不同种族帕金森病的遗传同质性和异质性

发表期刊:npj Parkinson’s Disease 影响因子:8.7 测序方式:WGS 研究对象:人 1 研究背景 帕金森病(PD)是一种常见的与年龄相关的神经退行性疾病,其特征是运动迟缓、姿势不稳定、僵硬和静息…

Redis中的过期键删除策略

过期键删除策略 概述 数据库键的过期时间都保存在过期字典中,并且知道根据过期时间去判断一个键是否过期,剩下的问题是:如果一个键过期了,那么它什么时候会被删除呢? 这个问题有三种可能的答案,它们分别代表了三种不同的删除策…

【linux】进程的地址空间

1.代码看现象引入 #include<stdio.h>#include<unistd.h>#include<string.h> #include<stdlib.h>int val100;int main (){ printf("i am father,pid:%d,ppid:%d,val:%d&#xff0c;&val:%p\n",getpid(),getppid(),val,&val);size_t…

vue2 和 vue3 配置路由有什么区别

vue2 和 vue3 配置路由有什么区别 初始化路由器实例&#xff1a;注入到应用中&#xff1a;动态路由参数和捕获所有路由&#xff1a;编程式导航 API&#xff1a;异步加载组件&#xff1a; vue2 如何 使用路由 第一步&#xff1a;安装 vue-router第二步&#xff1a;创建路由组件第…

【k8s】kubeasz 3.6.3 + virtualbox 搭建本地虚拟机openeuler 22.03 三节点集群 离线方案

kubeasz项目源码地址 GitHub - easzlab/kubeasz: 使用Ansible脚本安装K8S集群&#xff0c;介绍组件交互原理&#xff0c;方便直接&#xff0c;不受国内网络环境影响 拉取代码&#xff0c;并切换到最近发布的分支 git clone https://github.com/easzlab/kubeasz cd kubeasz gi…

<Linux> 模拟实现文件流 - 简易版

目录 1. FILE 结构设计 2、函数使用及分析 3、文件打开 fopen 4. 缓冲区刷新fflush 5. 数据写入fwrite 6. 文件关闭 fclose 7. 测试 8. 小结 1. FILE 结构设计 在设计 FILE 结构体前&#xff0c;首先要清楚 FILE 中有自己的缓冲区及冲刷方式 缓冲区的大小和刷新方式因…

JVM监控工具

JVM监控工具 文章目录 JVM监控工具jpsjmapjmap -histo 进程idjmap -heap 进程id (查看堆信息)jmap -dump:formatb,filefilename.hprof 进程id (将堆当前时刻快照信息dump到文件中) JSTACKjstack 查看死锁信息jstack找出占用cpu最高的线程堆栈信息 jinfo查看jvm参数查看java系统…

Perfetto Trace抓取

1. Perfetto简介 Perfetto 是一个用于 Android 系统的性能跟踪工具&#xff0c;可以帮助开发者分析系统性能和调试问题。 Perfetto 是 Android 10 中引入的全新平台级跟踪工具。这是适用于 Android、Linux 和 Chrome 的更加通用和复杂的开源跟踪项目。 在低于Android R的版本上…

量子计算新“尺度”:用经典计算机评估复杂量子系统!

未来的量子计算机有望在计算机科学、医疗、商业、化学、物理学等多个领域解决难题&#xff0c;从而超越传统计算机。然而&#xff0c;目前的量子计算机仍存在局限&#xff0c;主要是由于它们固有的错误率。为此&#xff0c;研究者正致力于降低这些错误率。 一种研究量子计算机误…

Linux系统部署Paperless-Ngx文档管理系统结合内网穿透实现公网访问

文章目录 1. 部署Paperless-ngx2. 本地访问Paperless-ngx3. Linux安装Cpolar4. 配置公网地址5. 远程访问6. 固定Cpolar公网地址7. 固定地址访问 Paperless-ngx是一个开源的文档管理系统&#xff0c;可以将物理文档转换成可搜索的在线档案&#xff0c;从而减少纸张的使用。它内置…

机械硬盘与固态硬盘究竟该适合选用哪种,看完本文你就了解了!

随着科技的发展,计算机存储技术经历了从传统的机械硬盘(HDD)到现代固态硬盘(SSD)的革新变迁。在这篇文章中,我们将深入探讨机械硬盘与固态硬盘在功能特点上的显著区别,帮助用户更好地理解这两种存储设备的核心优势与不足。 一、存储原理与结构差异 1. 机械硬盘(HDD) …

微服务架构的实现:选择最佳方案,构建未来的应用生态(二)

本系列文章简介&#xff1a; 在本系列文章中&#xff0c;我们将探讨微服务架构的实现&#xff0c;并提供一些实用的建议和最佳实践。我们将从架构设计开始&#xff0c;介绍各种微服务架构模式和组件&#xff0c;包括服务发现、负载均衡、服务网关、事件驱动等。我们还将讨论微服…

ffmpeg开发异步AI推理Filter

ffmpeg开发异步AI推理Filter 1.环境搭建、推理服务及客户端SDK2.编译原版ffmpeg3.测试原版ffmpeg的filter功能4.准备异步推理filter5.修改点6.重新编译ffmpeg7.测试异步推理filter本文旨在阐述如何开发一个FFmpeg Filter,该模块利用gRPC异步通信机制调用远程视频处理服务。这一…

Midjourney AI绘图工具介绍及使用

介绍 Midjourney是一款目前被誉为最强的AI绘图工具。只要输入想到的文字&#xff0c;就能通过人工智能产出相对应的图片。 官网只是宣传和登录入口&#xff0c;提供个人主页、订阅管理等功能&#xff0c;Midjourney实际的绘画功能&#xff0c;是在另外一个叫discord的产品中实…

网络电视盒子哪个品牌好?2024畅销电视盒子排行榜

电视盒子的品牌和产品非常多&#xff0c;让新手在选购时难度增大&#xff0c;大部分消费者在此时会选择参考销量排名情况&#xff0c;小编这次结合各个电商平台的销量和用户评价整理了电视盒子排行榜&#xff0c;想买电视盒子不知道网络电视盒子哪个品牌好可以收藏。 TOP 1.泰捷…

论文导读 | 漫谈图神经网络

本文主要介绍图神经网络相关内容&#xff0c;包括图神经网络的基本结构以及近期研究进展。 背景 在实际生活中&#xff0c;许多数据都可以用图的形式表达&#xff0c;比如社交网络、分子模型、知识图谱、计算机网络等。图深度学习旨在&#xff0c;显式利用这些数据中的拓扑结…