AI Agent白手起家26: 使用标准事件驱动大模型实践

纲要

• 练习目标：掌握大模型标准事件的调用
• 回顾 LangChain 中的核心标准事件
  • invoke
  • stream
  • batch
  • astream_events
  • with_structured_output
• 环境准备
• 实战代码：多种事件调用对比
  • 同步调用与流式输出
  • 批量处理
  • 异步事件流监听
  • 结构化输出
• 运行说明与预期结果
• 总结与扩展建议

练习目标

在学习完大模型组件的理论知识后，动手实践是巩固的最佳途径。本次练习的目标是：使用任意已申请到的大模型资源（无论来自 OpenAI、DeepSeek、Anthropic 等云端 API，还是通过 Ollama 运行的本地模型），通过 LangChain 的标准事件接口来驱动模型，实际感受不同调用方式的行为差异。

你将学会如何优雅地切换同步调用、流式输出、批量处理、异步事件流以及结构化输出，为后续开发 AI Agent 打下坚实基础。

核心标准事件回顾

LangChain 对大模型组件抽象了统一的调用接口，下表对比了最常用的五种事件：

下面的流程图展示了从调用到输出的决策路径：

环境准备

本示例使用 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# ---------- 初始化模型 ----------llm=ChatOpenAI(model="gpt-3.5-turbo",temperature=0.7,api_key=os.getenv("OPENAI_API_KEY"),)# ---------- 1. invoke：同步调用 ----------print("=== 1. invoke 同步调用 ===")response=llm.invoke("用一句话介绍 Python 语言")print("完整回答:",response.content)print("Token 用量:",response.usage_metadata)print()# ---------- 2. stream：流式输出 ----------print("=== 2. stream 流式输出 ===")print("模型输出: ",end="",flush=True)forchunkinllm.stream("写一首关于编程的七言绝句"):print(chunk.content,end="",flush=True)print("\n")# ---------- 3. batch：批量处理 ----------print("=== 3. batch 批量处理 ===")questions=["什么是上下文窗口？","Token 是什么？","工具调用有什么作用？"]results=llm.batch(questions)forq,rinzip(questions,results):print(f"Q:{q}\nA:{r.content}\n")# ---------- 4. astream_events：异步事件流 ----------asyncdefdemo_astream_events():print("=== 4. astream_events 异步事件流 ===")asyncforeventinllm.astream_events("解释什么是深度学习",version="v2"):kind=event["event"]ifkind=="on_chat_model_start":print("[模型开始响应]")elifkind=="on_chat_model_stream":data=event["data"]["chunk"]ifdata.content:print(data.content,end="",flush=True)elifkind=="on_chat_model_end":print("\n[模型响应结束]")output=event["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:str=Field(description="笑话的铺垫")punchline:str=Field(description="笑话的包袱/笑点")rating:int|None=Field(default=None,description="好笑程度 1-10")structured_llm=llm.with_structured_output(ProgrammingJoke)joke=structured_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 交互。

动手练习是内化知识的最佳方式，尝试修改代码、切换模型、观察输出差异，你会对大模型的标准事件有更深刻的理解。