LangChain中使用llm.with_structured_output实现流式输出的最佳实践是什么？

原代码问题分析 #

你提供的代码无法正常运行，核心问题在于对with_structured_output配合stream方法的返回值处理错误：

• structured_llm.stream()返回的是部分结构化对象（Partial[Joke]），而非字典，因此用'setup' in part的判断方式不生效
• 未处理属性为None的情况，直接拼接会引发类型错误

最佳实践步骤 #

• 确认模型兼容性：必须使用支持JSON模式流式输出的OpenAI模型，比如gpt-3.5-turbo-0125、gpt-4-0125-preview及更高版本，这些模型能在生成过程中逐步输出结构化片段
• 正确处理流式返回的部分对象：直接访问对象属性，判断属性是否非空后再进行拼接
• 同步/异步调用适配：同步场景直接遍历stream()返回的迭代器，异步场景使用astream()并配合async for

修正后的代码 #

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")

# 初始化支持流式结构化输出的模型
model = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0, streaming=True)
structured_llm = model.with_structured_output(Joke)

def stream_joke():
    # 获取流式迭代器
    response_stream = structured_llm.stream("Tell me a joke about cats")
    
    # 初始化空的Joke对象用于累积结果
    current_joke = Joke(setup="", punchline="")
    
    for part in response_stream:
        # 处理setup部分的流式输出
        if part.setup is not None:
            current_joke.setup += part.setup
            print(f"Setup: {current_joke.setup}")
        # 处理punchline部分的流式输出
        if part.punchline is not None:
            current_joke.punchline += part.punchline
            print(f"Punchline: {current_joke.punchline}")

# 执行流式生成
stream_joke()

代码说明 #

1. 去掉了不必要的asyncio导入，因为当前用的是同步调用
2. 将字典键判断改为直接访问对象属性，并增加is not None的判断，避免拼接空值引发错误
3. 保持了流式输出的实时打印效果，每收到一段结构化片段就更新并打印当前累积的内容

内容的提问来源于stack exchange，提问作者minglyu