介绍

大家好，博主又来给大家分享知识了。今天给大家分享的内容是使用LangChain进行大规模应用开发中的LangGraph快速构建Agent工作流应用。

通过对前几次对LangChain的技术分享。我们知道LangChain作为一个强大的工具集，为开发者们提供了丰富的资源和便捷的开发途径。而其中的LangGraph更是独具特色，它就像是一座桥梁，能够巧妙地连接起各个智能体组件，让它们协同工作，高效地完成复杂的任务。

通过LangGraph，我们可以轻松实现有状态、多参与者的应用程序构建，极大地提高开发效率，降低开发成本。接下来，我就详细地为大家介绍如何利用LangGraph来快速构建Agent工作流应用，希望能为大家的在实际开发中带来新的思路和启发。

Agent工具流

Agent工作流是指多个智能体(Agent)按照一定的逻辑顺序和规则，相互协作、交互并利用各种工具来完成复杂任务的流程体系。

大家在使用智能体(Agent)完成复杂任务时，有一种很实用的思路，就是创建“计划并执行”风格的智能体。这种方式在不少前沿研究，比如一些计划和解决相关的学术论文，还有Baby - AGI项目中都有应用和探索。

大家可选看

• Baby-AGI：它是由Yohei Nakajima开发的一个创新性Python脚本，是人工智能驱动的任务管理系统示例，旨在探索通用人工智能(AGI)的边界 。在应用场景上，它可用于个人任务管理、团队协作、自动化项目管理、客户支持自动化以及教育领域等多个方面。

它(计划并执行)的核心逻辑并不复杂。当接到任务后，智能体会先制定一个包含多个步骤的详细计划，就像我们平时做项目列步骤清单一样。制定好计划后，智能体便按照顺序一项一项地执行这些步骤。而且在完成某一项具体任务后，智能体会像我们检查作业一样，重新看看整个计划，要是发现有不合适的地方，就及时修改，以保证最终能把任务圆满完成。如下图所示：

我们来对比下“计划并执行”风格代理和典型的ReAct风格代理。ReAct风格就像走一步看一步，每次只思考当前这一步该怎么做。而“计划并执行”风格不同，它会先把整个任务的步骤都规划好，然后再按计划行动。
这种“计划并执行”风格的代理有两大明显优势：

• 一是能做出清晰的长期规划。通常，哪怕是很强大的大语言模型(LLM)，在长期规划方面也会有些吃力，但 “计划并执行” 风格能很好地做到这一点。
• 二是在执行具体步骤时，它可以使用相对较小、较弱的模型，只在制定计划的阶段，才启用更大、性能更好的模型。这样既能保证规划的质量，又能在执行时节省资源。

接下来，我会通过在LangGraph中的演练，为大家展示怎么实现“计划并执行”风格的代理。

安装相关库

安装命令

python">pip install -U langgraph langchain-community langchain-openai

配置环境变量

详细环境变量配置，请看博主的这篇博文LangChain大模型应用开发：构建Agent智能体-CSDN博客中的LangSmith和定义工具中的Tavily。

定义工具

我们先来看工具定义这一步。在这个简单示例里，我们们会用到搜索工具Tavily。借助这个工具，能方便地获取各类信息。

Tavily工具演示代码

python"># 从langchain_community工具模块中导入Tavily搜索结果工具类
from langchain_community.tools.tavily_search import TavilySearchResults

# 创建一个包含Tavily搜索结果工具实例的列表，设置最大结果数为1
tools = [TavilySearchResults(max_results=1)]

定义执行智能体

接下来，我们要创建执行任务的执行代理。在这个示例里，为了便于演示，每个任务都将使用同一个执行代理。

不过大家要知道的是，这只是讲解中的一种选择，并非固定规则。在实际的复杂场景或不同类型的任务中，我们可以根据任务的特点、需求和复杂程度等因素，灵活选择使用不同的执行代理，以更好地完成各类任务 。

python"># 从langchain库中导入hub模块，用于从LangChain Hub拉取资源
from langchain import hub
# 从langchain_openai库中导入ChatOpenAI类，用于调用OpenAI的聊天模型
from langchain_openai import ChatOpenAI
# 从langgraph.prebuilt模块导入create_react_agent函数，用于创建ReAct风格的智能体
from langgraph.prebuilt import create_react_agent

# 从LangChain Hub拉取名为"wfh/react-agent-executor"的提示模板
prompt = hub.pull("wfh/react-agent-executor")
# 以美观的格式打印拉取到的提示模板内容
prompt.pretty_print()

# 创建一个ChatOpenAI实例，指定使用"gpt-3.5-turbo"模型
chat_model = ChatOpenAI(model="gpt-3.5-turbo")
# 调用create_react_agent函数，结合大语言模型、工具列表和提示模板修改器，创建一个ReAct智能体执行器
agent_executor = create_react_agent(chat_model, tools, messages_modifier=prompt)

我开始对智能体进行测试，运行下代码。

智能体测试代码

python"># 从langchain_community工具模块中导入Tavily搜索结果工具类
from langchain_community.tools.tavily_search import TavilySearchResults
# 从langchain库中导入hub模块，用于从LangChain Hub拉取资源
from langchain import hub
# 从langchain_openai库中导入ChatOpenAI类，用于调用OpenAI的聊天模型
from langchain_openai import ChatOpenAI
# 从langgraph.prebuilt模块导入create_react_agent函数，用于创建ReAct风格的智能体
from langgraph.prebuilt import create_react_agent

# 创建一个包含Tavily搜索结果工具实例的列表，设置最大结果数为1
tools = [TavilySearchResults(max_results=1)]

# 从LangChain Hub拉取名为"wfh/react-agent-executor"的提示模板
prompt = hub.pull("wfh/react-agent-executor")
# 以美观的格式打印拉取到的提示模板内容
prompt.pretty_print()

# 创建一个ChatOpenAI实例，指定使用"gpt-3.5-turbo"模型
chat_model = ChatOpenAI(model="gpt-3.5-turbo")
# 调用create_react_agent函数，结合大语言模型、工具列表和提示模板修改器，创建一个ReAct智能体执行器
agent_executor = create_react_agent(chat_model, tools, messages_modifier=prompt)
# 调用智能体执行器，询问问题
response = agent_executor.invoke({"messages": [("user", "中国的首都是哪里")]})
# 打印此次交互的响应结果
print(response)

智能体测试结果

python">================================ System Message ================================

You are a helpful assistant.

============================= Messages Placeholder =============================

{{messages}}
{'messages': [HumanMessage(content='中国的首都是哪里', additional_kwargs={}, response_metadata={}, id='2d383eef-c524-4cbc-9b41-cbc3b83ad38b'), AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_7kEYjvj9Dl1WDK8fWrwAjqNd', 'function': {'arguments': '{"query":"中国的首都是哪里"}', 'name': 'tavily_search_results_json'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 25, 'prompt_tokens': 95, 'total_tokens': 120, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-9415da3f-4ab1-446c-897c-51977438df09-0', tool_calls=[{'name': 'tavily_search_results_json', 'args': {'query': '中国的首都是哪里'}, 'id': 'call_7kEYjvj9Dl1WDK8fWrwAjqNd', 'type': 'tool_call'}], usage_metadata={'input_tokens': 95, 'output_tokens': 25, 'total_tokens': 120, 'input_token_details': {}, 'output_token_details': {}}), ToolMessage(content='[{"url": "https://www.gov.cn/guoqing/2005-05/24/content_2615214.htm", "content": "中华人民共和国首都_中华人民共和国中央人民政府门户网站 简体 | 繁体 | English 2025年2月8日 星期六   公务邮箱 旧版回顾 国务院 新闻 专题 政策 服务 问政 数据 国情 首页> 国情 中华人民共和国首都 中央政府门户网站\u3000www.gov.cn 来源： 中国政府网 【字体：大 中 小】打印本页 分享 新华微博 人民微博 新浪微博 \xa0 １９４９年９月２７日，中国人民政治协商会议第一届全体会议一致通过中华人民共和国的国都定于北平，即日起北平改名北京。\u3000\u3000北京，简称京，是中国共产党中央委员会、中华人民共和国中央人民政府所在地。中央四个直辖市之一，是全国政治、经济和科学文化的中心，也是国内国际交往的中心之一，是中国历史文化名城和古都之一。\u3000\xa0定都北京的经过\xa0\xa0\xa0 １９４８年党中央关于召集新政协的《五一宣言》发表后，各民主党派和无党派爱国民主人士纷纷响应。同年９月，党中央计划在１９４９年下半年成立中央政府。不久，东北战场辽西大捷，毛主席开始部署第四野战军的部队南下夺取平津。这一年的１１月８日，党中央决定：在北平解放后，由薄一波先行赴平，为党中央机关进驻北平打前站。\u3000\u3000就在党中央作出这一重大决定的时候，北平已被解放军包围。１９４９年１月３１日，北平和平解放，古都获得了新生。２月３日，解放军入城仪式在前门大街举行，北平各界群众沸腾了，举城同庆，迎接这一神圣时刻的到来。\u3000\u3000１９４９年３月２３日上午，毛泽东、朱德、刘少奇、周恩来、任弼时率中共中央机关离开西柏坡。２５日凌晨６时，毛泽东等中央领导人乘专列抵达清华园火车站，下午赴西苑机场阅兵，受到各界民主人士的热烈欢迎。６月１５日，中国新政治协商会议筹备会在北平召开。次日，周恩来主持筹备会常委会第一次会议，会议决定在常委会领导下设立六个小组。其中第六小组的任务是研究草拟国旗、国徽、国歌、纪年、国都等方案，组长是中国著名教育家、中国民主促进会负责人马叙伦，副组长是北平军管会主任叶剑英，不久又增加沈雁冰任副组长，组员有张奚若、田汉、马寅初、郭沫若、廖承志等１６人。经过４次讨论，第六小组于９月１４日一致提出建都北平，改名为北京。\u3000\u3000１９４９年９月中旬，毛泽东等中央领导由香山移居中南海，并出席了政协筹备会第二次会议。９月２１日，中国人民政治协商会议第一届全体会议在中南海怀仁堂举行，２７日大会讨论《国旗、国都、纪年、国歌决议草案》，并逐项进行表决，全体代表以举手和热烈的掌声通过四个决议案：一、中华人民共和国国都定于北平，自即日起北平改名为北京。二、中华人民共和国的纪年采用公元，本年为一九四九年。三、在中华人民共和国的国歌未正式制定前，以《义勇军进行曲》为国歌。四、中华人民共和国的国旗为红地五星旗，象征中国革命人民大团结。\u3000\u3000在表决前，沈雁冰汇报了第六小组的研究讨论意见，提出了定都北平的理由：“国民党反动派过去定都南京，主要原因是在政治上和经济上，便于依赖帝国主义，因为南京靠近上海，而上海是帝国主义和买办资产阶级剥削中国人民的中心城市。中华人民共和国为人民自己的国家，它依靠的是中国人民，自不一定要建都南京了。北平为中国的首都已有七百多年的历史。在政治上，北平位于华北老解放区内，人民力量雄厚，规模弘伟，文物集中，是世界上有名的历史的大都市之一，且自五四以来，这里就是新文化思想的摇篮。\xa0\u3000\u3000此外，在地理上，北平位于一个大平原之中，将来有足够的扩充的余地，在交通上是四通八达，有平沈、平绥、平汉、平沪等铁路干线，连络全国各地。总之从各种条件看，北平实具备现代大国首都的各种资格。因此，我们提议，中华人民共和国应以北平为首都，并改名为北京。”\u3000\u3000当晚电台播出这一振奋人心的消息后，北京城鞭炮齐鸣，热烈庆贺。\u3000\u3000１９４９年９月２９日，中国人民政治协商会议宣布自己执行全国人民代表大会的职权，一致通过了《中国人民政治协商共同纲领》。１０月１日，毛泽东主席在天安门城楼上向全世界庄严宣布，中华人民共和国中央人民政府成立了！从此，作为新中国首都的北京与共和国一起，在中华民族的历史长河中翻开了崭新的一页。 全国人大全国政协最高法院最高检察院 版权所有：中国政府网 | 关于我们 | 网站声明 | 网站地图 | 联系我们 京ICP备05070218号 中文域名：中国政府网.政务 中国政府网 微博、微信"}]', name='tavily_search_results_json', id='faf095c1-2f98-4da1-a086-2c050ff865cc', tool_call_id='call_7kEYjvj9Dl1WDK8fWrwAjqNd', artifact={'query': '中国的首都是哪里', 'follow_up_questions': None, 'answer': None, 'images': [], 'results': [{'url': 'https://www.gov.cn/guoqing/2005-05/24/content_2615214.htm', 'title': '中华人民共和国首都_中华人民共和国中央人民政府门户网站', 'content': '中华人民共和国首都_中华人民共和国中央人民政府门户网站 简体 | 繁体 | English 2025年2月8日 星期六   公务邮箱 旧版回顾 国务院 新闻 专题 政策 服务 问政 数据 国情 首页> 国情 中华人民共和国首都 中央政府门户网站\u3000www.gov.cn 来源： 中国政府网 【字体：大 中 小】打印本页 分享 新华微博 人民微博 新浪微博 \xa0 １９４９年９月２７日，中国人民政治协商会议第一届全体会议一致通过中华人民共和国的国都定于北平，即日起北平改名北京。\u3000\u3000北京，简称京，是中国共产党中央委员会、中华人民共和国中央人民政府所在地。中央四个直辖市之一，是全国政治、经济和科学文化的中心，也是国内国际交往的中心之一，是中国历史文化名城和古都之一。\u3000\xa0定都北京的经过\xa0\xa0\xa0 １９４８年党中央关于召集新政协的《五一宣言》发表后，各民主党派和无党派爱国民主人士纷纷响应。同年９月，党中央计划在１９４９年下半年成立中央政府。不久，东北战场辽西大捷，毛主席开始部署第四野战军的部队南下夺取平津。这一年的１１月８日，党中央决定：在北平解放后，由薄一波先行赴平，为党中央机关进驻北平打前站。\u3000\u3000就在党中央作出这一重大决定的时候，北平已被解放军包围。１９４９年１月３１日，北平和平解放，古都获得了新生。２月３日，解放军入城仪式在前门大街举行，北平各界群众沸腾了，举城同庆，迎接这一神圣时刻的到来。\u3000\u3000１９４９年３月２３日上午，毛泽东、朱德、刘少奇、周恩来、任弼时率中共中央机关离开西柏坡。２５日凌晨６时，毛泽东等中央领导人乘专列抵达清华园火车站，下午赴西苑机场阅兵，受到各界民主人士的热烈欢迎。６月１５日，中国新政治协商会议筹备会在北平召开。次日，周恩来主持筹备会常委会第一次会议，会议决定在常委会领导下设立六个小组。其中第六小组的任务是研究草拟国旗、国徽、国歌、纪年、国都等方案，组长是中国著名教育家、中国民主促进会负责人马叙伦，副组长是北平军管会主任叶剑英，不久又增加沈雁冰任副组长，组员有张奚若、田汉、马寅初、郭沫若、廖承志等１６人。经过４次讨论，第六小组于９月１４日一致提出建都北平，改名为北京。\u3000\u3000１９４９年９月中旬，毛泽东等中央领导由香山移居中南海，并出席了政协筹备会第二次会议。９月２１日，中国人民政治协商会议第一届全体会议在中南海怀仁堂举行，２７日大会讨论《国旗、国都、纪年、国歌决议草案》，并逐项进行表决，全体代表以举手和热烈的掌声通过四个决议案：一、中华人民共和国国都定于北平，自即日起北平改名为北京。二、中华人民共和国的纪年采用公元，本年为一九四九年。三、在中华人民共和国的国歌未正式制定前，以《义勇军进行曲》为国歌。四、中华人民共和国的国旗为红地五星旗，象征中国革命人民大团结。\u3000\u3000在表决前，沈雁冰汇报了第六小组的研究讨论意见，提出了定都北平的理由：“国民党反动派过去定都南京，主要原因是在政治上和经济上，便于依赖帝国主义，因为南京靠近上海，而上海是帝国主义和买办资产阶级剥削中国人民的中心城市。中华人民共和国为人民自己的国家，它依靠的是中国人民，自不一定要建都南京了。北平为中国的首都已有七百多年的历史。在政治上，北平位于华北老解放区内，人民力量雄厚，规模弘伟，文物集中，是世界上有名的历史的大都市之一，且自五四以来，这里就是新文化思想的摇篮。\xa0\u3000\u3000此外，在地理上，北平位于一个大平原之中，将来有足够的扩充的余地，在交通上是四通八达，有平沈、平绥、平汉、平沪等铁路干线，连络全国各地。总之从各种条件看，北平实具备现代大国首都的各种资格。因此，我们提议，中华人民共和国应以北平为首都，并改名为北京。”\u3000\u3000当晚电台播出这一振奋人心的消息后，北京城鞭炮齐鸣，热烈庆贺。\u3000\u3000１９４９年９月２９日，中国人民政治协商会议宣布自己执行全国人民代表大会的职权，一致通过了《中国人民政治协商共同纲领》。１０月１日，毛泽东主席在天安门城楼上向全世界庄严宣布，中华人民共和国中央人民政府成立了！从此，作为新中国首都的北京与共和国一起，在中华民族的历史长河中翻开了崭新的一页。 全国人大全国政协最高法院最高检察院 版权所有：中国政府网 | 关于我们 | 网站声明 | 网站地图 | 联系我们 京ICP备05070218号 中文域名：中国政府网.政务 中国政府网 微博、微信', 'score': 0.8319231, 'raw_content': None}], 'response_time': 1.2}), AIMessage(content='中国的首都是北京。中华人民共和国的首都是北京，也是中国共产党中央委员会、中华人民共和国中央人民政府的所在地。北京是全国政治、经济和科学文化的中心，同时也是国内国际交往的中心之一，以及中国历史文化名城和古都之一。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 109, 'prompt_tokens': 2191, 'total_tokens': 2300, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None}, id='run-7da2a023-762f-4a26-ba85-bada6d21fc92-0', usage_metadata={'input_tokens': 2191, 'output_tokens': 109, 'total_tokens': 2300, 'input_token_details': {}, 'output_token_details': {}})]}

进程已结束，退出代码为 0

定义状态

在开发和使用这个代理时，定义其状态是关键的第一步，我们通过记录几个关键信息来跟踪它的运行情况。

• 首先是当前计划，它就像代理的行动指南。我们可以用字符串列表的形式来呈现，每一个字符串代表计划中的一个步骤，这样既直观又便于管理。
• 接着是先前执行的步骤，这些步骤及其产生的结果是了解代理工作过程的重要依据。我们把它们记录为元组列表，每个元组都清晰地包含步骤内容和对应的结果，方便后续追溯和分析。
• 最后，原始输入是代理执行任务的起点，最终响应则是任务完成的输出。为了全面掌握代理的工作全貌，我们也需要记录这两部分信息，它们构成了代理工作的起始和结束状态 。

演示代码

python"># 导入operator模块，这里后续用于类型注解中的特定操作
import operator
# 从typing模块导入Annotated、List、Tuple、TypedDict用于类型注解
from typing import Annotated, List, Tuple, TypedDict

# 定义一个名为PlanExecute的TypedDict类，用于表示特定的数据结构
class PlanExecute(TypedDict):
    # 定义一个名为input的字段，类型为字符串，代表输入信息
    input: str
    # 定义一个名为plan的字段，类型为字符串列表，代表计划步骤
    plan: List[str]
    # 定义一个名为past_steps的字段，类型为经过Annotated注解的元组列表，使用operator.add作为额外元数据
    past_steps: Annotated[List[Tuple], operator.add]
    # 定义一个名为response的字段，类型为字符串，代表响应信息
    response: str

规划步骤

接下来进入规划步骤的创建环节。在这个过程中，我们将借助方法调用来生成任务计划。

演示代码

python"># 从 pydantic模块导入BaseModel和Field类，用于定义数据模型
from pydantic import BaseModel, Field
# 从langchain_core.prompts模块导入ChatPromptTemplate类，用于创建聊天提示模板
from langchain_core.prompts import ChatPromptTemplate


# 定义一个名为Plan的数据模型类，继承自BaseModel
class Plan(BaseModel):
    """未来要执行的计划"""
    # 定义steps字段，类型为字符串列表，使用Field进行描述说明
    steps: List[str] = Field(
        # 说明steps是需要按顺序排列的不同执行步骤
        description="需要执行的不同步骤，应该按顺序排列"
    )


# 创建一个计划生成的提示模板
planner_prompt = ChatPromptTemplate.from_messages(
    [
        (
            # 定义系统消息，为生成计划提供规则和要求
            "system",
            """对于给定的目标，提出一个简单的逐步计划。这个计划应该包含独立的任务，如果正确执行将得出正确的答案。不要添加任何多余的步骤。最后一步的结果应该是最终答案。确保每一步都有所有必要的信息 - 不要跳过步骤。""",
        ),
        # 定义占位符消息，用于后续填充实际的用户输入
        ("placeholder", "{messages}"),
    ]
)

# 使用指定的提示模板创建一个计划生成器，使用OpenAI的GPT模型
planner = planner_prompt | ChatOpenAI(
    # 指定使用模型
    model="gpt-3.5-turbo",
    # 设置temperature=0不考虑随机性
    temperature=0
).with_structured_output(Plan)

# 调用计划生成器，传入用户的问题信息
planner.invoke(
    {
        "messages": [
            # 用户提出的问题
            ("user", "现任中国斯诺克冠军是哪的人?")
        ]
    }
)

规划步骤测试完整代码

python"># 从typing模块导入List用于类型注解
from typing import List
# 从langchain_openai库中导入ChatOpenAI类，用于调用OpenAI的聊天模型
from langchain_openai import ChatOpenAI
# 从 pydantic模块导入BaseModel和Field类，用于定义数据模型
from pydantic import BaseModel, Field
# 从langchain_core.prompts模块导入ChatPromptTemplate类，用于创建聊天提示模板
from langchain_core.prompts import ChatPromptTemplate


# 定义一个名为Plan的数据模型类，继承自BaseModel
class Plan(BaseModel):
    """未来要执行的计划"""
    # 定义steps字段，类型为字符串列表，使用Field进行描述说明
    steps: List[str] = Field(
        # 说明steps是需要按顺序排列的不同执行步骤
        description="需要执行的不同步骤，应该按顺序排列"
    )


# 创建一个计划生成的提示模板
planner_prompt = ChatPromptTemplate.from_messages(
    [
        (
            # 定义系统消息，为生成计划提供规则和要求
            "system",
            """对于给定的目标，提出一个简单的逐步计划。这个计划应该包含独立的任务，如果正确执行将得出正确的答案。不要添加任何多余的步骤。最后一步的结果应该是最终答案。确保每一步都有所有必要的信息 - 不要跳过步骤。""",
        ),
        # 定义占位符消息，用于后续填充实际的用户输入
        ("placeholder", "{messages}"),
    ]
)

# 使用指定的提示模板创建一个计划生成器，使用OpenAI的GPT模型
planner = planner_prompt | ChatOpenAI(
    # 指定使用模型
    model="gpt-4o",
    # 设置temperature=0不考虑随机性
    temperature=0
).with_structured_output(Plan)

# 调用计划生成器，传入用户的问题信息
response = planner.invoke(
    {
        "messages": [
            # 用户提出的问题
            ("user", "现任中国斯诺克冠军是哪的人?")
        ]
    }
)
# 打印此次交互的响应结果
print(response)

规划步骤测试运行结果

python">steps=['规定要找到现任中国斯诺克冠军的身份。', '在可信来源中查找有关中国斯诺克冠军的信息。', '确认他们的获胜身份和籍贯。', '如果适用，给出答案，例如“现任中国斯诺克冠军是出生于xx地方的选手。”。']

进程已结束，退出代码为 0

重新规划步骤

接下来，我们进入重新规划步骤的环节。在此之前，我们已经完成了一些操作并得到了相应结果。 

首先，我们需要准确获取上一步骤产生的结果数据。这些数据将成为重新规划的重要依据，比如结果中的关键信息、完成情况、出现的问题等。

然后，依据特定的规则和目标，对这些结果进行分析。若上一步结果显示某些任务未达成预期，或是发现了新的情况，我们就需要针对性地调整原计划，新增、修改或删除一些步骤，从而制定出更契合实际情况的新计划 。

python"># 从typing模块导入Union类型，用于定义联合类型注解
from typing import Union


# 定义一个名为Response的数据模型类，继承自BaseModel
class Response(BaseModel):
    """用户响应"""
    # 定义response字段，类型为字符串，代表用户的响应内容
    response: str


# 定义一个名为Act的数据模型类，继承自BaseModel
class Act(BaseModel):
    """要执行的行为"""
    # 定义action字段，类型为Response或Plan的联合类型，使用Field进行描述说明
    action: Union[Response, Plan] = Field(
        # 说明action字段的用途，根据情况选择使用Response或Plan
        description="要执行的行为。如果要回应用户，使用Response。如果需要进一步使用工具获取答案，使用Plan。"
    )


# 创建一个重新计划的提示模板
replanner_prompt = ChatPromptTemplate.from_template(
    """
        对于给定的目标，提出一个简单的逐步计划。这个计划应该包含独立的任务，如果正确执行将得出正确的答案。不要添加任何多余的步骤。最后一步的结果应该是最终答案。确保每一步都有所有必要的信息 - 不要跳过步骤。

        你的目标是：
        {input}

        你的原计划是：
        {plan}

        你目前已完成的步骤是：
        {past_steps}

        相应地更新你的计划。如果不需要更多步骤并且可以返回给用户，那么就这样回应。如果需要，填写计划。只添加仍然需要完成的步骤。不要返回已完成的步骤作为计划的一部分。
    """
)
# 使用重新计划的提示模板和OpenAI模型创建一个重新计划器
replanner = replanner_prompt | ChatOpenAI(
    # 指定使用模型
    model="gpt-4o",
    # 设置temperature=0不考虑随机性
    temperature=0
).with_structured_output(Act)

创建Graph

借助这些已有的成果，我们可以进一步探索如何将这些信息进行可视化呈现。具体来说，我们可以尝试创建一个图来展示计划的执行过程。

创建Graph演示代码

python"># 从langchain_community工具模块中导入Tavily搜索结果工具类
from langchain_community.tools.tavily_search import TavilySearchResults
# 从langchain库中导入hub模块，用于从LangChain Hub拉取资源
from langchain import hub
# 从langchain_openai库中导入ChatOpenAI类，用于调用OpenAI的聊天模型
from langchain_openai import ChatOpenAI
# 从langgraph.prebuilt模块导入create_react_agent函数，用于创建ReAct风格的智能体
from langgraph.prebuilt import create_react_agent
# 导入operator模块，这里后续用于类型注解中的特定操作
import operator
# 从typing模块导入Annotated、List、Tuple、TypedDict用于类型注解
from typing import Annotated, List, Tuple, TypedDict, Literal
# 从 pydantic模块导入BaseModel和Field类，用于定义数据模型
from pydantic import BaseModel, Field
# 从langchain_core.prompts模块导入ChatPromptTemplate类，用于创建聊天提示模板
from langchain_core.prompts import ChatPromptTemplate
# 从typing模块导入Union类型，用于定义联合类型注解
# 导入asyncio库，用于编写异步I/O、协程等异步代码，支持高效的并发操作
import asyncio
# 从typing模块导入Union类型，用于定义变量可以是多种类型中的任意一种
from typing import Union
# 从langgraph.graph模块导入StateGraph类和START常量，前者可能用于构建状态图，后者或表示起始状态
from langgraph.graph import StateGraph, START


# 定义一个名为PlanExecute的TypedDict类，用于表示特定的数据结构
class PlanExecute(TypedDict):
    # 定义一个名为input的字段，类型为字符串，代表输入信息
    input: str
    # 定义一个名为plan的字段，类型为字符串列表，代表计划步骤
    plan: List[str]
    # 定义一个名为past_steps的字段，类型为经过Annotated注解的元组列表，使用operator.add作为额外元数据
    past_steps: Annotated[List[Tuple], operator.add]
    # 定义一个名为response的字段，类型为字符串，代表响应信息
    response: str


# 定义一个名为Plan的数据模型类，继承自BaseModel
class Plan(BaseModel):
    """未来要执行的计划"""
    # 定义steps字段，类型为字符串列表，使用Field进行描述说明
    steps: List[str] = Field(
        # 说明steps是需要按顺序排列的不同执行步骤
        description="需要执行的不同步骤，应该按顺序排列"
    )


# 定义一个名为Response的数据模型类，继承自BaseModel
class Response(BaseModel):
    """用户响应"""
    # 定义response字段，类型为字符串，代表用户的响应内容
    response: str


# 定义一个名为Act的数据模型类，继承自BaseModel
class Act(BaseModel):
    """要执行的行为"""
    # 定义action字段，类型为Response或Plan的联合类型，使用Field进行描述说明
    action: Union[Response, Plan] = Field(
        # 说明action字段的用途，根据情况选择使用Response或Plan
        description="要执行的行为。如果要回应用户，使用Response。如果需要进一步使用工具获取答案，使用Plan。"
    )


# 创建一个包含Tavily搜索结果工具实例的列表，设置最大结果数为1
tools = [TavilySearchResults(max_results=1)]
# 从LangChain Hub拉取名为"wfh/react-agent-executor"的提示模板
prompt = hub.pull("wfh/react-agent-executor")
# 创建一个ChatOpenAI实例，指定使用模型
chat_model = ChatOpenAI(model="gpt-3.5-turbo")
# 调用create_react_agent函数，结合大语言模型、工具列表和提示模板修改器，创建一个ReAct智能体执行器
agent_executor = create_react_agent(chat_model, tools, messages_modifier=prompt)

# 创建一个计划生成的提示模板
planner_prompt = ChatPromptTemplate.from_messages(
    [
        (
            # 定义系统消息，为生成计划提供规则和要求
            "system",
            """对于给定的目标，提出一个简单的逐步计划。这个计划应该包含独立的任务，如果正确执行将得出正确的答案。不要添加任何多余的步骤。最后一步的结果应该是最终答案。确保每一步都有所有必要的信息 - 不要跳过步骤。""",
        ),
        # 定义占位符消息，用于后续填充实际的用户输入
        ("placeholder", "{messages}"),
    ]
)

# 使用指定的提示模板创建一个计划生成器，使用OpenAI的GPT模型
planner = planner_prompt | ChatOpenAI(
    # 指定使用模型
    model="gpt-3.5-turbo",
    # 设置temperature=0不考虑随机性
    temperature=0
).with_structured_output(Plan)

# 调用计划生成器，传入用户的问题信息
response = planner.invoke(
    {
        "messages": [
            # 用户提出的问题
            ("user", "2022年NBA冠军是哪支队伍")
        ]
    }
)

# 创建一个重新计划的提示模板
replanner_prompt = ChatPromptTemplate.from_template(
    """
        对于给定的目标，提出一个简单的逐步计划。这个计划应该包含独立的任务，如果正确执行将得出正确的答案。不要添加任何多余的步骤。最后一步的结果应该是最终答案。确保每一步都有所有必要的信息 - 不要跳过步骤。

        你的目标是：
        {input}

        你的原计划是：
        {plan}

        你目前已完成的步骤是：
        {past_steps}

        相应地更新你的计划。如果不需要更多步骤并且可以返回给用户，那么就这样回应。如果需要，填写计划。只添加仍然需要完成的步骤。不要返回已完成的步骤作为计划的一部分。
    """
)

# 使用重新计划的提示模板和OpenAI模型创建一个重新计划器
replanner = replanner_prompt | ChatOpenAI(
    # 指定使用模型
    model="gpt-3.5-turbo",
    # 设置temperature=0不考虑随机性
    temperature=0
).with_structured_output(Act)


# 定义一个异步函数，用于执行计划中的单个步骤
async def execute_step(state: PlanExecute):
    # 从状态中获取当前计划
    plan = state["plan"]
    # 将计划中的步骤转换为格式化的字符串，方便后续展示
    plan_str = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(plan))
    # 取出计划中的第一个任务
    task = plan[0]
    # 格式化任务描述，包含完整计划和当前要执行的步骤
    task_formatted = f"""对于以下计划：
                        {plan_str}\n\n你的任务是执行第{1}步，{task}。
                       """
    # 异步调用智能体执行器，传入格式化后的任务描述
    agent_response = await agent_executor.ainvoke(
        {"messages": [("user", task_formatted)]}
    )
    # 返回更新后的状态，包含已执行的步骤和对应的响应
    return {
        "past_steps": state["past_steps"] + [(task, agent_response["messages"][-1].content)],
    }


# 定义一个异步函数，用于生成初始计划
async def plan_step(state: PlanExecute):
    # 异步调用计划生成器，传入用户输入，生成计划
    plan = await planner.ainvoke({"messages": [("user", state["input"])]})
    # 返回生成的计划步骤
    return {"plan": plan.steps}


# 定义一个异步函数，用于在需要时重新规划计划
async def replan_step(state: PlanExecute):
    # 异步调用重新规划器，传入当前状态
    output = await replanner.ainvoke(state)
    # 判断重新规划器的输出是否为用户响应类型
    if isinstance(output.action, Response):
        # 如果是，返回用户响应内容
        return {"response": output.action.response}
    else:
        # 否则，返回重新规划后的计划步骤
        return {"plan": output.action.steps}


# 定义一个函数，用于判断是否应该结束整个流程
def should_end(state: PlanExecute) -> Literal["agent", "__end__"]:
    # 检查状态中是否存在响应且响应不为空
    if "response" in state and state["response"]:
        # 如果是，返回结束标志
        return "__end__"
    else:
        # 否则，返回继续执行标志
        return "agent"


# 创建一个状态图，初始化PlanExecute
workflow = StateGraph(PlanExecute)

# 添加计划节点
workflow.add_node("planner", plan_step)

# 添加执行步骤节点
workflow.add_node("agent", execute_step)

# 添加重新计划节点
workflow.add_node("replan", replan_step)

# 设置从开始到计划节点的边
workflow.add_edge(START, "planner")

# 设置从计划到代理节点的边
workflow.add_edge("planner", "agent")

# 设置从代理到重新计划节点的边
workflow.add_edge("agent", "replan")

# 添加条件边，用于判断下一步操作
workflow.add_conditional_edges(
    "replan",
    # 传入判断函数，确定下一个节点
    should_end,
)


# 编译工作流，将工作流配置转换为可执行的应用程序对象
app = workflow.compile()
# 获取应用程序的图结构，并将其以Mermaid格式绘制为PNG图片
graph_png = app.get_graph().draw_mermaid_png()
# 以二进制写入模式打开一个名为plan_execute.png的文件
with open("plan_execute.png", "wb") as file:
    # 将绘制好的 PNG 图片数据写入文件
    file.write(graph_png)

然后我们编译工作流，获取应用程序的图结构，并将其以Mermaid格式绘制为PNG图片，最后将绘制好的PNG图片数据写入文件。

编译工具流并写入图片演示代码

python"># 编译工作流，将工作流配置转换为可执行的应用程序对象
app = workflow.compile()
# 获取应用程序的图结构，并将其以Mermaid格式绘制为PNG图片
graph_png = app.get_graph().draw_mermaid_png()
# 以二进制写入模式打开一个名为plan_execute.png的文件
with open("plan_execute.png", "wb") as file:
    # 将绘制好的PNG图片数据写入文件
    file.write(graph_png)

最后我们再提出一个问题，执行该应用程序并处理结果，打印应用程序的流式输出。

执行处理演示代码

python"># 配置执行过程中的递归限制为50次
config = {"recursion_limit": 50}

# 定义输入数据，包含一个问题。
inputs = {"input": "2022年NBA冠军队伍中的MVP是谁?请用中文答复"}

# 定义一个异步主函数，用于执行应用程序并处理结果
async def main():
    # 异步迭代应用程序的流式输出，传入输入数据和配置信息
    async for event in app.astream(inputs, config=config):
        # 遍历事件中的每个键值对
        for event_key, event_value in event.items():
            # 如果键不是结束标志
            if event_key != "__end__":
                # 打印事件的值
                print(event_value)

完整代码

python"># 从langchain_community工具模块中导入Tavily搜索结果工具类
from langchain_community.tools.tavily_search import TavilySearchResults
# 从langchain库中导入hub模块，用于从LangChain Hub拉取资源
from langchain import hub
# 从langchain_openai库中导入ChatOpenAI类，用于调用OpenAI的聊天模型
from langchain_openai import ChatOpenAI
# 从langgraph.prebuilt模块导入create_react_agent函数，用于创建ReAct风格的智能体
from langgraph.prebuilt import create_react_agent
# 导入operator模块，这里后续用于类型注解中的特定操作
import operator
# 从typing模块导入Annotated、List、Tuple、TypedDict用于类型注解
from typing import Annotated, List, Tuple, TypedDict, Literal
# 从 pydantic模块导入BaseModel和Field类，用于定义数据模型
from pydantic import BaseModel, Field
# 从langchain_core.prompts模块导入ChatPromptTemplate类，用于创建聊天提示模板
from langchain_core.prompts import ChatPromptTemplate
# 从typing模块导入Union类型，用于定义联合类型注解
# 导入asyncio库，用于编写异步I/O、协程等异步代码，支持高效的并发操作
import asyncio
# 从typing模块导入Union类型，用于定义变量可以是多种类型中的任意一种
from typing import Union
# 从langgraph.graph模块导入StateGraph类和START常量，前者可能用于构建状态图，后者或表示起始状态
from langgraph.graph import StateGraph, START


# 定义一个名为PlanExecute的TypedDict类，用于表示特定的数据结构
class PlanExecute(TypedDict):
    # 定义一个名为input的字段，类型为字符串，代表输入信息
    input: str
    # 定义一个名为plan的字段，类型为字符串列表，代表计划步骤
    plan: List[str]
    # 定义一个名为past_steps的字段，类型为经过Annotated注解的元组列表，使用operator.add作为额外元数据
    past_steps: Annotated[List[Tuple], operator.add]
    # 定义一个名为response的字段，类型为字符串，代表响应信息
    response: str


# 定义一个名为Plan的数据模型类，继承自BaseModel
class Plan(BaseModel):
    """未来要执行的计划"""
    # 定义steps字段，类型为字符串列表，使用Field进行描述说明
    steps: List[str] = Field(
        # 说明steps是需要按顺序排列的不同执行步骤
        description="需要执行的不同步骤，应该按顺序排列"
    )


# 定义一个名为Response的数据模型类，继承自BaseModel
class Response(BaseModel):
    """用户响应"""
    # 定义response字段，类型为字符串，代表用户的响应内容
    response: str


# 定义一个名为Act的数据模型类，继承自BaseModel
class Act(BaseModel):
    """要执行的行为"""
    # 定义action字段，类型为Response或Plan的联合类型，使用Field进行描述说明
    action: Union[Response, Plan] = Field(
        # 说明action字段的用途，根据情况选择使用Response或Plan
        description="要执行的行为。如果要回应用户，使用Response。如果需要进一步使用工具获取答案，使用Plan。"
    )


# 创建一个包含Tavily搜索结果工具实例的列表，设置最大结果数为1
tools = [TavilySearchResults(max_results=1)]
# 从LangChain Hub拉取名为"wfh/react-agent-executor"的提示模板
prompt = hub.pull("wfh/react-agent-executor")
# 创建一个ChatOpenAI实例，指定使用模型
chat_model = ChatOpenAI(model="gpt-3.5-turbo")
# 调用create_react_agent函数，结合大语言模型、工具列表和提示模板修改器，创建一个ReAct智能体执行器
agent_executor = create_react_agent(chat_model, tools, messages_modifier=prompt)

# 创建一个计划生成的提示模板
planner_prompt = ChatPromptTemplate.from_messages(
    [
        (
            # 定义系统消息，为生成计划提供规则和要求
            "system",
            """对于给定的目标，提出一个简单的逐步计划。这个计划应该包含独立的任务，如果正确执行将得出正确的答案。不要添加任何多余的步骤。最后一步的结果应该是最终答案。确保每一步都有所有必要的信息 - 不要跳过步骤。""",
        ),
        # 定义占位符消息，用于后续填充实际的用户输入
        ("placeholder", "{messages}"),
    ]
)

# 使用指定的提示模板创建一个计划生成器，使用OpenAI的GPT模型
planner = planner_prompt | ChatOpenAI(
    # 指定使用模型
    model="gpt-3.5-turbo",
    # 设置temperature=0不考虑随机性
    temperature=0
).with_structured_output(Plan)

# 调用计划生成器，传入用户的问题信息
response = planner.invoke(
    {
        "messages": [
            # 用户提出的问题
            ("user", "2022年NBA冠军是哪支队伍")
        ]
    }
)

# 创建一个重新计划的提示模板
replanner_prompt = ChatPromptTemplate.from_template(
    """
        对于给定的目标，提出一个简单的逐步计划。这个计划应该包含独立的任务，如果正确执行将得出正确的答案。不要添加任何多余的步骤。最后一步的结果应该是最终答案。确保每一步都有所有必要的信息 - 不要跳过步骤。

        你的目标是：
        {input}

        你的原计划是：
        {plan}

        你目前已完成的步骤是：
        {past_steps}

        相应地更新你的计划。如果不需要更多步骤并且可以返回给用户，那么就这样回应。如果需要，填写计划。只添加仍然需要完成的步骤。不要返回已完成的步骤作为计划的一部分。
    """
)

# 使用重新计划的提示模板和OpenAI模型创建一个重新计划器
replanner = replanner_prompt | ChatOpenAI(
    # 指定使用模型
    model="gpt-3.5-turbo",
    # 设置temperature=0不考虑随机性
    temperature=0
).with_structured_output(Act)


# 定义一个异步函数，用于执行计划中的单个步骤
async def execute_step(state: PlanExecute):
    # 从状态中获取当前计划
    plan = state["plan"]
    # 将计划中的步骤转换为格式化的字符串，方便后续展示
    plan_str = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(plan))
    # 取出计划中的第一个任务
    task = plan[0]
    # 格式化任务描述，包含完整计划和当前要执行的步骤
    task_formatted = f"""对于以下计划：
                        {plan_str}\n\n你的任务是执行第{1}步，{task}。
                       """
    # 异步调用智能体执行器，传入格式化后的任务描述
    agent_response = await agent_executor.ainvoke(
        {"messages": [("user", task_formatted)]}
    )
    # 返回更新后的状态，包含已执行的步骤和对应的响应
    return {
        "past_steps": state["past_steps"] + [(task, agent_response["messages"][-1].content)],
    }


# 定义一个异步函数，用于生成初始计划
async def plan_step(state: PlanExecute):
    # 异步调用计划生成器，传入用户输入，生成计划
    plan = await planner.ainvoke({"messages": [("user", state["input"])]})
    # 返回生成的计划步骤
    return {"plan": plan.steps}


# 定义一个异步函数，用于在需要时重新规划计划
async def replan_step(state: PlanExecute):
    # 异步调用重新规划器，传入当前状态
    output = await replanner.ainvoke(state)
    # 判断重新规划器的输出是否为用户响应类型
    if isinstance(output.action, Response):
        # 如果是，返回用户响应内容
        return {"response": output.action.response}
    else:
        # 否则，返回重新规划后的计划步骤
        return {"plan": output.action.steps}


# 定义一个函数，用于判断是否应该结束整个流程
def should_end(state: PlanExecute) -> Literal["agent", "__end__"]:
    # 检查状态中是否存在响应且响应不为空
    if "response" in state and state["response"]:
        # 如果是，返回结束标志
        return "__end__"
    else:
        # 否则，返回继续执行标志
        return "agent"


# 创建一个状态图，初始化PlanExecute
workflow = StateGraph(PlanExecute)

# 添加计划节点
workflow.add_node("planner", plan_step)

# 添加执行步骤节点
workflow.add_node("agent", execute_step)

# 添加重新计划节点
workflow.add_node("replan", replan_step)

# 设置从开始到计划节点的边
workflow.add_edge(START, "planner")

# 设置从计划到代理节点的边
workflow.add_edge("planner", "agent")

# 设置从代理到重新计划节点的边
workflow.add_edge("agent", "replan")

# 添加条件边，用于判断下一步操作
workflow.add_conditional_edges(
    "replan",
    # 传入判断函数，确定下一个节点
    should_end,
)


# 编译工作流，将工作流配置转换为可执行的应用程序对象
app = workflow.compile()
# 获取应用程序的图结构，并将其以Mermaid格式绘制为PNG图片
graph_png = app.get_graph().draw_mermaid_png()
# 以二进制写入模式打开一个名为plan_execute.png的文件
with open("plan_execute.png", "wb") as file:
    # 将绘制好的PNG图片数据写入文件
    file.write(graph_png)

# 配置执行过程中的递归限制为50次
config = {"recursion_limit": 50}

# 定义输入数据，包含一个问题。
inputs = {"input": "2022年NBA冠军队伍中的MVP是谁?请用中文答复"}

# 定义一个异步主函数，用于执行应用程序并处理结果
async def main():
    # 异步迭代应用程序的流式输出，传入输入数据和配置信息
    async for event in app.astream(inputs, config=config):
        # 遍历事件中的每个键值对
        for event_key, event_value in event.items():
            # 如果键不是结束标志
            if event_key != "__end__":
                # 打印事件的值
                print(event_value)

# 运行异步主函数
asyncio.run(main())

运行结果

python">{'plan': ['查找2022年NBA冠军队伍', '确定该队伍的MVP球员']}
{'past_steps': [('查找2022年NBA冠军队伍', '2022年NBA冠军队伍是金州勇士队，他们在总决赛中以总比分4比2战胜波士顿凯尔特人队，这也是他们的第7个总冠军。')]}
{'plan': ['确定金州勇士队的MVP球员']}
{'past_steps': [('查找2022年NBA冠军队伍', '2022年NBA冠军队伍是金州勇士队，他们在总决赛中以总比分4比2战胜波士顿凯尔特人队，这也是他们的第7个总冠军。'), ('确定金州勇士队的MVP球员', '根据最新的信息，金州勇士队的MVP球员是斯蒂芬·库里（Stephen Curry）。他被评选为2023-24赛季的杰瑞·韦斯特奖（Jerry West Trophy），成为NBA的关键时刻最佳球员。')]}
{'response': '金州勇士队的2022年NBA冠军MVP是斯蒂芬·库里（Stephen Curry）'}

进程已结束，退出代码为 0

我们从运行结果看到，完整代码主要是基于LangChain和LangGraph构建的智能问答系统，该系统通过定义数据模型、创建工具和模型实例、构建状态图和工作流，最终实现了一个能够针对用户问题生成计划、执行计划步骤、重新规划计划并给出答案的智能问答系统。

博主经过一段时间的研究和实践，发现LangGraph在快速构建Agent工作流应用方面表现出色。它在快速构建Agent工作流应用上优势突出：

• 简化建模：借状态图等直观定义工作流，清晰呈现状态转换，降低编程复杂度。
• 模块组件化：工作流可拆为可复用模块，按需组合，提升开发效率，便于维护扩展。
• 模型集成佳：与语言模型紧密集成，简化交互，利用其 NLP 能力实现智能交互。
• 异步处理强：支持异步操作，处理耗时任务不阻塞，提升响应和并发能力。
• 可视化助力：部分实现有可视化工具，便于查看编辑工作流，利沟通协作。
• 迭代实验快：架构简洁设计灵活，利于快速迭代优化，加速应用开发。
• 工具生态全：提供多样工具库，社区资源丰富，为开发提供有力支持。

结束 

好了，以上就是本次分享的全部内容了。不知道大家是否理解与掌握了在本次分享中所讲解的使用 LangGraph实现有状态、多参与者的应用程序构建的相关要点。

如果大家对LangGraph感兴趣，可以去其官网Home看看，官网上有详细的文档介绍、丰富的使用教程以及示例代码等资源，有助于大家对它有更深刻的认识和更全面的学习。

时光飞逝，一转眼博主已经分享了11篇关于LangChain的文章了。在这段分享知识的时光里，非常开心能感受到大家满满的支持与鼓励。每一个点赞、每一次收藏，以及大家选择关注博主，都让我备受鼓舞，真心感谢大家的认可与陪伴！

目前，关于LangChain的分享先暂时告一段落。不过要是之后在研究中有新的发现和感悟，博主肯定还会继续更新相关内容，和大家一起交流探讨。

接下来，博主打算开启新的知识旅程，将重点聚焦于自然语言处理领域。希望大家能持续关注，我们一同在知识的海洋中成长进步，碰撞出更多思维的火花！

那么本次分享就到这里了。最后，博主还是那句话：请大家多去大胆的尝试和使用，成功总是在不断的失败中试验出来的，敢于尝试就已经成功了一半。如果大家对博主分享的内容感兴趣或有帮助，请点赞和关注。大家的点赞和关注是博主持续分享的动力🤭，博主也希望让更多的人学习到新的知识。