LangChain .stream() — output arrives as one giant chunk

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, PydanticOutputParser
from pydantic import BaseModel

class Answer(BaseModel):
    answer: str
    confidence: float

prompt = ChatPromptTemplate.from_template("Answer: {question}")
model = ChatOpenAI(model="gpt-4o-mini")
parser = PydanticOutputParser(pydantic_object=Answer)

# Chain A: streams ✓
chain_a = prompt | model | StrOutputParser()

# Chain B: does not stream past the parser ✗
chain_b = prompt | model | parser

async def diagnose(chain, name):
    print(f"\n=== {name} ===")
    chunk_count = 0
    async for event in chain.astream_events({"question": "hi"}, version="v2"):
        kind = event["event"]
        if kind in ("on_chat_model_stream", "on_parser_stream", "on_chain_stream"):
            data = event.get("data", {}).get("chunk", "")
            if hasattr(data, "content"):
                data = data.content
            if data:
                chunk_count += 1
                print(f"  [{event['name']:20s}] chunk: {str(data)[:40]!r}")
    print(f"  Total streaming chunks: {chunk_count}")

asyncio.run(diagnose(chain_a, "chain_a: prompt | model | StrOutputParser"))
asyncio.run(diagnose(chain_b, "chain_b: prompt | model | PydanticOutputParser"))

# chain_a: many chunks streaming from model and passing through StrOutputParser
# chain_b: chunks from model, then a single emission at the end from parser

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
from typing import List

# ❌ NON-STREAMING — PydanticOutputParser buffers
# from langchain_core.output_parsers import PydanticOutputParser
# chain = prompt | model | PydanticOutputParser(pydantic_object=Answer)

# ✓ STREAMING — JsonOutputParser emits progressive partial dicts
class Answer(BaseModel):
    steps: List[str] = Field(description="Reasoning steps")
    final: str = Field(description="Final answer")

# Use JSON mode on the model, JsonOutputParser at the end
model_json = ChatOpenAI(model="gpt-4o-mini").bind(
    response_format={"type": "json_object"}
)

parser = JsonOutputParser(pydantic_object=Answer)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer as JSON. {format_instructions}"),
    ("human", "{question}"),
]).partial(format_instructions=parser.get_format_instructions())

chain = prompt | model_json | parser

# Now streaming yields progressively-populated dicts:
for chunk in chain.stream({"question": "Explain photosynthesis"}):
    print(chunk)
# {}
# {'steps': []}
# {'steps': ['Plants absorb']}
# {'steps': ['Plants absorb sunlight']}
# {'steps': ['Plants absorb sunlight', 'Chlorophyll captures']}
# ...
# {'steps': [...], 'final': 'Photosynthesis converts...'}

from typing import Iterator
from langchain_core.runnables import RunnableGenerator
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4o-mini")

# ❌ BUFFERING — full text arrives, then transformation runs
# chain = model | StrOutputParser() | (lambda s: s.upper())

# ✓ STREAMING — generator preserves per-token flow
def upper_generator(chunks: Iterator[str]) -> Iterator[str]:
    """Uppercase every chunk as it arrives."""
    for chunk in chunks:
        yield chunk.upper()

chain = model | StrOutputParser() | RunnableGenerator(upper_generator)

for chunk in chain.stream("tell me a haiku"):
    print(chunk, end="", flush=True)
# Streams uppercase chunks in real time

# For async chains, use an async generator
async def upper_agen(chunks):
    async for chunk in chunks:
        yield chunk.upper()

# Async version
achain = model | StrOutputParser() | RunnableGenerator(upper_agen)

# For running arbitrary sync transformations that ARE per-chunk:
# Just use a lambda with StrOutputParser BEFORE it, and remember every
# lambda receives ONE full accumulated input by default. For per-chunk,
# always use a generator.