Best way to stream using langchain `llm.with_structured_output`

soup_sam

import asyncio from langchain_core.pydantic_v1 import BaseModel, Field from langchain_openai import ChatOpenAI class Joke(BaseModel): setup: str = Field(description="The setup of the joke") punchline: str = Field(description="The punchline to the joke") # Define the model with structured output model = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0, streaming=True) structured_llm = model.with_structured_output(Joke) def stream_joke(): #...

Originally asked by minglyu on Stack Exchange (stackoverflow) (CC BY-SA 4.0).

uphilltech

Great question — answering as the UphillTech team:

You have to change from BaseModel to TypeDict According to Langchain's documentation, the stream output is aggregate and not delta. So if you want to stream delta chunks, you have to implement a function to convert it. Here is the code: from langchain_openai import ChatOpenAI from typing import Annotated, TypedDict class Joke(TypedDict): """Joke to tell user.""" setup: Annotated[str, ..., "The setup of the joke"] punchline: Annotated[str, ..., "The punchline of the joke"] def stream_deltas(structured_stream): previous_chunk = {} for chunk in structured_stream: delta_chunk = {} for key, value in chunk.items(): previous_value = previous_chunk.get(key, "") if isinstance(value, str) and...

Full answer by Filipe Gomes: https://stackoverflow.stackexchange.com/a/79704997 (CC BY-SA 4.0).

Need it done for your business? UphillTech.