纲要练习目标掌握大模型标准事件的调用回顾 LangChain 中的核心标准事件invokestreambatchastream_eventswith_structured_output环境准备实战代码多种事件调用对比同步调用与流式输出批量处理异步事件流监听结构化输出运行说明与预期结果总结与扩展建议练习目标在学习完大模型组件的理论知识后动手实践是巩固的最佳途径。本次练习的目标是使用任意已申请到的大模型资源无论来自 OpenAI、DeepSeek、Anthropic 等云端 API还是通过 Ollama 运行的本地模型通过 LangChain 的标准事件接口来驱动模型实际感受不同调用方式的行为差异。你将学会如何优雅地切换同步调用、流式输出、批量处理、异步事件流以及结构化输出为后续开发 AI Agent 打下坚实基础。核心标准事件回顾LangChain 对大模型组件抽象了统一的调用接口下表对比了最常用的五种事件事件调用方式适用场景返回类型invoke同步阻塞返回完整结果后台任务、批处理脚本AIMessagestream同步迭代逐 token 产出聊天机器人前端、实时反馈Iterator[chunk]batch并发发送多个请求批量评估、数据增强list[AIMessage]astream_events异步流式按事件类型精细控制需要监听开始/结束/流式进度AsyncIterator[event]with_structured_output基于 Pydantic 模型的结构化输出下游系统对接、JSON 解析Pydantic 实例下面的流程图展示了从调用到输出的决策路径否否是是否是开始调用需要流式?需要结构化输出?使用 invoke使用 with_structured_output invoke需要异步事件?使用 stream使用 astream_events返回完整 AIMessage逐 chunk 输出按事件回调环境准备本示例使用 OpenAI 的 GPT-3.5-turbo 模型成本较低适合练习。请先安装依赖并设置 API 密钥。pipinstalllangchain-openai pydantic确保环境变量OPENAI_API_KEY已设置为你的有效密钥。若使用本地模型如 Ollama可将ChatOpenAI替换为ChatOllama并修改base_url即可。实战代码一次演示所有事件以下代码将连续演示invoke、stream、batch、astream_events和with_structured_output五种调用方式。importos,asynciofrompydanticimportBaseModel,Fieldfromlangchain_openaiimportChatOpenAI# ---------- 初始化模型 ----------llmChatOpenAI(modelgpt-3.5-turbo,temperature0.7,api_keyos.getenv(OPENAI_API_KEY),)# ---------- 1. invoke同步调用 ----------print( 1. invoke 同步调用 )responsellm.invoke(用一句话介绍 Python 语言)print(完整回答:,response.content)print(Token 用量:,response.usage_metadata)print()# ---------- 2. stream流式输出 ----------print( 2. stream 流式输出 )print(模型输出: ,end,flushTrue)forchunkinllm.stream(写一首关于编程的七言绝句):print(chunk.content,end,flushTrue)print(\n)# ---------- 3. batch批量处理 ----------print( 3. batch 批量处理 )questions[什么是上下文窗口,Token 是什么,工具调用有什么作用]resultsllm.batch(questions)forq,rinzip(questions,results):print(fQ:{q}\nA:{r.content}\n)# ---------- 4. astream_events异步事件流 ----------asyncdefdemo_astream_events():print( 4. astream_events 异步事件流 )asyncforeventinllm.astream_events(解释什么是深度学习,versionv2):kindevent[event]ifkindon_chat_model_start:print([模型开始响应])elifkindon_chat_model_stream:dataevent[data][chunk]ifdata.content:print(data.content,end,flushTrue)elifkindon_chat_model_end:print(\n[模型响应结束])outputevent[data][output]ifoutput.usage_metadata:print(Token 用量:,output.usage_metadata)asyncio.run(demo_astream_events())print()# ---------- 5. with_structured_output结构化输出 ----------print( 5. with_structured_output 结构化输出 )classProgrammingJoke(BaseModel):程序员笑话setup:strField(description笑话的铺垫)punchline:strField(description笑话的包袱/笑点)rating:int|NoneField(defaultNone,description好笑程度 1-10)structured_llmllm.with_structured_output(ProgrammingJoke)jokestructured_llm.invoke(给我讲一个编程相关的笑话)print(f铺垫:{joke.setup})print(f笑点:{joke.punchline})print(f评分:{joke.rating})将以上代码保存为practice.py运行后你将直观看到每种事件的特点invoke等待完整回复后一次性打印。stream逐字出现模拟打字机效果。batch同时发送三个问题按列表顺序返回答案。astream_events在流式输出的前后触发了开始和结束事件并报告了 Token 用量。with_structured_output返回了一个可直接读属性的 Pydantic 对象无需手动解析 JSON。扩展建议尝试替换模型将ChatOpenAI改为ChatAnthropic需要langchain-anthropic或ChatOllama看看事件支持是否一致。为batch处理加入进度条使用tqdm提升批处理体验。在astream_events中捕获更多事件类型如on_chat_model_stream的详细元数据实现自定义日志记录。结合之前学习的工具调用bind_tools在stream模式下处理工具请求构建更真实的 Agent 交互。动手练习是内化知识的最佳方式尝试修改代码、切换模型、观察输出差异你会对大模型的标准事件有更深刻的理解。
