綱要練習目標掌握大模型標準事件的調(diào)用回顧 LangChain 中的核心標準事件invokestreambatchastream_eventswith_structured_output環(huán)境準備實戰(zhàn)代碼多種事件調(diào)用對比同步調(diào)用與流式輸出批量處理異步事件流監(jiān)聽結構化輸出運行說明與預期結果總結與擴展建議練習目標在學習完大模型組件的理論知識后動手實踐是鞏固的最佳途徑。本次練習的目標是使用任意已申請到的大模型資源無論來自 OpenAI、DeepSeek、Anthropic 等云端 API還是通過 Ollama 運行的本地模型通過 LangChain 的標準事件接口來驅動模型實際感受不同調(diào)用方式的行為差異。你將學會如何優(yōu)雅地切換同步調(diào)用、流式輸出、批量處理、異步事件流以及結構化輸出為后續(xù)開發(fā) AI Agent 打下堅實基礎。核心標準事件回顧LangChain 對大模型組件抽象了統(tǒng)一的調(diào)用接口下表對比了最常用的五種事件事件調(diào)用方式適用場景返回類型invoke同步阻塞返回完整結果后臺任務、批處理腳本AIMessagestream同步迭代逐 token 產(chǎn)出聊天機器人前端、實時反饋Iterator[chunk]batch并發(fā)發(fā)送多個請求批量評估、數(shù)據(jù)增強list[AIMessage]astream_events異步流式按事件類型精細控制需要監(jiān)聽開始/結束/流式進度AsyncIterator[event]with_structured_output基于 Pydantic 模型的結構化輸出下游系統(tǒng)對接、JSON 解析Pydantic 實例下面的流程圖展示了從調(diào)用到輸出的決策路徑否否是是否是開始調(diào)用需要流式?需要結構化輸出?使用 invoke使用 with_structured_output invoke需要異步事件?使用 stream使用 astream_events返回完整 AIMessage逐 chunk 輸出按事件回調(diào)環(huán)境準備本示例使用 OpenAI 的 GPT-3.5-turbo 模型成本較低適合練習。請先安裝依賴并設置 API 密鑰。pipinstalllangchain-openai pydantic確保環(huán)境變量OPENAI_API_KEY已設置為你的有效密鑰。若使用本地模型如 Ollama可將ChatOpenAI替換為ChatOllama并修改base_url即可。實戰(zhàn)代碼一次演示所有事件以下代碼將連續(xù)演示invoke、stream、batch、astream_events和with_structured_output五種調(diào)用方式。importos,asynciofrompydanticimportBaseModel,Fieldfromlangchain_openaiimportChatOpenAI# ---------- 初始化模型 ----------llmChatOpenAI(modelgpt-3.5-turbo,temperature0.7,api_keyos.getenv(OPENAI_API_KEY),)# ---------- 1. invoke同步調(diào)用 ----------print( 1. invoke 同步調(diào)用 )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 是什么,工具調(diào)用有什么作用]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逐字出現(xiàn)模擬打字機效果。batch同時發(fā)送三個問題按列表順序返回答案。astream_events在流式輸出的前后觸發(fā)了開始和結束事件并報告了 Token 用量。with_structured_output返回了一個可直接讀屬性的 Pydantic 對象無需手動解析 JSON。擴展建議嘗試替換模型將ChatOpenAI改為ChatAnthropic需要langchain-anthropic或ChatOllama看看事件支持是否一致。為batch處理加入進度條使用tqdm提升批處理體驗。在astream_events中捕獲更多事件類型如on_chat_model_stream的詳細元數(shù)據(jù)實現(xiàn)自定義日志記錄。結合之前學習的工具調(diào)用bind_tools在stream模式下處理工具請求構建更真實的 Agent 交互。動手練習是內(nèi)化知識的最佳方式嘗試修改代碼、切換模型、觀察輸出差異你會對大模型的標準事件有更深刻的理解。