我今天终于逮到了现行，DeepSeek一本正经的胡说八道，之前让DeepSeek写前端或者后端代码感觉还挺好的，生成代码基本复制过去就能用，讨论好前后端架构设计后，让他直接根据讨论好的架构设计直接生成代码，基本复制过去就能跑通，我也是很赞赏他的，但是，让他排查Langgraph的代码就完全不懂装懂，一本正经的胡说八道，让我在错误的道理上越走越远。 我写个很小的场景，Langgraph流式输出信息，
stream_mode 只有一个值
for metadata,chunk in graph.stream(
Command(resume=val),
config=config,
stream_mode="messages"
):
if isinstance(chunk, AIMessageChunk) and chunk.content:
print(chunk.content, end="", flush=True)
stream_mode 有多个值
for mode, data in graph.stream(
Command(resume=input_value),
config=config,
stream_mode=["messages","updates","custom"],
):
if mode == "messages":
# LLM 的 Token 流
token, metadata = data
print(token.content, end="", flush=True)
elif mode == "updates":
# 节点完成后的状态增量
rprint(f"\n[Update] 节点完成: {list(data.keys())}")
rprint("uu",data)
elif mode == "custom":
rprint("cc",data)
# 你在节点里手动发出的进度事件
rprint(f"\n[Custom] {data['msg']}")
stream_mode只有一个值时，解构出两个值 metadata,chunk ，生成的内容就在chunk.content里，但是有多个值时，解构出第一个值时mode模式名称，第二个值还能再解构出 token, metadata两个值，生成的内容在token.content里。
但是我不小心把多值模式下的两个值位置搞反了，我就无法输出打字效果，我就问DeepSeek，它一本正经的说出很多种可能原因，我就跟着在它的坑里翻滚了好久出不来
没有办法我只能翻出之前写的代码，一行行比对，才找出这个小问题,我重开一个对话，把同样的问题问他，它还是绕不出来。
所以得出结论，DeepSeek写Langgraph能力很差，小伙伴们一定不能全信它
最后贴上代码
from langchain_core.messages import HumanMessage, AIMessageChunk
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.config import get_stream_writer
from langgraph.graph import StateGraph, START, END, MessagesState
from dotenv import load_dotenv
from langgraph.types import interrupt, Command
load_dotenv(override=True)
from rich import print as rprint
# 1. 初始化 DeepSeek 模型（按你的配置）
model = init_chat_model(
model="deepseek-chat",
extra_body={"thinking": {"type": "disabled"}},
)
# 2. 定义图状态
class State(MessagesState):
text: str
# 3. 节点函数：内部发射 custom 进度事件，同时调用模型产生 messages 流
def generate_node(state: State):
writer = get_stream_writer()
writer({"type": "progress", "stage": "start", "msg": "开始生成内容..."})
val=interrupt("请输入")
# 调用 DeepSeek 模型（messages 模式会自动流式输出这里的 Token）
response = model.invoke(
[HumanMessage(val)]
)
writer({"type": "progress", "stage": "finish", "msg": "生成完成"})
return {"messages": [response]}
# 4. 构建图
builder = StateGraph(State)
builder.add_node("generate", generate_node)
builder.add_edge(START, "generate")
builder.add_edge("generate", END)
checkpointer=InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {
"configurable": {"thread_id": "966347"}
}
graph.invoke({},config=config,)
# 5. 同时使用三种流模式
input_value= input("请输入")
for mode, data in graph.stream(
Command(resume=input_value),
config=config,
stream_mode=["messages","updates","custom"],
):
if mode == "messages":
# LLM 的 Token 流
token, metadata = data
print(token.content, end="", flush=True)
elif mode == "updates":
# 节点完成后的状态增量
rprint(f"\n[Update] 节点完成: {list(data.keys())}")
rprint("uu",data)
elif mode == "custom":
rprint("cc",data)
# 你在节点里手动发出的进度事件
rprint(f"\n[Custom] {data['msg']}")
本文由 AI图鉴自动抓取并翻译，版权归原作者所有；如需查看原文请点下方链接。
I finally caught DeepSeek today with a serious piece of bullshit, and it was nice to have DeepSeek write a front or back end code before, to have the code basically copied in the past, to talk about the design of the back and back structures, to have him generate the code directly on the basis of the discussed architecture design, and to run through the past, and I appreciate that, but, having him run over Langgraph's code, I don't get it. I'm writing a very small scene, Langgraph-flowing output.
stream mode only
for metadata,chunk in graph.stream(
Command(resume=val),
config=config,
stream_mode="messages"
):
if isinstance(chunk, AIMessageChunk) and chunk.content:
print(chunk.content, end="", flush=True)
stread mode has multiple values
for mode, data in graph.stream(
Command(resume=input_value),
config=config,
stream_mode=["messages","updates","custom"],
):
if mode == "messages":
# LLM token flow
token, metadata = data
print(token.content, end="", flush=True)
elif mode == "updates":
# State increment after completion of node
rprint(f"\n[Update] node finished: {list(data.keys()}”))
rprint("uu",data)
elif mode == "custom":
rprint("cc",data)
# The progress you're making at the node
rprint(f"\n[Custom] {data['msg']}")
When stream mode has only one value, the two values metada, chunk generated are in chunk.content, but when there are multiple values, the first mode mode name is deconstructed, and the second value is reconstructed token, metada and the resulting content is in token.content.
But I accidentally turned the two points under the multi-value pattern upside down, and I couldn't output the typing effect, and I asked DeepSeek, who was serious about a number of possible reasons, and I followed him to the pit for a long time.
There's no way I can figure out this little problem by flipping out the code I wrote earlier, by matching lines, and I'll reopen the conversation and ask him the same question.
So it's concluded that DeepSeek wrote that Langgraph's skills are so bad that the little guys can't trust it all.
Finally put the code on.
from langchain_core.messages import HumanMessage, AIMessageChunk
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.config import get_stream_writer
from langgraph.graph import StateGraph, START, END, MessagesState
from dotenv import load_dotenv
from langgraph.types import interrupt, Command
load_dotenv(override=True)
from rich import print as rprint
#1. Initialize the DeepSeek model (as you configure)
model = init_chat_model(
model="deepseek-chat",
extra_body={"thinking": {"type": "disabled"}},
)
2. Define the chart status
class State(MessagesState):
text: str
#3. Node function: Internal launch Custodial progress events, and call on models to generate message flows
def generate_node(state: State):
writer = get_stream_writer()
Writer ("type": "product", "stage": "start", "msg": "start generating content...")
val=interrupt
# Call the DeepSeek model (messages mode autostreams token)
response = model.invoke(
[HumanMessage(val)]
)
Writer.
return {"messages": [response]}
#4 Build Chart
builder = StateGraph(State)
builder.add_node("generate", generate_node)
builder.add_edge(START, "generate")
builder.add_edge("generate", END)
checkpointer=InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {
"configurable": {"thread_id": "966347"}
}
graph.invoke({},config=config,)
# 5. Use three flow patterns at the same time
input value=input( "please enter")
for mode, data in graph.stream(
Command(resume=input_value),
config=config,
stream_mode=["messages","updates","custom"],
):
if mode == "messages":
# LLM token flow
token, metadata = data
print(token.content, end="", flush=True)
elif mode == "updates":
# State increment after completion of node
rprint(f"\n[Update] node finished: {list(data.keys()}”))
rprint("uu",data)
elif mode == "custom":
rprint("cc",data)
# The progress you're making at the node
rprint(f"\n[Custom] {data['msg']}")
Auto-fetched and translated by AI Atlas. All rights belong to the original author.