My stream output mode is configured for “updates” and “custom”, but the mode being printed is “values”

[WangJie-7]

Checked other resources

• This is a bug, not a usage question. For questions, please use GitHub Discussions.
• I added a clear and detailed title that summarizes the issue.
• I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example).
• I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue.

Example Code

async def execute_step(state: PlanExecute):
    today = date.today()
    today_str_cn = today.strftime("%Y year,%m month,%d day")
    plan = state["plan"]
    plan_str = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(plan))
    task = plan[0]
    task_formatted = f"""For the following plan:
{plan_str}\n\nYou are tasked with executing step {1}, {task}.
If this step includes words like ‘latest’ or ‘today,’ query bing tool for information on
""" + today_str_cn + "Otherwise, do not add this date to the tool parameters."
    for stream_mode, chunk in agent_executor.stream(
            {"messages": task_formatted},
            stream_mode=["updates", "custom"],
    ):
        print(f"Stream mode: {stream_mode}")
        print(chunk)
        print("\n")

Error Message and Stack Trace (if applicable)

Stream mode: values
{'messages': [HumanMessage(content='For the following plan:\n1. You are tasked with executing step 1, 打开geojson_render工具.\nIf this step includes words like ‘latest’ or ‘today,’ query bing tool for information on\n2025 year,03 month,11 dayOtherwise, do not add this date to the tool parameters.', additional_kwargs={}, response_metadata={}, id='46d24327-b50c-44a1-9643-f5fcb45ed050')]}

Description

The execute_step function is a node in my graph

System Info

System Information

OS: Windows
OS Version: 10.0.26100
Python Version: 3.11.9 (tags/v3.11.9:de54cf5, Apr 2 2024, 10:12:12) [MSC v.1938 64 bit (AMD64)]

Package Information

langchain_core: 0.3.43
langchain: 0.3.20
langchain_community: 0.3.19
langsmith: 0.3.11
langchain_chroma: 0.2.2
langchain_openai: 0.3.7
langchain_text_splitters: 0.3.6
langgraph_bigtool: 0.0.2
langgraph_sdk: 0.1.53

Optional packages not installed

langserve

Other Dependencies

aiohttp<4.0.0,>=3.8.3: Installed. No version info available.
async-timeout<5.0.0,>=4.0.0;: Installed. No version info available.
chromadb!=0.5.10,!=0.5.11,!=0.5.12,!=0.5.4,!=0.5.5,!=0.5.7,!=0.5.9,<0.7.0,>=0.4.0: Installed. No version info available.
dataclasses-json<0.7,>=0.5.7: Installed. No version info available.
httpx: 0.28.1
httpx-sse<1.0.0,>=0.4.0: Installed. No version info available.
jsonpatch<2.0,>=1.33: Installed. No version info available.
langchain-anthropic;: Installed. No version info available.
langchain-aws;: Installed. No version info available.
langchain-cohere;: Installed. No version info available.
langchain-community;: Installed. No version info available.
langchain-core!=0.3.0,!=0.3.1,!=0.3.10,!=0.3.11,!=0.3.12,!=0.3.13,!=0.3.14,!=0.3.2,!=0.3.3,!=0.3.4,!=0.3.5,!=0.3.6,!=0.3.7,!=0.3.8,!=0.3.9,<0.4.0,>=0.2.43: Installed. No version info available.
langchain-core<1.0.0,>=0.3.34: Installed. No version info available.
langchain-core<1.0.0,>=0.3.39: Installed. No version info available.
langchain-core<1.0.0,>=0.3.41: Installed. No version info available.
langchain-deepseek;: Installed. No version info available.
langchain-fireworks;: Installed. No version info available.
langchain-google-genai;: Installed. No version info available.
langchain-google-vertexai;: Installed. No version info available.
langchain-groq;: Installed. No version info available.
langchain-huggingface;: Installed. No version info available.
langchain-mistralai;: Installed. No version info available.
langchain-ollama;: Installed. No version info available.
langchain-openai;: Installed. No version info available.
langchain-text-splitters<1.0.0,>=0.3.6: Installed. No version info available.
langchain-together;: Installed. No version info available.
langchain-xai;: Installed. No version info available.
langchain<1.0.0,>=0.3.20: Installed. No version info available.
langgraph<0.4.0,>=0.3.0: Installed. No version info available.
langsmith-pyo3: Installed. No version info available.
langsmith<0.4,>=0.1.125: Installed. No version info available.
langsmith<0.4,>=0.1.17: Installed. No version info available.
numpy<2.0.0,>=1.22.4;: Installed. No version info available.
numpy<2.0.0,>=1.26.2;: Installed. No version info available.
numpy<3,>=1.26.2: Installed. No version info available.
openai<2.0.0,>=1.58.1: Installed. No version info available.
orjson: 3.10.15
packaging: 24.2
packaging<25,>=23.2: Installed. No version info available.
pydantic: 2.10.6
pydantic-settings<3.0.0,>=2.4.0: Installed. No version info available.
pydantic<3.0.0,>=2.5.2;: Installed. No version info available.
pydantic<3.0.0,>=2.7.4: Installed. No version info available.
pydantic<3.0.0,>=2.7.4;: Installed. No version info available.
pytest: Installed. No version info available.
PyYAML>=5.3: Installed. No version info available.
requests: 2.32.3
requests-toolbelt: 1.0.0
requests<3,>=2: Installed. No version info available.
rich: 13.9.4
SQLAlchemy<3,>=1.4: Installed. No version info available.
tenacity!=8.4.0,<10,>=8.1.0: Installed. No version info available.
tenacity!=8.4.0,<10.0.0,>=8.1.0: Installed. No version info available.
tiktoken<1,>=0.7: Installed. No version info available.
typing-extensions>=4.7: Installed. No version info available.
zstandard: 0.23.0

[vbarda]

Wang Jie (@WangJie-7) you would need to provide a minimal reproducible example. but from reading your description it sounds like you return full updates state from your node instead of just an update. stream_mode="updates" simply streams the outputs from the nodes

[WangJie-7]

Vadym Barda (@vbarda) this is all my code:

model = ChatOpenAI(model="gpt-4o-mini", streaming=True, temperature=0)


@tool
def geojson_render(geo) -> str:
    """
    根据数据生成渲染数据结果。
    Args:
        geo: 地理数据（例如GeoJSON对象）

    Returns:
        return: 渲染结果
    """
    data = geo
    # 示例逻辑：这里可以替换为实际渲染代码
    return "已经成功渲染数据"


@tool
def bingSearch(query) -> str:
    """
    你只需要获取地理信息数据才用这个工具

    Args:
        query: a search query

    Returns:
        return: an array of the query results
    """
    api_wrapper = BingSearchAPIWrapper(k=3)
    bing = BingSearchResults(api_wrapper=api_wrapper)
    return bing.run(query)  # 实际调用搜索


# 工具初始化（封装为函数以便复用）
tools = [geojson_render, bingSearch]
prompt = "你是一个有用的助手"
# tool agent
agent_executor = create_react_agent(model, tools, prompt=prompt)




class PlanExecute(TypedDict):
    input: str
    plan: List[str]
    past_steps: Annotated[List[Tuple], operator.add]
    response: str


class Plan(BaseModel):
    """Plan to follow in future"""

    steps: List[str] = Field(
        description="different steps to follow, should be in sorted order"
    )


planner_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            """
            For the given objective, come up with a simple step by step plan. \
This plan should involve individual tasks, that if executed correctly will yield the correct answer. Do not add any superfluous steps. \
The result of the final step should be the final answer. Make sure that each step has all the information needed - do not skip steps.""",
        ),
        ("placeholder", "{messages}"),
    ]
)
planner = planner_prompt | model.with_structured_output(Plan)


class Response(BaseModel):
    """Response to user."""

    response: str


class Act(BaseModel):
    """Action to perform."""

    action: Union[Response, Plan] = Field(
        description="Action to perform. If you want to respond to user, use Response. "
                    "If you need to further use tools to get the answer, use Plan."
    )


replanner_prompt = ChatPromptTemplate.from_template(
    """
Your objective was this:
{input}

Your original plan was this:
{plan}

You have currently done the follow steps:
{past_steps}

Based on the original plan and the steps that you have currently done, 
if the objective requirements have been met, consolidate the target’s answer and return it as a Response object.
Otherwise,based on the original plan and the steps that you have currently done,come up with a simple, step-by-step revised plan to achieve this objective.
Do not add any superfluous steps. Do not include previously completed steps as part of the plan.
 """
)

replanner = replanner_prompt | model.with_structured_output(Act)


async def execute_step(state: PlanExecute):
    today = date.today()
    today_str_cn = today.strftime("%Y year,%m month,%d day")
    plan = state["plan"]
    plan_str = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(plan))
    task = plan[0]
    task_formatted = f"""For the following plan:
{plan_str}\n\nYou are tasked with executing step {1}, {task}.
If this step includes words like ‘latest’ or ‘today,’ query bing tool for information on
""" + today_str_cn + "Otherwise, do not add this date to the tool parameters."
    for stream_mode, chunk in agent_executor.stream(
            {"messages": task_formatted},
            stream_mode=["updates", "custom"],
    ):
        print(f"Stream mode: {stream_mode}")
        print(chunk)
        print("\n")

    return {
        "past_steps": [(task, "已完成")],
    }


async def plan_step(state: PlanExecute):
    plan = await planner.ainvoke({"messages": [("user", state["input"])]})
    return {"plan": plan.steps}


async def replan_step(state: PlanExecute):
    output = await replanner.ainvoke(state)
    if isinstance(output.action, Response):
        return {"response": output.action.response}
    else:
        return {"plan": output.action.steps}


def should_end(state: PlanExecute):
    if "response" in state and state["response"]:
        return END
    else:
        return "agent"


workflow = StateGraph(PlanExecute)

# Add the plan node
workflow.add_node("planner", plan_step)

# Add the execution step
workflow.add_node("agent", execute_step)

# Add a replan node
workflow.add_node("replan", replan_step)

workflow.add_edge(START, "planner")

# From plan we go to agent
workflow.add_edge("planner", "agent")

# From agent, we replan
workflow.add_edge("agent", "replan")

workflow.add_conditional_edges(
    "replan",
    # Next, we pass in the function that will determine which node is called next.
    should_end,
    ["agent", END],
)

# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
graph = workflow.compile()
print(graph.get_graph().draw_ascii())

config = {"recursion_limit": 50}
inputs = {
    "input": '根据这个地理数据{"type": "Feature", "properties": {"population": "1,000,000"}}输入到geojson_render工具中给出我渲染结果'}


async def process_events():
    flag = 0
    async for message_chunk, metadata in graph.astream(inputs, config=config, stream_mode="messages"):
        if message_chunk.content:
            if message_chunk.content == "response":
                flag = 1
            if flag == 1:
                print(message_chunk.content, end='')


# # 执行异步函数
asyncio.run(process_events())

[vbarda]

Wang Jie (@WangJie-7) ah, i see. when a graph is invoked inside another graph's node, the stream mode is overwritten to be stream_mode="values". if you want to stream data from the subgraphs, you can add subgraphs=True on the parent graph - this is the recommended way of streaming from subgraphs

graph.astream(inputs, config=config, stream_mode=["messages", ...], subgraphs=True)

[WangJie-7]

Vadym Barda (@vbarda) I added subgraphs=True, but the stream mode updates and custom logic used in the subgraphs are still printing in values mode. I don’t think execute_step is a subgraph. This is really frustrating.

[vbarda]

Wang Jie (@WangJie-7) i am sorry it's frustrating

execute_step is not a subgraph, but you are invoking a graph inside it (agent_executor is a graph). any graph invoked inside a node function of a parent graph is by definition a subgraph

i would recommend replacing this

 for stream_mode, chunk in agent_executor.stream(
            {"messages": task_formatted},
            stream_mode=["updates", "custom"],
    ):
        print(f"Stream mode: {stream_mode}")
        print(chunk)
        print("\n")

with just

results = agent_executor.invoke({"messages": task_formatted})
...

and then streaming from parent graph with subgraphs=True

but the stream mode updates and custom logic used in the subgraphs are still printing in values mode.

subgraphs=True won't change that, it will just stream data from inside subgraphs

[WangJie-7]

Vadym Barda (@vbarda) yes！you are right.It seems the problem has been resolved. Thank you for your reply.

[vbarda]

Awesome, glad I could help!