如何在LangGraph节点中输出LLM Tokens？内部API调用场景求助

1. 修正生成节点逻辑 #

原代码存在两个核心问题：yield 位置错误（仅返回最后一个chunk）、未按LangGraph状态规范返回更新。修正后节点需流式输出每个token，并通过yield提交状态更新：

def generation_node(self, state):
    # 从状态中读取对话历史，无则初始化
    messages = state.get("messages", [])
    # 新建助手消息容器，用于追加流式token
    assistant_msg = {"role": "assistant", "content": ""}
    messages.append(assistant_msg)

    # 必须设置stream=True，开启请求流式响应
    with requests.post("some api url", stream=True) as response:
        # 按字节流式读取返回内容，chunk_size可根据API输出调整
        for chunk in response.iter_content(chunk_size=1):
            if chunk:
                token = chunk.decode()
                assistant_msg["content"] += token
                # 提交状态更新，LangGraph会捕获该中间状态用于流式输出
                yield {"messages": messages}
    
    # 可选：返回最终完整状态
    return {"messages": messages}

2. 正确配置工作流 #

原代码未将节点绑定到实现函数，导致编译失败。需补充节点与函数的关联，并设置入口点：

from langgraph.graph import END, StateGraph
from typing import List, Dict

# 定义状态类型（可选，用于类型提示）
class AgentState(Dict):
    messages: List[Dict]

# 初始化工作流
self.workflow = StateGraph(AgentState)
# 绑定节点到实现函数
self.workflow.add_node("generation", self.generation_node)
# 设置工作流入口为generation节点
self.workflow.set_entry_point("generation")
self.workflow.add_edge("generation", END)
# 编译工作流
self.compiled_workflow = self.workflow.compile()

3. 流式运行并输出到前端 #

使用LangGraph的astream_events（异步）或astream（同步）方法运行工作流，捕获每个流式token并传递给前端：

# 初始状态：用户提问
initial_state = {"messages": [{"role": "user", "content": "请介绍LangGraph"}]}

# 异步方式获取流式事件（推荐用于Web场景）
async for event in self.compiled_workflow.astream_events(initial_state, version="v1"):
    # 过滤generation节点的流式输出事件
    if event["event"] == "on_chain_stream" and event["name"] == "generation":
        # 提取最新生成的token内容
        latest_content = event["data"]["output"]["messages"][-1]["content"]
        # 此处可将latest_content通过WebSocket/HTTP流式响应发送到前端
        print(latest_content, end="", flush=True)

关键注意事项 #

• 必须为requests.post设置stream=True，否则API会一次性返回所有内容，无法实现流式输出
• 生成节点必须通过yield提交状态更新，而非直接返回字符串，这是LangGraph处理流式输出的核心要求
• 不能使用普通的run()方法运行工作流，必须用astream或astream_events才能捕获中间输出

内容的提问来源于stack exchange，提问作者Chengjian Jin