澄清一個(gè)Langgraph中的bug——在智能體中獲取到工具節(jié)點(diǎn)中的流式輸出 原創(chuàng)
“ Langgraph對模型功能進(jìn)行了適度封裝，有些可能乍一看很不合理，因此我們要多看其官網(wǎng)說明?！?/strong>
前一段時(shí)間在學(xué)習(xí)使用Langgraph開發(fā)框架時(shí)，由于對框架理解不夠深刻，以為發(fā)現(xiàn)了一個(gè)bug，然后今天才發(fā)現(xiàn)這不是bug，而是框架本身就是這么設(shè)計(jì)的。
這個(gè)bug就是，在智能體的流式輸出的時(shí)候，會一塊輸出工具節(jié)點(diǎn)調(diào)用模型結(jié)果；之所以發(fā)現(xiàn)這個(gè)問題就是在具體的工具中又調(diào)用了模型進(jìn)行數(shù)據(jù)處理，但在打印輸出的時(shí)候發(fā)現(xiàn)工具的執(zhí)行過程也被輸出了出來。
如下圖所示，在工具chat中調(diào)用了模型：
Langgraph流式輸出問題
查看Langgraph官網(wǎng)介紹發(fā)現(xiàn)，其本身提供了從任何地方捕獲令牌流等流式輸出。
這種設(shè)計(jì)哲學(xué)有好處也有壞處，好處就是在最上層可以捕獲到所有子節(jié)點(diǎn)的輸出流；壞處就是，我們可能并不需要子節(jié)點(diǎn)的輸出流。
所以怎么解決這個(gè)問題呢？
就是使用標(biāo)簽過濾，在Langgraph中提供了元數(shù)據(jù)(metadata)的功能，可以通過給模型添加標(biāo)簽數(shù)據(jù)，以此來區(qū)分不同功能的流式輸出，防止數(shù)據(jù)污染。
代碼案例如下：
from langchain.chat_models import init_chat_model
llm_1 = init_chat_model(model="openai:gpt-4o-mini", tags=['joke'])
llm_2 = init_chat_model(model="openai:gpt-4o-mini", tags=['poem'])
graph = ... # define a graph that uses these LLMs
async for msg, metadata in graph.astream(
{"topic": "cats"},
stream_mode="messages",
):
if metadata["tags"] == ["joke"]:
print(msg.content, end="|", flush=True)文檔地址如下: https://github.langchain.ac.cn/langgraph/how-tos/streaming/#messages
或者也可以按照節(jié)點(diǎn)名稱進(jìn)行篩選：
from typing import TypedDict
from langgraph.graph import START, StateGraph
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini")
class State(TypedDict):
topic: str
joke: str
poem: str
def write_joke(state: State):
topic = state["topic"]
joke_response = model.invoke(
[{"role": "user", "content": f"Write a joke about {topic}"}]
)
return {"joke": joke_response.content}
def write_poem(state: State):
topic = state["topic"]
poem_response = model.invoke(
[{"role": "user", "content": f"Write a short poem about {topic}"}]
)
return {"poem": poem_response.content}
graph = (
StateGraph(State)
.add_node(write_joke)
.add_node(write_poem)
# write both the joke and the poem concurrently
.add_edge(START, "write_joke")
.add_edge(START, "write_poem")
.compile()
)
for msg, metadata in graph.stream(
{"topic": "cats"},
stream_mode="messages",
):
if msg.content and metadata["langgraph_node"] == "write_poem":
print(msg.content, end="|", flush=True)通過這種方式就可以對不同節(jié)點(diǎn)的模型輸出進(jìn)行過濾，實(shí)現(xiàn)節(jié)點(diǎn)之間的數(shù)據(jù)分隔。
當(dāng)然，Langgraph還提供了自定義數(shù)據(jù)，使用Langgraph提供的輸出類型會對模型輸出進(jìn)行一定的限制；因此，如果用對靈活性要求較高的情況下，就可以使用自定義數(shù)據(jù)類型。
總之，Langgraph對大模型的流式輸出進(jìn)行了適度的封裝，如果沒搞明白其封裝的特性，可能會出現(xiàn)一些意想不到的問題。
本文轉(zhuǎn)載自???AI探索時(shí)代??? 作者：DFires
