How to get structured output when using astream() or stream()

OSS Product Help LangChain
1

Hello, I am using agent.astream() to print streaming message, but I don’t know how to get structured output like agent.invoke(). Anyone has some ideas ?


def print_msg(msg):
    if isinstance(msg, AIMessage):
        print(f"[AIMessage] content: {msg.content}")
        if msg.tool_calls:
            for tc in msg.tool_calls:
                print(f"  → tool call: {tc['name']}({tc['args']})  id={tc['id']}")
    elif isinstance(msg, ToolMessage):
        print(f"[ToolMessage] name={msg.name}  content: {msg.content[:300]}...")
    elif isinstance(msg, HumanMessage):
        print(f"[Human] {msg.content}")
    else:
        print(f"[{type(msg).__name__}] {msg}")


async def run_agent(msg:str,
                    provider_id:str,
                    job_env_name:str):
    thread_config = {"configurable": {"thread_id": "12345"}}
    structured_response = None
    async for chunk in agent.astream({"messages": [HumanMessage(msg)]}, 
                            stream_mode="updates", 
                            config=thread_config,
                            context=...,
                            version="v2"):
        for node_name, update in chunk.items():
            print(f"\n===== Node: {node_name} =====")
            if "messages" in update:
                for msg in update["messages"]:
                    print_msg(msg) 

            # does not work
            #if "structured_response" in chunk and chunk["structured_response"]:
            #   structured_response = chunk["structured_response"]
2

Hello @MSJavaScript ,

structured_response lives inside the node update, not on the chunk. With version="v2" the chunk is {type, ns, data}, so "structured_response" in chunk is always false.

structured_response = None
async for chunk in agent.astream(..., stream_mode="updates", version="v2"):
    if chunk["type"] != "updates":
        continue
    for node_name, update in chunk["data"].items():
        if "messages" in update:
            for m in update["messages"]:
                print_msg(m)
        if update.get("structured_response") is not None:
            structured_response = update["structured_response"]

Or stream full state and take the last snapshot: stream_mode="values" → chunk["data"]["structured_response"] (same as invoke()). Requires response_format on create_agent.

Langchain Docs Reference