整体课程设计介绍
第一次直播：
-
开班介绍+环境适配+自主学习任务布置
任务：预习系列（参考2期）
- day20-大模型基本概述+API调用
- day21-大模型基本概述+API调用
- day22-langchain框架基础
- day23-langchain框架进阶
第二次直播：
-
api接口调用和Functioncalling回顾
-
langchain快速入门
第三次直播：
- swarm基础
第四次直播：
- swarm进阶+agent sdk
第五次直播：
- agent sdk + streamlit
第六次直播：
- Excel智能体项目
LangGraph概述与应用
一、LangGraph入门介绍
1. LangGraph：超越LangChain的全新一代Agent开发框架
说到现阶段目前最流行、最通用的Agent开发框架，毫无疑问，肯定是LangChain。LangChain作为2022年就已经开源的元老级开发框架，历经数年的发展，其功能和生态都已非常完善，并且拥有数量众多的开发者。在统计的今年大模型招聘岗位中，有90%以上的Agent开发岗位要求掌握LangChain，可以说LangChain就是目前最通用、最流行的Agent开发框架没有之一。
LangGraph和LangChain同宗同源，底层架构完全相同、接口完全相通。从开发者角度来说，LangGraph也是使用LangChain底层API来接入各类大模型、LangGraph也完全兼容LangChain内置的一系列工具。换而言之，LangGraph的核心功能都是依托LangChain来完成。但是和LangChain的链式工作流思想完全不同的是，LangGraph的基础思想是构建图结构的工作流，并引入“状态”这一核心概念来描绘任务执行情况，从而拓展了LangChain LCEL链式语法的功能灵活程度。
很明显，图状工作流肯定比链式工作流功能要更加灵活，功能可拓展性也更强。
用于创建图结构的工作流，这也是LangGraph中Graph一词的由来。
不过需要注意的是，LangGraph是基于LangChain进行的构建，无论图结构多复杂，单独每个任务执行链路仍然是线性的，其背后仍然是靠着LangChain的Chain来实现的。因此我们可以这么来描述LangChain和LangGraph之间的关系，LangGraph是LangChain工作流的高级编排工具，其中“高级”之处就是LangGraph能按照图结构来编排工作流。
2.环境安装&核心概念
注意：需要将langgraph下载安装到有langchain环境的虚拟环境中！
python版本建议：3.12
langchain版本：0.3.20
使用LangGraph需要pip install langgraph==0.4.5。
-
核心概念：
- 状态（State）：是LangGraph应用的基础，包含了应用运行时的所有信息，如消息列表、当前输入、工具输出等。
- 节点（Node）：通常是Python函数，代表不同的操作或步骤，如调用LLM、处理用户输入等，用于处理状态并返回更新后的状态。
- 边（Edge）：定义了节点之间的连接关系和路由逻辑，包括标准边和条件边，标准边定义固定的执行路径，条件边可根据状态决定下一步走向。
-
主要特性：
- 结构化工作流：能创建具有分支、循环和条件逻辑的复杂工作流，相比单一的链式调用更具灵活性。
- 状态管理：提供强大的状态管理机制，自动保存和管理状态，支持暂停和恢复执行，便于处理长时间运行的对话。
- 与LangChain无缝集成：可复用现有的LangChain组件，还有丰富的工具和模型支持。
- 实现复杂逻辑：传统的智能体开发方式在处理复杂任务时存在局限，如缺乏对外部环境的感知能力、对话历史记忆有限等。LangGraph允许创建具有循环、条件分支等复杂逻辑的工作流，能更好地应对各种复杂场景和需求，例如根据不同的输入和状态动态调整执行路径，实现多步骤的推理和决策。
3. LangGraph开发必备工具套件（了解，后期会应用）
LangGraph Cli
在对LangGraph有了一定的基础了解之后，对于开发者来说，还需要进一步了解和掌握LangGraph必备的开发者套件：LangGraph服务部署工具LangGraph Cli。
LangGraph CLI 是用于本地启动、调试、测试和托管 LangGraph 智能体图的开发者命令行工具。
而对于LangGraph构建的智能体，除了能够本地部署外，官方也提供了云托管服务，借助LangGraph Platform，开发者可以将构建的智能体 Graph部署到云端，并允许公开访问，同时支持支持长时间运行、文件上传、外部 API 调用等功能。
LangGraph Agent前端可视化工具：Agent Chat UI
Agent Chat UI 是 LangGraph/LangChain 官方提供的多智能体前端对话面板，用于与后端 Agent（Graph 或 Chain）进行实时互动，支持上传文件、多工具协同、结构化输出、多轮对话、调试标注等功能。
Agent Chat UI官网地址：https://langchain-ai.github.io/langgraph/agents/ui/
4. LangGraph VS 其他开发框架
而关于LangGraph框架与其他开发框架的对比，其实从上面对于LangGrpah的技术介绍不难看出，依托LangChain的强大生态，LangGraph可以说是目前功能最完善、性能最强的开发框架没有之一。
那么LangGraph VS LangChain？
首先，如果要对比LangGraph和LangChain，其实二者定位有本质的区别。从目前LangChain工具功能划分来看，LangChain未来将继续专注于更加底层的API开发与维护，以及链式基础功能的维护，而作为LangChain链式功能更高级编排工具，LangGraph未来将承担更多Agent开发的重任。
二、langgraph实操
1. langgraph实现Agent基础操作
from typing import Literal
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAIfrom langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END,StateGraph,MessagesState
from langgraph.prebuilt import ToolNode#定义工具函数，用于Agent调用外部工具
@tool
def search(query:str):"""模拟一个气象查询搜索工具"""if "北京" in query.lower() or "Beijing" in query.lower():return "阴天有雾，气温25度"return "天气晴朗温度较高39度"#将工具函数存放在工具列表中
tools = [search]
创建工具集节点：ToolNode是LangGraph中的一个预构建节点，用于封装一组工具函数。这些工具函数可以通过模型调用来执行特定的任务。
tool_node = ToolNode(tools)
定义模型对象
#定义模型对象
API_KEY = "sxxxx25b3"
model = ChatOpenAI(model_name="deepseek-chat",api_key=API_KEY,base_url="https://api.deepseek.com")
#将工具列表绑定到模型对象上
model = model.bind_tools(tools)
定义路由函数/状态转换函数：should_continue函数用于决定当前状态之后应该转移到哪个节点。它接收一个MessagesState对象作为输入，并返回一个字符串，表示下一个节点的名称。
消息状态：MessagesState是LangGraph中的一个状态类，用于存储对话过程中的消息列表。每个状态对象都包含一个messages字段，该字段是一个消息对象的列表。
from typing import Literal 是 Python 3.8 及以上版本中引入的一种类型注解工具，用于表示某个变量或函数参数只能是特定的几个值之一。Literal 是 typing 模块中的一个特殊类型，它允许你精确地指定一个或多个字面量作为类型约束。
def should_continue(state:MessagesState)->Literal["tools",END]:messages = state['messages']#获取用户提问消息last_message = messages[-1] #如果llm调用工具，则转到tools节点if last_message.tool_calls:return "tools"return END
定义模型调用函数
def call_model(state:MessagesState):#获取消息列表messages = state['messages']#调用模型返回结果response = model.invoke(messages)return {"messages":[response]}
定义一个新的状态图,使用MessagesState作为状态类型
workflow = StateGraph(MessagesState)
在状态图上添加节点
workflow.add_node("agent",call_model)
workflow.add_node("tools",tool_node)
设置入口节点为agent(入口节点指向agent节点)，这意味着agent是第一个被调用的节点
workflow.set_entry_point("agent")
添加条件边：agent节点根据should_continue进行边的连接（虚线边）
workflow.add_conditional_edges('agent',should_continue)
定义普通边:tools工具节点连接agent节点的边（实线边）
workflow.add_edge("tools","agent")
初始化内存以在图运行之间持久化状态：MemorySaver是LangGraph中的一个检查点保存器，用于在内存中保存状态图的中间状态。这对于调试和监控非常有用，因为它允许你在运行时查看和恢复状态。
checkpointer = MemorySaver()
编译图：将其编译成一个langchain可运行的一个对象,在编译时传递内存
app = workflow.compile(checkpointer=checkpointer)
执行图
final_state = app.invoke({"messages":[HumanMessage(content="北京天气如何？")]},config={"configurable":{"thread_id":42}}
)
result = final_state['messages'][-1].content
result
配置选项（config）实现上下文共享：如果两个任务在同一个线程上执行，它们可以共享同一个上下文（例如全局变量、线程本地存储等）。这对于需要维护状态或会话信息的应用非常重要。
final_state = app.invoke({"messages":[HumanMessage(content="我刚才问的是哪个城市？")]},config={"configurable":{"thread_id":42}}
)
result = final_state['messages'][-1].content
result
保存图文件
graph_png = app.get_graph().draw_mermaid_png()
with open('graph.png','wb') as fp:fp.write(graph_png)
2. langgraph实现Multi-Agent Systems
SerpAPI是一个搜索引擎结果页面API，它允许开发者和研究人员通过编程方式获取Google、Bing、Yahoo和其他搜索引擎的搜素结果。使用SerpAPI，用户可以避免直接与搜索引擎进行交互（无需xx上网），从而避免了可能遇到的各种问题，例如：用户代理、请求限制等问题。
环境安装：
- pip install google-search-results
- pip install langchain-community
官网进行API KEY的申请：https://serpapi.com/
from langchain_core.messages import (
BaseMessage,
HumanMessage,
ToolMessage,
)
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from typing import Literal
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END,StateGraph,MessagesState
from langgraph.prebuilt import ToolNode# 导⼊聊天提示模板和消息占位符
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# 导⼊状态图相关的常量和类
from langgraph.graph import END, StateGraph, STARTAPI_KEY = "sk-xxxx5b3"
llm = ChatOpenAI(model_name="deepseek-chat",api_key=API_KEY,base_url="https://api.deepseek.com")# 定义⼀个函数，⽤于创建代理，system_message表示模型背景信息
def create_agent(llm, tools, system_message: str):"""创建⼀个代理。"""# 创建⼀个聊天提示模板prompt = ChatPromptTemplate.from_messages([("system","你是⼀个有帮助的AI助⼿，与其他助⼿合作。"" 使⽤提供的⼯具来推进问题的回答。"" 如果你不能完全回答，没关系，另⼀个拥有不同⼯具的助⼿"" 会接着你的位置继续帮助。执⾏你能做的以取得进展。"" 如果你或其他助⼿有最终答案或交付物，"" 在你的回答前加上FINAL ANSWER，以便团队知道停⽌。"" 你可以使⽤以下⼯具: {tool_names}。\n{system_message}",),# 消息占位符MessagesPlaceholder(variable_name="messages"),])# 传递系统消息参数prompt = prompt.partial(system_message=system_message)# 传递⼯具名称参数prompt = prompt.partial(tool_names=", ".join([tool.name for tool in tools]))# 绑定⼯具并返回提示模板return prompt | llm.bind_tools(tools)#定义工具函数
@tool
def get_search_result(question):"""互联网搜索函数:param question: 必要参数，字符串类型，用于表示在互联网上进行搜素的关键词或者搜索内容的简短描述，\:return：SerpAPI API根据参数question进行互联网搜索后的结果，其中包含了全部重要的搜索结果内容。"""from langchain_community.utilities import SerpAPIWrapperserpapi_api_key = "60f286e601f44a26600e42c65e7a9b3ceb06a3f0dc8e0fe7ce56ec93d6274ccd"search = SerpAPIWrapper(serpapi_api_key=serpapi_api_key)result = search.run(question)return result#定义工具函数，用于Agent调用外部工具
@tool
def send_email(query:str):"""邮件发送工具，可以接受query内容，然后进行邮件发送"""return "邮件已成功发送。"#定义工具节点
# 导⼊预构建的⼯具节点
from langgraph.prebuilt import ToolNode
# 定义⼯具列表
tools = [get_search_result, send_email]
# 创建⼯具节点
tool_node = ToolNode(tools)#定义状态：我们⾸先定义图的状态。这只是⼀个消息列表，以及⼀个⽤于跟踪最新发送者的键
# 导⼊操作符和类型注解
import operator
from typing import Annotated, Sequence, TypedDict
# 导⼊OpenAI聊天模型
from langchain_openai import ChatOpenAI
# 定义⼀个对象，⽤于在图的每个节点之间传递
# 我们将为每个代理和⼯具创建不同的节点class AgentState(TypedDict):messages: Annotated[Sequence[BaseMessage], operator.add]sender: str#定义代理节点
import functools
from langchain_core.messages import AIMessage# 模型调用函数封装，后面用于节点创建环节
def agent_node(state, agent, name):#name:agent代理的名字# 调⽤代理result = agent.invoke(state)#将 result 转换为 AIMessage 类型，并进行进一步处理#使用模型的 model_dump 方法将 result 转换为字典格式，同时排除 "type" 和 "name" 字段。这通常用于序列化对象以便传输或存储。result = AIMessage(**result.model_dump(exclude={"type", "name"}), name=name)return {"messages": [result],# 由于我们有⼀个严格的⼯作流程，我们可以跟踪发送者，以便知道下⼀个传递给谁。"sender": name,}#创建搜索Agent代理对象和节点对象
research_agent = create_agent(llm,[get_search_result],system_message="你应该提供准确的数据供MailOpt使⽤。",
)
# 创建Agent节点对象：使用agent和name的值填充到agent_node函数中对应的两个参数
research_node = functools.partial(agent_node, agent=research_agent, name="Researcher")#创建发邮件Agent代理对象和节点对象
mail_agent = create_agent(llm,[send_email],system_message="你用于进行邮件发送业务实现",
)
# 创建Agent节点对象
mail_node = functools.partial(agent_node, agent=mail_agent, name="MailOpt")#定义路由函数，决定是否继续执行
from typing import Literal
# 定义路由器函数,continue 表示代理应该继续处理消息队列中的下一条消息。
def router(state) -> Literal["call_tool", "__end__", "continue"]:# 这是路由器messages = state["messages"]last_message = messages[-1]if last_message.tool_calls:# 上⼀个代理正在调⽤⼯具return "call_tool"if "FINAL ANSWER" in last_message.content:# 任何代理决定⼯作完成return "__end__"return "continue"#图、节点和边的创建
# 创建状态图实例
workflow = StateGraph(AgentState)
# 添加搜索节点
workflow.add_node("Researcher", research_node)
# 添加邮件节点
workflow.add_node("MailOpt", mail_node)
# 添加⼯具调⽤节点
workflow.add_node("call_tool", tool_node)# 添加条件边
workflow.add_conditional_edges("Researcher",router,{"continue": "MailOpt", "call_tool": "call_tool", "__end__": END},
)
workflow.add_conditional_edges("MailOpt",router,{"continue": "Researcher", "call_tool": "call_tool", "__end__": END},
)
# 添加条件边
workflow.add_conditional_edges("call_tool",#如果 x["sender"] 的值是 "Researcher"，那么边会连接到 "Researcher" 节点。#如果 x["sender"] 的值是 "MailOpt"，那么边会连接到 "MailOpt" 节点。lambda x: x["sender"],{"Researcher": "Researcher","MailOpt": "MailOpt",},
)
# 添加起始边
workflow.add_edge(START, "Researcher")
# 编译⼯作流图
graph = workflow.compile()# 将⽣成的图⽚保存到⽂件
graph_png = graph.get_graph().draw_mermaid_png()
with open("collaboration.png", "wb") as f:f.write(graph_png)#调用
events = graph.invoke({"messages": [HumanMessage(content="获取过去5年AI软件市场规模，归纳成100字"" 然后进行邮件发送。"" ⼀旦发送完邮件表示你完成了任务。")],}
)#获取最终结果
result = events['messages'][-1].content
result#查看中间结果
for message in events['messages']:print(message.content)print("-----------------------------------")
