用 LangGraph 实现基础流式打字效果

作者:袖梨 2026-09-14

在基于大语言模型构建交互应用时,一次性显示完整回复会让等待过程显得生硬。借助 LangGraph 的状态管理与流式执行能力,可以将模型生成的内容按片段即时输出。下面从 MessagesState 的继承方式入手,搭建一个包含两个节点的状态图,并实现终端打字效果。

from langchain_core.messages import AIMessageChunk
from langgraph.graph.message import add_messages,MessagesState
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
load_dotenv()
llm=init_chat_model(
  model="deepseek-chat"
)
class overAllState(MessagesState):
  userValue: str
  mes: str

def node1(state:overAllState):
  return {
    "userValue": [HumanMessage(state["userValue"])]
  }
def node2(state:overAllState):
  res=llm.invoke(state["userValue"])
  return {
    "messages":[res],
    "mes": res.content
  }
graph=(
  StateGraph(overAllState)
  .add_node(node1)
  .add_node(node2)
  .add_edge(START, "node1")
  .add_edge("node1", "node2")
  .add_edge("node2", END)
  .compile()
)
display.display(graph)

for chunk in graph.stream(
    {"userValue":"你是谁"},
    stream_mode="messages",
):
    msg, meta = chunk
    if isinstance(msg, AIMessageChunk) and msg.content:
        print(msg.content, end="", flush=True)

image.png
全局状态继承官方的MessagesState

image.png
这是MessagesState的源码,它帮用户定义了

messages: Annotated[list[AnyMessage], add_messages]

相关文章

精彩推荐