深入 LangChain：掌握结构化输出与流式传输两大利器----《Hello LangChain LangGraph!》(4)

文章目录结构化输出结构化输出流式传输自定义流式输出解析器结构化输出结构化输出聊天模型的with_structured_output方法允许我们预先定义一个期望的数据结构并要求大模型必须按照这个结构返回信息。–LangChain输出解析器则是依赖模型的文本去生成所以有时模型会输出格式错误的内容因为优先使用上面这种模型原生结构化输出的方法该输出结构可以指定为 TypedDict 类、JSON Schema 或 Pydantic 类。如果使用 TypedDict 或 JSON Schema则 Runnable 将返回一个字典 eg:{name: 张三, age: 25}如果使用 Pydantic 类则将返回一个 Pydantic 对象 eg:name张三 age25modelChatOpenAI(modelgpt-4o-mini)# 这里列举三种写法#第一种 Pydantic 对象classJoke(BaseModel):给用户讲的一个笑话setup:strField(description这个笑话的开头)punchline:strField(description这个笑话的妙语)rating:Optional[int]Field(defaultNone,description从1-10分给这个笑话评分)classData(BaseModel):获取关于笑话的数据列表jokes:List[Joke]# 第二种 TypedDictclassJoke(TypedDict):给用户讲的一个笑话setup:Annotated[str,...,这个笑话的开头]punchline:Annotated[str,...,这个笑话的妙语]rating:Annotated[Optional[int],None,从1-10分给这个笑话评分]model_with_structuredmodel.with_structured_output(Joke,include_rawTrue)# 加上inclue_raw True这个参数的话会返回一个更大的字典(含parsed和raw)print(model_with_structured.invoke(讲一个关于跳舞的笑话))#第三种 JSON Schemajson_schema{title:joke,description:给用户讲一个笑话。,type:object,properties:{setup:{type:string,description:这个笑话的开头,},punchline:{type:string,description:这个笑话的妙语,},rating:{type:integer,description:从1到10分给这个笑话评分,default:None,},},required:[setup,punchline],}model_with_structuredmodel.with_structured_output(json_schema)#绑定schemaprint(model_with_structured.invoke(你是谁))# 执行关于输出格式的选择:用户每次的输入是不一样的最好是定义好几个不同的格式然后用Union把他们打包在一起交给大模型这样大模型就可以从中自主选择格式了举例:modelChatOpenAI(modelgpt-4o-mini)classJoke(BaseModel):给用户讲的一个笑话setup:strField(description这个笑话的开头)punchline:strField(description这个笑话的妙语)rating:Optional[int]Field(defaultNone,description从1-10分给这个笑话评分)classResponse(BaseModel):用以对话的方式回应。content:strField(description用于对用户查询的会话响应)classFinalResponse(BaseModel):最终回复选择合适的输出结构final_output:Union[Joke,Response]model_with_structuredmodel.with_structured_output(FinalResponse)print(model_with_structured.invoke(讲一个关于跳舞的笑话))print(model_with_structured.invoke(你是谁))结构化输出的使用场景:1.作为信息提取器 2.使用少样本提示来增强信息提取能力流式传输之前使用invoke的调用方式属于是非流式传输–就是聊天模型直接返回全量内容LLM支持SSE协议,这是一种基于HTTP的单项通信机制。客户端建立连接之后服务器可以源源不断的向客户端推送数据(且不需要客户端反复轮询)–LangChain的流式传输也是用的这个LangChain流式传输的完整流程和底层协议:angchain-openai 包通过集成 OpenAI Python SDK提供了一个 HTTP 客户端。因此支持 LangChain 向 OpenAI 的 API 发起调用请求。若希望发起流式传输请求则需在请求中加入streamTrue向 OpenAI 说明以 SSE 协议进行流式返回。LangChain 接收 OpenAI 的 SSE 格式的响应并将其转换为 LangChain 自封装的消息格式如AIMessageChunk消息。这样就可以以统一的方式处理来自不同模型提供商OpenAI, Anthropic 等的流式响应。stream()同步传输:在 LangChain 聊天模型中可以使用其.stream()方法来同步生成流式响应的效果聊天模型的.stream()方法返回一个迭代器该迭代器在生成输出时同步产生输出消息块。可以使用for循环实时处理每个块chunks[]forchunkinmodel.stream(写一段关于春天的作文1000字):chunks.append(chunk)# chunk是AIMessageChunk类型的,是AIMessage的一小块内容(也就是消息块)print(chunk.content,end|,flushTrue)astream异步传输:–采用这种方法的话可以在程序等待下一个chunk的过程中让服务器去处理其他哟弄个胡的请求等数据包到了再继续打印# 异步流式输出asyncdefasync_stream():#成协程函数了print(异步调用)asyncforchunkinmodel.astream(写一段关于春天的作文1000字):print(chunk.content,end|,flushTrue)asyncio.run(async_stream())# model.ainvoke() 这个是拿完整结果关于协程、线程、进程:协程是一个线程内实现的–比线程切换的消耗小协程是协作式调度同一时间一个线程里只有一个协程在运行举例:# 协程asyncdefboil_water_async():print(开始烧水...)awaitasyncio.sleep(5)# await表示等待这个操作完成但期间可以做别的事print(烧水完成...)# 协程asyncdefsend_message_async():print(开始发消息...)awaitasyncio.sleep(2)# 模拟烧水2sprint(发消息完成...)# 协程调度# 事件循环asyncdefmain():# 1、烧水任务task1asyncio.create_task(boil_water_async())# 2、发消息任务task2asyncio.create_task(send_message_async())awaittask1awaittask2# run 会创建一个事件循环asyncio.run(main())自定义流式输出解析器如果想要把输出改成其他格式的输出(比如一句话一句话的输出)并且保留流式处理功能就需要在链中使用生成器函数就能完成自定义流式输出的功能# 组件1聊天模型modelChatOpenAI(modelgpt-4o-mini)# 组件2输出解析器strparserStrOutputParser()# 自定义生成器defsplit_into_list(input:Iterator[str])-Iterator[List[str]]:bufferforchunkininput:bufferchunk# 遇到 。 需要刷新while。inbuffer:# 找到 。的位置stop_indexbuffer.index(。)# yield 用于创造生成器yield[buffer[:stop_index].strip()]bufferbuffer[stop_index1:]# 处理buffer最后几个字yield[buffer.strip()]# 定义链chainmodel|parser|split_into_listforchunkinchain.stream(写一段关于爱情的歌词需要5句话每句话用中文句号隔开。):# 使用 parser结果就是 strprint(chunk,end|,flushTrue)