綱要練習目標掌握大模型標準事件的調(diào)用回顧 LangChain 中的核心標準事件invokestreambatchastream_eventswith_structured_output環(huán)境準備實戰(zhàn)代碼多種事件調(diào)用對比同步調(diào)用與流式輸出批量處理異步事件流監(jiān)聽結(jié)構(gòu)化輸出運行說明與預(yù)期結(jié)果總結(jié)與擴展建議練習目標在學(xué)習完大模型組件的理論知識后動手實踐是鞏固的最佳途徑。本次練習的目標是使用任意已申請到的大模型資源無論來自 OpenAI、DeepSeek、Anthropic 等云端 API還是通過 Ollama 運行的本地模型通過 LangChain 的標準事件接口來驅(qū)動模型實際感受不同調(diào)用方式的行為差異。你將學(xué)會如何優(yōu)雅地切換同步調(diào)用、流式輸出、批量處理、異步事件流以及結(jié)構(gòu)化輸出為后續(xù)開發(fā) AI Agent 打下堅實基礎(chǔ)。核心標準事件回顧LangChain 對大模型組件抽象了統(tǒng)一的調(diào)用接口下表對比了最常用的五種事件事件調(diào)用方式適用場景返回類型invoke同步阻塞返回完整結(jié)果后臺任務(wù)、批處理腳本AIMessagestream同步迭代逐 token 產(chǎn)出聊天機器人前端、實時反饋Iterator[chunk]batch并發(fā)發(fā)送多個請求批量評估、數(shù)據(jù)增強list[AIMessage]astream_events異步流式按事件類型精細控制需要監(jiān)聽開始/結(jié)束/流式進度AsyncIterator[event]with_structured_output基于 Pydantic 模型的結(jié)構(gòu)化輸出下游系統(tǒng)對接、JSON 解析Pydantic 實例下面的流程圖展示了從調(diào)用到輸出的決策路徑否否是是否是開始調(diào)用需要流式?需要結(jié)構(gòu)化輸出?使用 invoke使用 with_structured_output invoke需要異步事件?使用 stream使用 astream_events返回完整 AIMessage逐 chunk 輸出按事件回調(diào)環(huán)境準備本示例使用 OpenAI 的 GPT-3.5-turbo 模型成本較低適合練習。請先安裝依賴并設(shè)置 API 密鑰。pipinstalllangchain-openai pydantic確保環(huán)境變量OPENAI_API_KEY已設(shè)置為你的有效密鑰。若使用本地模型如 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(寫一首關(guān)于編程的七言絕句):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(解釋什么是深度學(xué)習,versionv2):kindevent[event]ifkindon_chat_model_start:print([模型開始響應(yīng)])elifkindon_chat_model_stream:dataevent[data][chunk]ifdata.content:print(data.content,end,flushTrue)elifkindon_chat_model_end:print(\n[模型響應(yīng)結(jié)束])outputevent[data][output]ifoutput.usage_metadata:print(Token 用量:,output.usage_metadata)asyncio.run(demo_astream_events())print()# ---------- 5. with_structured_output結(jié)構(gòu)化輸出 ----------print( 5. with_structured_output 結(jié)構(gòu)化輸出 )classProgrammingJoke(BaseModel):程序員笑話setup:strField(description笑話的鋪墊)punchline:strField(description笑話的包袱/笑點)rating:int|NoneField(defaultNone,description好笑程度 1-10)structured_llmllm.with_structured_output(ProgrammingJoke)jokestructured_llm.invoke(給我講一個編程相關(guān)的笑話)print(f鋪墊:{joke.setup})print(f笑點:{joke.punchline})print(f評分:{joke.rating})將以上代碼保存為practice.py運行后你將直觀看到每種事件的特點invoke等待完整回復(fù)后一次性打印。stream逐字出現(xiàn)模擬打字機效果。batch同時發(fā)送三個問題按列表順序返回答案。astream_events在流式輸出的前后觸發(fā)了開始和結(jié)束事件并報告了 Token 用量。with_structured_output返回了一個可直接讀屬性的 Pydantic 對象無需手動解析 JSON。擴展建議嘗試替換模型將ChatOpenAI改為ChatAnthropic需要langchain-anthropic或ChatOllama看看事件支持是否一致。為batch處理加入進度條使用tqdm提升批處理體驗。在astream_events中捕獲更多事件類型如on_chat_model_stream的詳細元數(shù)據(jù)實現(xiàn)自定義日志記錄。結(jié)合之前學(xué)習的工具調(diào)用bind_tools在stream模式下處理工具請求構(gòu)建更真實的 Agent 交互。動手練習是內(nèi)化知識的最佳方式嘗試修改代碼、切換模型、觀察輸出差異你會對大模型的標準事件有更深刻的理解。