前言

大家好，如果你剛接觸LangChain v1.0框架，這篇文章將帶你快速上手三個核心概念：模型消息格式、流式響應和結構化輸出。這些功能是構建聊天機器人、代理或複雜AI應用的基礎。文章基於官方文檔，結合完整樣例代碼，力求通俗易懂。假設你已安裝LangChain（pip install langchain）和OpenAI（pip install langchain-openai），並設置了API密鑰。我們將使用OpenAI模型作為示例，但這些API在其他模型（如Anthropic、Gemini）上同樣適用。

為了讓初學者更容易上手，在文章末尾，給出一個綜合應用樣例：搭建流式響應的多輪問答機器人，它結合了消息格式和流式響應，模擬真實聊天場景。

1.LangChain 1.0 的基礎模型消息格式

在LangChain 1.0中，消息（Message）是模型輸入和輸出的核心構建塊。它提供了一個跨模型統一的Message標準，無論你使用OpenAI、Anthropic、Gemini還是本地模型，都能保持一致的行為。這種統一抽象帶來了三大好處：

好處

解釋

兼容性強

不同模型的消息格式自動對齊，無需編寫提供商特定的代碼。

可擴展性高

支持多模態內容（如文本、圖像、音頻）、工具調用和自定義元數據，便於擴展新數據類型。

可追蹤性好

通過元數據（如ID、token使用量、響應信息）為LangSmith等調試工具提供一致的上下文結構。

每個消息包含角色（role）（如system、user、assistant、tool）、內容（content）（文本或其他）和可選的元數據（metadata）。核心消息類型包括：

  • SystemMessage：設置助手的角色或行為。
  • HumanMessage：用户輸入。
  • AIMessage：模型輸出。

樣例代碼：構建消息並調用模型

下面是一個簡單示例：創建一個系統消息定義助手角色，然後添加用户消息，調用模型生成響應。

from langchain_openai import ChatOpenAI  # 導入OpenAI聊天模型from langchain_core.messages import SystemMessage, HumanMessage, AIMessage# 初始化模型（替換為你的API密鑰）model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)# 系統消息：定義助手角色system_msg = SystemMessage(content="你叫小智，是一名助人為樂的助手。")# 用户消息：用户輸入human_msg = HumanMessage(content="你好，好久不見，請介紹下你自己。")# 消息列表：按順序傳遞給模型messages = [system_msg, human_msg]# 調用模型response = model.invoke(messages)print(response)  # 輸出：AIMessage(content='你好！我是小智，一名樂於助人的AI助手...', ...)

響應結果解釋

模型的invoke()方法返回一個AIMessage對象，包含生成的文本和其他元數據。以下表格解釋了其關鍵屬性（基於官方文檔）：

屬性

類型

描述

content

str 或 list[dict]

原始內容負載（字符串或提供商原生塊）。

text

str

人類可讀的文本內容（簡化版content）。

content_blocks

list[ContentBlock]

內容塊的標準化視圖（如推理塊、圖像）。

tool_calls

list[dict] 或 None

模型調用的工具（名稱、參數、ID）。

id

str

唯一標識符（自動生成或來自提供商）。

usage_metadata

dict 或 None

Token使用量（如input_tokens、output_tokens）。

response_metadata

ResponseMetadata 或 None

提供商特定響應數據（如完成原因）。

運行上述代碼後，你會看到類似輸出：

AIMessage(content='你好！我是小智，一名樂於助人的AI助手。我可以幫你解答問題、生成代碼或聊天。有什麼我能幫忙的嗎？', id='run-abc123', usage_metadata={'input_tokens': 25, 'output_tokens': 50, ...})

這讓調試和追蹤變得簡單——例如，通過response.usage_metadata監控token消耗。

常見問題排查

  • **錯誤：No API key provided**：確保設置了OPENAI_API_KEY環境變量（export OPENAI_API_KEY=your_key）。
  • 消息順序混亂：始終按[system, human, ai, …]順序構建列表，模型會記住歷史。
  • 多模態擴展：想加圖像？用HumanMessage(content=[{"type": "text", "text": "描述"}, {"type": "image_url", "image_url": {"url": "img.jpg"}}])。

2.模型的流式響應

流式響應（Streaming）是提升用户體驗的關鍵：在LangChain 1.0中，通過stream()方法，模型不會一次性返回完整結果，而是返回一個迭代器（iterator）。每次迭代生成一個AIMessageChunk，包含部分內容（如單個token或文本塊）。這允許實時打印輸出，減少感知延遲，非常適合聊天界面。

啓用流式需要設置streaming=True（在模型初始化時）。迭代器會逐步yield chunk，你可以處理每個chunk的content屬性。

樣例代碼：實時打印流式輸出

以下示例使用簡單用户輸入，逐步打印模型生成的token（以“|”分隔，模擬實時流）。

from langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessage# 初始化模型，啓用流式model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0, streaming=True)# 用户消息messages = [HumanMessage(content="你好，好久不見。講個關於AI的笑話吧。")]# 流式調用：返回迭代器for chunk in model.stream(messages):    if chunk.content:  # 確保chunk有內容        print(chunk.content, end="|", flush=True)  # 實時打印，無換行print("\n")  # 結束時換行

運行結果示例

運行後，你會看到類似實時輸出：

為什麼AI不怕冷？因為它有許多層！|哈哈，是不是很智能？|

每個chunk是一個AIMessageChunk對象，chunk.content是增量文本。flush=True確保立即刷新到控制枱，而end="|"添加分隔符模擬流式效果。

提示：如果使用stream_mode="messages"（高級用法），chunk將代表完整消息塊。但對於初學者，上述簡單循環就夠用。流式特別適合Web應用，能讓用户看到“打字機”效果。

常見問題排查

  • 無輸出或卡住：檢查streaming=True是否設置；如果chunk為空，可能是模型不支持或網絡問題。
  • 性能瓶頸：流式會增加少量開銷；生產中，用asyncio異步處理以支持併發。
  • 歷史消息：流式也支持消息列表，確保傳入完整歷史以保持上下文。

3.模型的結構化輸出方法

結構化輸出讓模型返回可預測的格式（如JSON對象或Pydantic模型），避免解析自然語言的麻煩。在LangChain 1.0中，支持多種Schema：

  • Pydantic：最豐富，支持字段校驗、描述、默認值和嵌套結構。生產首選。
  • TypedDict：輕量類型約束，適合簡單場景。
  • JSON Schema：通用，與前後端/跨語言接口兼容。

這些通過with_structured_output()方法綁定到模型，確保輸出符合Schema。Pydantic提供驗證（如類型檢查、範圍約束），如果輸出無效會自動重試。

樣例代碼：使用Pydantic生成電影信息

以下示例定義一個電影Schema，綁定到模型，然後輸入提示，輸出結構化對象。

from pydantic import BaseModel, Fieldfrom langchain_openai import ChatOpenAI# 定義Pydantic模型：電影詳情class Movie(BaseModel):    """A movie with details."""    title: str = Field(..., description="The title of the movie")    year: int = Field(..., description="The year the movie was released")    director: str = Field(..., description="The director of the movie")    rating: float = Field(..., description="The movie's rating out of 10")# 初始化模型model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)# 綁定結構化輸出：返回Movie實例model_with_structure = model.with_structured_output(Movie)# 輸入提示response = model_with_structure.invoke("Provide details about the movie Inception")print(response)  # 輸出：Movie(title='Inception', year=2010, director='Christopher Nolan', rating=8.8)

解釋與好處

  • 綁定過程：with_structured_output(Movie)創建了一個新模型變體，內部使用工具調用或提供商原生功能強制Schema。
  • 輸出：response直接是Movie實例（Pydantic對象），便於訪問屬性如response.title。無需手動解析JSON！
  • 好處：減少錯誤（自動校驗）、提高可靠性（模型“知道”輸出格式）、易集成（直接序列化為JSON）。

擴展提示：對於TypedDict，使用typing.TypedDict定義；對於JSON Schema，直接傳dict。生產中，結合錯誤處理（如method_kwargs={"handle_errors": True}）以重試無效輸出。

常見問題排查

  • 校驗失敗：模型可能生成不匹配數據；用include_raw=True保留原始輸出調試。
  • 嵌套結構：Pydantic支持類內嵌套類，如actors: List[Actor]（需導入List）。
  • 兼容性：非OpenAI模型可能需額外配置（如mode="tool_calling"）。

4.綜合應用：搭建流式響應的多輪問答機器人

現在，讓我們將前兩部分知識結合：使用消息格式維護對話歷史，並啓用流式響應構建一個交互式聊天機器人。這個機器人支持多輪對話、實時輸出，並自動限制歷史長度以防token超限。適合初學者快速原型化聊天應用。

優化説明

  • 模型選擇：採用OpenAI（更通用），但你可輕鬆替換為其他模型。
  • 改進點：添加了異常處理、消息長度限制（最近50條，避免token爆炸）、退出機制優化。使用messages列表維護狀態，支持上下文記憶。
  • 運行方式：在終端運行，輸入exit退出。流式輸出模擬“打字機”效果。

完整樣例代碼

from langchain_openai import ChatOpenAI  # 使用OpenAI作為示例（可替換為其他模型）from langchain_core.messages import SystemMessage, HumanMessage, AIMessage# 1️⃣ 初始化模型（LangChain 1.0 接口，啓用流式）model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7, streaming=True)# 2️⃣ 初始化系統提示詞（System Prompt）system_message = SystemMessage(    content="你叫小智，是一名樂於助人的智能助手。請在對話中保持温和、有耐心的語氣，並記住之前的上下文。")# 3️⃣ 初始化消息歷史messages = [system_message]print("🔹 歡迎使用小智聊天機器人！輸入 'exit' 退出對話\n")# 4️⃣ 主循環（支持多輪對話 + 流式輸出）while True:    try:        user_input = input("👤 你：").strip()        if user_input.lower() in {"exit", "quit", ""}:            print("🧩 對話結束，再見！")            break        # 追加用户消息到歷史        messages.append(HumanMessage(content=user_input))        # 實時輸出模型生成內容        print("🤖 小智：", end="", flush=True)        full_reply = ""  # 收集完整回覆，用於添加到歷史        # ✅ LangChain 1.0 標準寫法：流式輸出（傳入完整消息歷史）        for chunk in model.stream(messages):            if chunk.content:                print(chunk.content, end="", flush=True)                full_reply += chunk.content        print("\n" + "-" * 50)  # 分隔線，提升可讀性        # 追加 AI 回覆消息到歷史        messages.append(AIMessage(content=full_reply))        # 保持消息長度（只保留最近50輪，避免token超限）        if len(messages) > 50:            messages = [messages[0]] + messages[-49:]  # 保留系統消息 + 最近49條    except KeyboardInterrupt:        print("\n🧩 對話被中斷，再見！")        break    except Exception as e:        print(f"\n❌ 發生錯誤：{e}。請重試。")        continue

運行示例

啓動後：

👤 你：你好，小智！今天天氣怎麼樣？🤖 小智：你好！作為AI，我無法直接查看實時天氣，但如果你告訴我你的位置，我可以幫你查詢或建議。有什麼具體城市嗎？--------------------------------------------------👤 你：在北京。🤖 小智：北京今天晴朗，温度約15-20°C。記得帶件外套哦！有什麼其他想聊的？--------------------------------------------------👤 你：exit🧩 對話結束，再見！

機器人記住上下文（如位置），流式輸出讓響應更自然。

擴展建議：集成LangSmith追蹤（pip install langsmith），或用Streamlit/Flask轉為Web app。結合結構化輸出，可讓機器人返回JSON回覆（如天氣數據）。

結語

通過這四部分，你已從基礎消息到高級應用，全面掌握LangChain 1.0的核心交互方式：統一消息構建對話、流式提升響應性、結構化確保數據質量、多輪機器人實現端到端聊天。