澄清一個Langgraph中的bug——在智能體中獲取到工具節點中的流式輸出原創

AI探索時代

發布于 2025-10-9 06:50

瀏覽

0收藏

“ Langgraph對模型功能進行了適度封裝，有些可能乍一看很不合理，因此我們要多看其官網說明。”

前一段時間在學習使用Langgraph開發框架時，由于對框架理解不夠深刻，以為發現了一個bug，然后今天才發現這不是bug，而是框架本身就是這么設計的。

這個bug就是，在智能體的流式輸出的時候，會一塊輸出工具節點調用模型結果；之所以發現這個問題就是在具體的工具中又調用了模型進行數據處理，但在打印輸出的時候發現工具的執行過程也被輸出了出來。

如下圖所示，在工具chat中調用了模型：

澄清一個Langgraph中的bug——在智能體中獲取到工具節點中的流式輸出-AI.x社區

Langgraph流式輸出問題

查看Langgraph官網介紹發現，其本身提供了從任何地方捕獲令牌流等流式輸出。

澄清一個Langgraph中的bug——在智能體中獲取到工具節點中的流式輸出-AI.x社區

這種設計哲學有好處也有壞處，好處就是在最上層可以捕獲到所有子節點的輸出流；壞處就是，我們可能并不需要子節點的輸出流。

所以怎么解決這個問題呢？

就是使用標簽過濾，在Langgraph中提供了元數據(metadata)的功能，可以通過給模型添加標簽數據，以此來區分不同功能的流式輸出，防止數據污染。

澄清一個Langgraph中的bug——在智能體中獲取到工具節點中的流式輸出-AI.x社區

代碼案例如下：

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

或者也可以按照節點名稱進行篩選：

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)

通過這種方式就可以對不同節點的模型輸出進行過濾，實現節點之間的數據分隔。

當然，Langgraph還提供了自定義數據，使用Langgraph提供的輸出類型會對模型輸出進行一定的限制；因此，如果用對靈活性要求較高的情況下，就可以使用自定義數據類型。

澄清一個Langgraph中的bug——在智能體中獲取到工具節點中的流式輸出-AI.x社區