内容纲要
事件流应用概述
在实际应用中，流式输出尤其适用于需要快速反馈的业务场景，如聊天机器人，因为大语言模型可能需要几秒钟才能生成对查询的完整响应，这远远慢于应用程序对最终用户的响应速度约为 200-300 毫秒的阈值，如果是涉及多个大模型调用的复杂应用程序，这种延时会变得更加明显。让应用程序感觉响应更快的关键策略是显示中间进度；即，通过 token 流式传输大模型 Token 的输出，以此来显著提升用户体验。而在开发阶段，利用流式输出功能可以准确追踪到事件的具体执行阶段，并捕获相关数据，从而接入不同逻辑的数据处理和决策流程。是我们在应用开发中必须理解和掌握的技术点。
流式输出功能在 LangGraph 框架中的实现方式比较简单，因为 LangGraph 底层是基于 LangChain 构建的，所有就直接把 LangChain 中的回调系统拿过来使用了。在 LangChain 中的流式输出是：以块的形式传输最终输出，即一旦监测到有可用的块，就直接生成它。最常见和最关键的流数据是大模型本身生成的输出。
I langchain的流输出
大模型通常需要时间才能生成完整的响应，通过实时流式传输输出，用户可以在生成时看到部分结果，这可以提供即时反馈并有助于减少用户的等待时间。如下所示：
# 1、配置模型信息
import getpass
import os
from dotenv import load_dotenv
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key, base_url=base_url, temperature=0,)
# import asyncio
# chunks = []
# async def async_calling():
# async for chunk in llm.astream("你好，请你详细的介绍一下你自己。限制在20字以内"):
# chunks.append(chunk)
# print(chunk.content, end="", flush=True)
# asyncio.run(async_calling())
chunks = []
for chunk in llm.stream("你好，请你详细的介绍一下你自己。"):
chunks.append(chunk)
print(chunk.content, end="|", flush=True)
print(f"总共接收到的 chunk 数量: {len(chunks)}")
# 每个块都是一个 AIMessageChunk 对象，用来代表 AIMessage 对象的一部分，消息块在设计上是可加的，比如：
print(chunks[23])
print(chunks[23]+chunks[24]+chunks[25]+chunks[26]+chunks[27])
# 而进一步的，除了流式传输大模型的输出之外，通过更复杂的工作流程或管道流式传输进度也很有用，比如 AI Agent 中的中间处理过程，这就涉及到工作流概念。
II LangGraph 输出流
0 流输出，一般有两种形式：同步、异步。异步可以通过事件去提取指定字段的内容，进行流式输出。同步则可以通过stream_mode（流模式），从而做流输出
LangGraph 框架中的工作流由各个步骤的节点和边组成。这里的流式传输涉及在各个节点请求更新时跟踪状态的变化。这样可以更精细地监控工作流中当前处于活动状态的节点，并在工作流经过不同阶段时提供有关工作流状态的实时更新。其实现方式也是和 LangChain 一样通过 .stream 和 .astream 方法执行流式输出，只不过适配到了图结构中。调用 .stream 和 .astream 方法时可以指定几种不同的模式，即：
- "values"：在图中的每个步骤之后流式传输状态的完整值。
- "updates"：在图中的每个步骤之后将更新流式传输到状态。如果在同一步骤中进行多个更新（例如运行多个节点），则这些更新将单独流式传输。
- "debug"：在整个图的执行过程中流式传输尽可能多的信息，主要用于调试程序。
- "messages"：记录每个 messages 中的增量 token。
- "custom"：自定义流，通过 LangGraph 的 StreamWriter 方法
两种核心模式对比
首先来看 Stream 方法，该方法返回一个迭代器，在生成输出块时同步生成它们。我们可以使用 for 循环来实时处理每个块。生成的块的类型取决于正在流式传输的组件。例如，当从大模型流式传输时，每个组件将是一个 AIMessageChunk，但是，对于其他组件，块可能会有所不同。其 LangGraph 框架中实现的源码如下：
LangGraph Graph stream 文档：https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.graph.CompiledGraph.stream
stream 方法完整签名：
def stream(
self,
input: Union[dict[str, Any], Any], # 图的输入数据，写入状态
config: Optional[RunnableConfig] = None,
*,
stream_mode: Optional[Union[StreamMode, list[StreamMode]]] = None,
output_keys: Optional[Union[str, Sequence[str]]] = None, # 流输出字段，默认全部非上下文通道
interrupt_before: Optional[Union[All, Sequence[str]]] = None, # 在指定节点执行前中断，默认所有节点
interrupt_after: Optional[Union[All, Sequence[str]]] = None, # 在指定节点执行后中断，默认所有节点
debug: Optional[bool] = None, # 是否打印调试日志，默认False
subgraphs: bool = False, # 是否同步流式输出子图内部事件
) -> Iterator[Union[dict[str, Any], Any]]input
工作流初始输入，字典类型，会合并写入图的状态；多轮续跑时传 None。（用户输入，prompt）
config
运行上下文配置，核心携带 thread_id 区分对话会话。
stream_mode
流式输出模式，支持单个 / 多个模式组合：values/updates/debug/messages/custom。
output_keys
限定仅流式返回状态内指定字段，用于裁剪返回数据体积。
interrupt_before / interrupt_after
断点控制参数，配置需要人工审批的节点，分别代表执行前中断、执行完成后中断，实现人机介入工作流。
debug
开启后打印完整节点调度、状态变更、工具调用日志，用于开发调试。
subgraphs
嵌套子图场景开关，开启后子图内部节点事件也会同步流出。
1 代码演示：（根据智能体开发的需求，提取到指定字段的信息，在前端给用户展示）
from langchain_core.tools import tool
from typing import Union, Optional
from pydantic import BaseModel, Field
import requests
from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.orm import sessionmaker, declarative_base
import json
# 1、读取配置信息
from dotenv import load_dotenv
load_dotenv()
import os
passwrod = os.environ["MYSQL_PASSWORD"]
acc = os.environ["MYSQL_account"]
DATABASE_URI = f'mysql+pymysql://{acc}:{passwrod}@localhost/langgraph?charset=utf8mb4'
engine = create_engine(DATABASE_URI)
weather_key = os.environ["OPEN_WEATHER_KEY"]
GOOGLE_SEARCH_KEY = os.environ["GOOGLE_SEARCH_KEY"]
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key,base_url=base_url,temperature=0,)
# 数据库连接 URI，这里要替换成自己的Mysql 连接信息，以下是各个字段的对应解释：
# gpt: MySQL 数据库的用户名。
# gpt: MySQL 数据库的密码。
# localhost: MySQL 服务器的 IP 地址。
# langgraph: 要连接的数据库的名称。
# charset=utf8mb4: 设置数据库的字符集为 utf8mb4，支持更广泛的 Unicode 字符
# 创建会话
Session = sessionmaker(bind=engine)
# 创建基类
Base = declarative_base()
# 定义 WeatherInfo 模型
class Weather(Base):
__tablename__ = 'weather'
city_id = Column(Integer, primary_key=True) # 城市ID
city_name = Column(String(50)) # 城市名称
main_weather = Column(String(50)) # 主要天气状况
description = Column(String(100)) # 描述
temperature = Column(Float) # 温度
feels_like = Column(Float) # 体感温度
temp_min = Column(Float) # 最低温度
temp_max = Column(Float) # 最高温度
# 第一个工具
class WeatherLoc(BaseModel):
location: str = Field(description="The location name of the city")
class WeatherInfo(BaseModel):
"""Extracted weather information for a specific city."""
city_id: int = Field(..., description="The unique identifier for the city")
city_name: str = Field(..., description="The name of the city")
main_weather: str = Field(..., description="The main weather condition")
description: str = Field(..., description="A detailed description of the weather")
temperature: float = Field(..., description="Current temperature in Celsius")
feels_like: float = Field(..., description="Feels-like temperature in Celsius")
temp_min: float = Field(..., description="Minimum temperature in Celsius")
temp_max: float = Field(..., description="Maximum temperature in Celsius")
## 第一个工具：获取天气信息
@tool(args_schema = WeatherLoc)
def get_weather(location):
"""
Function to query current weather.
:param loc: Required parameter, of type string, representing the specific city name for the weather query. \
Note that for cities in China, the corresponding English city name should be used. For example, to query the weather for Beijing, \
the loc parameter should be input as 'Beijing'.
:return: The result of the OpenWeather API query for current weather, with the specific URL request address being: https://api.openweathermap.org/data/2.5/weather
The return type is a JSON-formatted object after parsing, represented as a string, containing all important weather information.
"""
# Step 1.构建请求
url = "https://api.openweathermap.org/data/2.5/weather"
# Step 2.设置查询参数
params = {
"q": location,
"appid": weather_key, # 输入API key
"units": "metric", # 使用摄氏度而不是华氏度
"lang":"zh_cn" # 输出语言为简体中文
}
# Step 3.发送GET请求
response = requests.get(url, params=params)
# Step 4.解析响应
data = response.json()
return json.dumps(data)
## 第二个工具
@tool(args_schema=WeatherInfo)
def insert_weather_to_db(city_id, city_name, main_weather, description, temperature, feels_like, temp_min, temp_max):
"""Insert weather information into the database."""
session = Session() # 确保为每次操作创建新的会话
try:
# 创建天气实例
weather = Weather(
city_id=city_id,
city_name=city_name,
main_weather=main_weather,
description=description,
temperature=temperature,
feels_like=feels_like,
temp_min=temp_min,
temp_max=temp_max
)
# 使用 merge 方法来插入或更新（如果已有记录则更新）
session.merge(weather)
# 提交事务
session.commit()
return {"messages": [f"天气数据已成功存储至Mysql数据库。"]}
except Exception as e:
session.rollback() # 出错时回滚
return {"messages": [f"数据存储失败，错误原因：{e}"]}
finally:
session.close() # 关闭会话
## 第三个工具
class QueryWeatherSchema(BaseModel):
"""Schema for querying weather information by city name."""
city_name: str = Field(..., description="The name of the city to query weather information")
@tool(args_schema=QueryWeatherSchema)
def query_weather_from_db(city_name: str):
"""Query weather information from the database by city name."""
session = Session()
try:
# 查询天气数据
weather_data = session.query(Weather).filter(Weather.city_name == city_name).first()
if weather_data:
return {
"city_id": weather_data.city_id,
"city_name": weather_data.city_name,
"main_weather": weather_data.main_weather,
"description": weather_data.description,
"temperature": weather_data.temperature,
"feels_like": weather_data.feels_like,
"temp_min": weather_data.temp_min,
"temp_max": weather_data.temp_max
}
else:
return {"messages": [f"未找到城市 '{city_name}' 的天气信息。"]}
except Exception as e:
return {"messages": [f"查询失败，错误原因：{e}"]}
finally:
session.close() # 关闭会话
## 第四个工具
class SearchQuery(BaseModel):
query: str = Field(description="Questions for networking queries")
@tool(args_schema = SearchQuery)
def fetch_real_time_info(query):
"""Get real-time Internet information"""
url = "https://google.serper.dev/search"
payload = json.dumps({
"q": query,
"num": 1,
})
headers = {
'X-API-KEY': GOOGLE_SEARCH_KEY,
'Content-Type': 'application/json'
}
response = requests.post(url, headers=headers, data=payload)
data = json.loads(response.text) # 将返回的JSON字符串转换为字典
if 'organic' in data:
return json.dumps(data['organic'], ensure_ascii=False) # 返回'organic'部分的JSON字符串
else:
return json.dumps({"error": "No organic results found"}, ensure_ascii=False) # 如果没有'organic'键，返回错误信息
# 测试天气工具调用示例
tool_result = get_weather.invoke('beijing')
print(f"tool_result:\n {tool_result}\n")
# 注册全部可用工具列表
tools = [fetch_real_time_info, get_weather, insert_weather_to_db, query_weather_from_db]
# 想看
# print(type(obj)) # 1. 先看类型
# print(obj) # 2. 打印看长相（有没有类名前缀、括号类型）
"""
如果想要看并提取出这种一长串，列表、字典、对象类型交加的数据格式中指定字段的内容，方法如下：
遇到不确定的结构，最快的排查二连
print(type(obj)) # 1. 先看类型
print(obj) # 2. 打印看长相（有没有类名前缀、括号类型）
取值方法：
[ ... ] 列表 list 变量[0]（整数索引）
{ 'key': value } 字典 dict 变量['key']
( ... ) 且里面是 key=value 逗号分隔 元组（对象的属性） 变量.属性
类名(参数=值, 参数=值) 对象 object 变量.属性名
"""
# print(f"{type(tools)} \n")
# print(f"{tools} \n")
# print(f"{type(tools[0])} \n")
# print(f"{tools[0]} \n")
# print(f"{tools[0].name} \n")
# print(f"{[i.name for i in tools]} \n")
# for i in tools:
# print(f"{i} \n")
# 基于预封装REACT模板创建Agent图
from langgraph.prebuilt import create_react_agent
graph = create_react_agent(llm, tools=tools)
# 演示1（values）在图表的每个步骤之后流式传输状态的完整值。
def print_stream(stream):
for sub_stream in stream:
#print(sub_stream) # 就是上面的示例中非流式直接调用的全部信息
message = sub_stream["messages"][-1]
"""
这是使用的 AI/大模型 SDK 提供的一个便捷调试工具，专门用来解决流式响应数据在控制台打印时“挤成一团、难以阅读”的问题。
在流式输出（Streaming）场景中，数据通常是一小块一小块（chunk）到达的。
如果直接使用普通的 print(sub_stream)，控制台会输出大量包含元数据、嵌套字典或对象的杂乱信息，难以阅读。
pretty_print() 的作用就是将这些复杂的嵌套结构（如字典、列表等）进行格式化排版，
通过合理的缩进和换行，将当前流式片段中的关键信息（如模型回复的文本内容）清晰、美观地展示在控制台或日志中。
"""
message.pretty_print()
# print("*"*70)
# print(message)
# input_message = {"messages": ["你好，北京现在的天气怎么样？"]}
# print_stream(graph.stream(input_message, stream_mode="values"))
"""
message.pretty_print(),打印效果如下：
================================= Ai Message ==================================
Tool Calls:
get_weather (chatcmpl-tool-b582f5e45499fb53)
Call ID: chatcmpl-tool-b582f5e45499fb53
Args:
location: Beijing
================================= Tool Message =================================
Name: get_weather
{"coord": {"lon": 116.3972, "lat": 39.9075}, "weather": [{"id": 804, "main": "Clouds", "description": "\u9634\uff0c\u591a\u4e91", "icon": "04d"}], "base": "stations", "main": {"temp": 36.95, "feels_like": 39.6, "temp_min": 36.95, "temp_max": 36.95, "pressure": 995, "humidity": 36, "sea_level": 995, "grnd_level": 990}, "visibility": 10000, "wind": {"speed": 3.06, "deg": 194, "gust": 3.48}, "clouds": {"all": 100}, "dt": 1783587906, "sys": {"country": "CN", "sunrise": 1783544020, "sunset": 1783597503}, "timezone": 28800, "id": 1816670, "name": "Beijing", "cod": 200}
print("*"*70)
print(message)，效果如下：
**********************************************************************
content='你好，北京现在的天气怎么样？' additional_kwargs={} response_metadata={} id='14aa5ea9-3399-4cf6-ab94-6720c2cf0342'
**********************************************************************
content='' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 45, 'prompt_tokens': 697, 'total_tokens': 742, 'cometails': None, 'prompt_tokens_details': None}, 'model_provider': 'openai', 'model_name': 'MiniMax-M2.5', 'system_fingerprint': None, 'id': 'chatcmpl-f002787-af63d727a6d8', 'finish_reason': 'tool_calls', 'logprobs': None} id='lc_run--019f4627-4344-7420-97d8-ae052ada5aa4-0' tool_calls=[{'name': 'get_weather',ion': 'Beijing'}, 'id': 'chatcmpl-tool-8fd3c68c77b536c7', 'type': 'tool_call'}] invalid_tool_calls=[] usage_metadata={'input_tokens': 697, 'output_tokens'ens': 742, 'input_token_details': {}, 'output_token_details': {}}
**********************************************************************
content='{"coord": {"lon": 116.3972, "lat": 39.9075}, "weather": [{"id": 804, "main": "Clouds", "description": "\\u9634\\uff0c\\u591a\\u4e91", "icon": "04tations", "main": {"temp": 36.95, "feels_like": 39.6, "temp_min": 36.95, "temp_max": 36.95, "pressure": 995, "humidity": 36, "sea_level": 995, "grnd_levellity": 10000, "wind": {"speed": 3.06, "deg": 194, "gust": 3.48}, "clouds": {"all": 100}, "dt": 1783587906, "sys": {"country": "CN", "sunrise": 1783544020,97503}, "timezone": 28800, "id": 1816670, "name": "Beijing", "cod": 200}' name='get_weather' id='d5e6cfc6-88b2-451e-bb38-44ffd46756af' tool_call_id='chatcc77b536c7'
**********************************************************************
"""
# 2、演示2（updates）：在图中的每个步骤之后将更新流式传输到状态。
def print_stream(stream):
for sub_stream in stream:
print(sub_stream) # 就是上面的示例中非流式直接调用的全部信息
# input_message = {"messages": ["你好，北京现在的天气怎么样？"]}
# print_stream(graph.stream(input_message, stream_mode="updates"))
# 3、演示3（debug）：在整个图中的执行过程中流式传输尽可能多的信息
def print_stream(stream):
for sub_stream in stream:
print(sub_stream) # 就是上面的示例中非流式直接调用的全部信息
# input_message = {"messages": ["你好，北京现在的天气怎么样？"]}
# print_stream(graph.stream(input_message, stream_mode="debug"))
# 如果在异步开发环境中，则可以使用 astream 方法来实现流式传输，是专为非阻塞工作流程而设计。可使用的模式和 stream 是一致的，只不过需要调整为异步函数的定义方法，代码如下所示：
async def attempt_g():
async for chunk in graph.astream(input={"messages": ["你好，丽江的天气怎么样？"]}, stream_mode="values"):
message = chunk["messages"][-1].pretty_print()
import asyncio
# asyncio.run(attempt_g())
print("*"*70)
# 如果只想得到最终结果，可以使用相同的方法并只跟踪收到的最后一个值，代码如下：
async def attempt2_g():
async for chunk in graph.astream(input={"messages": ["你好，北京的天气怎么样？"]}, stream_mode="values"):
final_result = chunk
final_result["messages"][-1].pretty_print()
# asyncio.run(attempt2_g())
# 可以找图的chunk里面的其他信息，做展示
inputs = {"messages": [("human", "你好，大理的天气怎么样？")]}
async def attempt3_g():
async for chunk in graph.astream(inputs, stream_mode="updates"):
for node, values in chunk.items():
print(f"接收到的更新节点: '{node}'")
print(values)
print("\n\n")
# asyncio.run(attempt3_g())
# 而如果我们想流式传输每个过程中的 Tokens，代码如下：
from langchain_core.messages import AIMessageChunk, HumanMessage
#inputs = [HumanMessage(content="what is the weather in sf")]
# messages模式下，只关心和打印AI、Tool、HumanMeesages这三种消息的内容
async def tokens_g():
first = True
async for msg, metadata in graph.astream({"messages": ["你好，帮我查询一下数据库中Shanghai的天气数据"]}, stream_mode="messages"):
if msg.content and not isinstance(msg, HumanMessage):
print(msg.content, end="|", flush=True)
if isinstance(msg, AIMessageChunk):
if first:
gathered = msg
first = False
else:
gathered = gathered + msg
if msg.tool_call_chunks:
print(gathered.tool_calls)
asyncio.run(tokens_g())
"""
AImessages
第一段：工具调用段（ 4 行 list）:print(gathered.tool_calls)
|[{'name': 'query_weather_from_db', 'args': {}, 'id': 'call_e8b44aa7cf614d43a61f829d', 'type': 'tool_call'}]
[{'name': 'query_weather_from_db', 'args': {}, 'id': 'call_e8b44aa7cf614d43a61f829d', 'type': 'tool_call'}]
[{'name': 'query_weather_from_db', 'args': {'city_name': 'Shanghai'}, 'id': 'call_e8b44aa7cf614d43a61f829d', 'type': 'tool_call'}]
[{'name': 'query_weather_from_db', 'args': {'city_name': 'Shanghai'}, 'id': 'call_e8b44aa7cf614d43a61f829d', 'type': 'tool_call'}]
这行 JSON 是 query_weather_from_db 工具查数据库返回的结果。
它是 ToolMessage 类型，由 ① 行的 print(msg.content, end="|") 打印（因为它不是 HumanMessage，所以会被打印；它也不是 AIMessageChunk，所以不会走 ② 行）。
{"city_id": 1796236, "city_name": "Shanghai", "main_weather": "Clouds", "description": "overcast clouds", "temperature": 28.28, "feels_like": 32.98, "temp_min": 28.28, "temp_max": 28.28}|
第二段：最终回答段: print(msg.content, end="|")
|根据|数据库|查询|结果|，上海|目前的|天气|数据|如下|：
||| 项目| || 数据| |
|||------|||------||
||| |城市|名称| || 上海| (|Shanghai|)| |
||| |城市|ID| || |179|623|6| |
||| |天气|状况| || 多|云| (|Cl|ouds|)| |
||| |详细|描述| || |阴|天| (|over|cast| clouds|)| |
||| 当前|温度| || |28|.|28|°C| |
||| |体感|温度| || |32|.|98|°C| |
||| |最低|温度| || |28|.|28|°C| |
||| |最高|温度| || |28|.|28|°C| |
|目前|上海|是|阴|天|，气|温|为|28|.|28|°C|，|体感|温度|略|高|，为|32|.|98|°C|。
|如果您|需要|查询|其他|城市的|天气|数据|，请|告诉我|！|
"""
III LangGraph 中的事件流
1 定义
对于上述使用的 .stream() 或 .astream() 仅流式传输链中最后一步的输出，这对于一些对话聊天类的应用程序来说基本就足够了，但是当我们的 AI Agent 是一个使用了多个大模型调用的更复杂的链时，我们有时希望在最终输出中也使用到一些中间值。
如果想获取到这样的中间事件和步骤，可以使用 LangGraph 框架中的 astream_events 方法，注意：此方法仅支持异步。用来访问自定义事件和中间输出。使用该方法运行图时，可以得到如下相关事件：
LangChain CallBack: https://python.langchain.com/docs/concepts/callbacks/
stream() / astream() 的局限性
前面介绍的四种stream_mode（values/updates/debug/messages）只能返回单节点执行后的状态 / 消息，无法完整捕获全链路中间事件：多轮 LLM 调用、工具执行、自定义埋点事件、回调生命周期等细分事件都会丢失，复杂多节点 Agent 场景不足以调试、采集中间过程数据。
可以基于事件、流模式
2 代码演示
from langchain_core.tools import tool
from typing import Union, Optional
from pydantic import BaseModel, Field
import requests
from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.orm import sessionmaker, declarative_base
import json
# 1、读取配置信息
from dotenv import load_dotenv
load_dotenv()
import os
passwrod = os.environ["MYSQL_PASSWORD"]
acc = os.environ["MYSQL_account"]
DATABASE_URI = f'mysql+pymysql://{acc}:{passwrod}@localhost/langgraph?charset=utf8mb4'
engine = create_engine(DATABASE_URI)
weather_key = os.environ["OPEN_WEATHER_KEY"]
GOOGLE_SEARCH_KEY = os.environ["GOOGLE_SEARCH_KEY"]
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key,base_url=base_url,temperature=0,)
# 数据库连接 URI，这里要替换成自己的Mysql 连接信息，以下是各个字段的对应解释：
# gpt: MySQL 数据库的用户名。
# gpt: MySQL 数据库的密码。
# localhost: MySQL 服务器的 IP 地址。
# langgraph: 要连接的数据库的名称。
# charset=utf8mb4: 设置数据库的字符集为 utf8mb4，支持更广泛的 Unicode 字符
# 创建会话
Session = sessionmaker(bind=engine)
# 创建基类
Base = declarative_base()
# 定义 WeatherInfo 模型
class Weather(Base):
__tablename__ = 'weather'
city_id = Column(Integer, primary_key=True) # 城市ID
city_name = Column(String(50)) # 城市名称
main_weather = Column(String(50)) # 主要天气状况
description = Column(String(100)) # 描述
temperature = Column(Float) # 温度
feels_like = Column(Float) # 体感温度
temp_min = Column(Float) # 最低温度
temp_max = Column(Float) # 最高温度
# 第一个工具
class WeatherLoc(BaseModel):
location: str = Field(description="The location name of the city")
class WeatherInfo(BaseModel):
"""Extracted weather information for a specific city."""
city_id: int = Field(..., description="The unique identifier for the city")
city_name: str = Field(..., description="The name of the city")
main_weather: str = Field(..., description="The main weather condition")
description: str = Field(..., description="A detailed description of the weather")
temperature: float = Field(..., description="Current temperature in Celsius")
feels_like: float = Field(..., description="Feels-like temperature in Celsius")
temp_min: float = Field(..., description="Minimum temperature in Celsius")
temp_max: float = Field(..., description="Maximum temperature in Celsius")
## 第一个工具：获取天气信息
@tool(args_schema = WeatherLoc)
def get_weather(location):
"""
Function to query current weather.
:param loc: Required parameter, of type string, representing the specific city name for the weather query. \
Note that for cities in China, the corresponding English city name should be used. For example, to query the weather for Beijing, \
the loc parameter should be input as 'Beijing'.
:return: The result of the OpenWeather API query for current weather, with the specific URL request address being: https://api.openweathermap.org/data/2.5/weather
The return type is a JSON-formatted object after parsing, represented as a string, containing all important weather information.
"""
# Step 1.构建请求
url = "https://api.openweathermap.org/data/2.5/weather"
# Step 2.设置查询参数
params = {
"q": location,
"appid": weather_key, # 输入API key
"units": "metric", # 使用摄氏度而不是华氏度
"lang":"zh_cn" # 输出语言为简体中文
}
# Step 3.发送GET请求
response = requests.get(url, params=params)
# Step 4.解析响应
data = response.json()
return json.dumps(data)
## 第二个工具
@tool(args_schema=WeatherInfo)
def insert_weather_to_db(city_id, city_name, main_weather, description, temperature, feels_like, temp_min, temp_max):
"""Insert weather information into the database."""
session = Session() # 确保为每次操作创建新的会话
try:
# 创建天气实例
weather = Weather(
city_id=city_id,
city_name=city_name,
main_weather=main_weather,
description=description,
temperature=temperature,
feels_like=feels_like,
temp_min=temp_min,
temp_max=temp_max
)
# 使用 merge 方法来插入或更新（如果已有记录则更新）
session.merge(weather)
# 提交事务
session.commit()
return {"messages": [f"天气数据已成功存储至Mysql数据库。"]}
except Exception as e:
session.rollback() # 出错时回滚
return {"messages": [f"数据存储失败，错误原因：{e}"]}
finally:
session.close() # 关闭会话
## 第三个工具
class QueryWeatherSchema(BaseModel):
"""Schema for querying weather information by city name."""
city_name: str = Field(..., description="The name of the city to query weather information")
@tool(args_schema=QueryWeatherSchema)
def query_weather_from_db(city_name: str):
"""Query weather information from the database by city name."""
session = Session()
try:
# 查询天气数据
weather_data = session.query(Weather).filter(Weather.city_name == city_name).first()
if weather_data:
return {
"city_id": weather_data.city_id,
"city_name": weather_data.city_name,
"main_weather": weather_data.main_weather,
"description": weather_data.description,
"temperature": weather_data.temperature,
"feels_like": weather_data.feels_like,
"temp_min": weather_data.temp_min,
"temp_max": weather_data.temp_max
}
else:
return {"messages": [f"未找到城市 '{city_name}' 的天气信息。"]}
except Exception as e:
return {"messages": [f"查询失败，错误原因：{e}"]}
finally:
session.close() # 关闭会话
## 第四个工具
class SearchQuery(BaseModel):
query: str = Field(description="Questions for networking queries")
@tool(args_schema = SearchQuery)
def fetch_real_time_info(query):
"""Get real-time Internet information"""
url = "https://google.serper.dev/search"
payload = json.dumps({
"q": query,
"num": 1,
})
headers = {
'X-API-KEY': GOOGLE_SEARCH_KEY,
'Content-Type': 'application/json'
}
response = requests.post(url, headers=headers, data=payload)
data = json.loads(response.text) # 将返回的JSON字符串转换为字典
if 'organic' in data:
return json.dumps(data['organic'], ensure_ascii=False) # 返回'organic'部分的JSON字符串
else:
return json.dumps({"error": "No organic results found"}, ensure_ascii=False) # 如果没有'organic'键，返回错误信息
# 测试天气工具调用示例
tool_result = get_weather.invoke('beijing')
# 注册全部可用工具列表
tools = [fetch_real_time_info, get_weather, insert_weather_to_db, query_weather_from_db]
# 基于预封装REACT模板创建Agent图
from langgraph.prebuilt import create_react_agent
graph = create_react_agent(llm, tools=tools)
# 演示1（values）在图表的每个步骤之后流式传输状态的完整值。
async def event_g():
async for event in graph.astream_events({"messages": ["你好，请你介绍一下你自己"]}, version="v2"):
kind = event["event"]
print(f"{kind}: {event['name']}")
import asyncio
# asyncio.run(event_g())
"""
graph = create_react_agent(llm, tools=tools)
on_chain_start: LangGraph
on_chain_start: agent
on_chain_start: call_model
on_chain_start: RunnableSequence
on_chain_start: Prompt
on_chain_end: Prompt
on_chat_model_start: ChatOpenAI
on_chat_model_stream: ChatOpenAI
on_chat_model_end: ChatOpenAI
on_chain_end: RunnableSequence
on_chain_end: call_model
on_chain_start: should_continue
on_chain_end: should_continue
on_chain_stream: agent
on_chain_end: agent
on_chain_stream: LangGraph
on_chain_end: LangGraph
"""
"""
这个过程明确标识了 Agent 执行的每个阶段。
从 on_chain_start: LangGraph 开始，写入 start 节点，
启动 call_model 节点（on_chain_start: call_model）。
然后开始聊天模型调用（on_chat_model_start: ChatOpenAI），
按 token 的增量流式返回（on_chat_model_stream: ChatOpenAI），
直到聊天模型（on_chat_model_end: ChatOpenAI）输出完全部内容后停止。
继而将结果写回通道（ChannelWrite<call_model,messages>），
再次回到 call_model 节点做决策，最终完成整个图的运行流程。
我们可以从中提取具体的某个 event（事件），比如：
"""
async def event2_g():
events = []
async for event in graph.astream_events({"messages": ["你好，请你介绍一下你自己"]}, version="v2"):
events.append(event)
print(events[0])
# asyncio.run(event2_g())
"""
{'event': 'on_chain_start', 'data': {'input': {'messages': ['你好，请你介绍一下你自己']}}, 'name': 'LangGraph', 'tags': [],
'run_id': '019f49ee-1ecf-7220-b1f9-91b91ef0a3f2', 'metadata': {'ls_integration': 'langgraph'}, 'parent_ids': []}
"""
"""
所有事件都会包含 event、name 和 data 字段，其中：
event：正在发出的事件类型。
name：这是事件的名称
data：这是与事件关联的数据。
基于此就可以按照 name、tags 或 type 等不同的字段来进行事件过滤，比如我们现在选择仅包含聊天模型的输出：
"""
# 一个事件就是一个chunk（token，一个字或者一个词组），打印出来密密麻麻会很多。
async def chat_output_g():
async for event in graph.astream_events({"messages": ["你好，请你介绍一下你自己"]}, version="v2"):
kind = event["event"]
if kind == "on_chat_model_stream":
print(event, end="|", flush=True)
# asyncio.run(chat_output_g())
# 每种类型的事件都包含不同格式的数据。而其中 data 是一个非常重要的，包含此事件的实际数据。在 on_chat_model_stream 事件中，
# 就是需要响应的流式 Token，如上图所示是一个 AIMessageChunk，其中包含消息的 content 以及 id，
# 提取的代码就非常简单了，和我们上面实现的方式一致，即直接采用如下代码：
from langchain_core.messages import AIMessageChunk, HumanMessage
async def extract_critical_info_g():
first = True
async for msg, metadata in graph.astream({"messages": ["你好，请你介绍一下你自己"]}, stream_mode="messages"):
if msg.content and not isinstance(msg, HumanMessage):
print(msg.content, end="|", flush=True)
if isinstance(msg, AIMessageChunk):
if first:
gathered = msg
first = False
else:
gathered = gathered + msg
if msg.tool_call_chunks:
print(gathered.tool_calls)
# asyncio.run(extract_critical_info_g())
# stream_mode="messages"模式是直接做的格式化提取的实现过程，当然，理解了上述事件流，我们也可以直接在当前的流程下自定义数据流，比如：
async def customer_extract_critical_info_g():
async for event in graph.astream_events({"messages": ["你好，请你介绍一下你自己"]}, version="v2"):
kind = event["event"]
if kind == "on_chat_model_stream":
print(event["data"]["chunk"].content, end="|", flush=True)
asyncio.run(customer_extract_critical_info_g())