多模態輸入
- 圖片處理(最高頻)
1.1 URL形式(推薦大文件)
from langchain.schema import HumanMessage
from langchain.chat_models import ChatOpenAIchat = ChatOpenAI(model="gpt-4-vision-preview")message = HumanMessage(content=[{"type": "text", "text": "描述這張圖片"},{"type": "image_url","image_url": {"url": "https://example.com/image.jpg","detail": "auto" # 可選參數:low/high(控制token消耗)}}]
)
response = chat([message])
1.2 Base64形式(推薦小文件)
import base64
from PIL import Image
import io# 圖像預處理
def process_image(image_path, max_size=512):img = Image.open(image_path)img.thumbnail((max_size, max_size))buffered = io.BytesIO()img.save(buffered, format="JPEG")return base64.b64encode(buffered.getvalue()).decode('utf-8')base64_image = process_image("local_image.jpg")message = HumanMessage(content=[{"type": "text", "text": "分析這張醫學影像"},{"type": "image_url","image_url": {"url": f"data:image/jpeg;base64,{base64_image}","detail": "high"}}]
)
- 音頻處理
2.1 URL形式(推薦長音頻)
audio_message = HumanMessage(content=[{"type": "text", "text": "總結這段會議錄音"},{"type": "audio_url","audio_url": {"url": "https://example.com/meeting.mp3","transcription_format": "srt" # 可選字幕格式}}]
)
2.2 Base64形式(推薦短音頻)
with open("audio.mp3", "rb") as audio_file:base64_audio = base64.b64encode(audio_file.read()).decode('utf-8')audio_message = HumanMessage(content=[{"type": "text", "text": "轉寫這段語音"},{"type": "audio_url","audio_url": {"url": f"data:audio/mp3;base64,{base64_audio}","duration": 30 # 單位:秒(幫助模型優化處理)}}]
)
- 視頻處理(需分幀處理)
3.1 關鍵幀提取方案
from moviepy.editor import VideoFileClip
import numpy as npdef extract_key_frames(video_path, interval=10):clip = VideoFileClip(video_path)frames = []for t in np.arange(0, clip.duration, interval):frame = clip.get_frame(t)# 轉換為base64(參考圖片處理部分)return frameskey_frames = extract_key_frames("video.mp4")video_message = HumanMessage(content=[{"type": "text", "text": "分析這段教學視頻"},*[{"type": "image_url", "image_url": frame} for frame in key_frames],{"type": "text","text": f"視頻總時長:{clip.duration}秒,采樣間隔:{interval}秒"}]
)
- 多圖處理策略
multi_image_message = HumanMessage(content=[{"type": "text", "text": "比較這三款產品的設計差異"},*[{"type": "image_url","image_url": {"url": url,"detail": "low" # 多圖時建議使用low模式}} for url in ["https://example.com/product1.jpg","data:image/png;base64,...","https://example.com/product3.jpg"]],{"type": "text","text": "從左到右依次是產品A、B、C,請重點分析色彩搭配"}]
)
最佳實踐建議:
- 格式選擇:
- URL更適合:文件>1MB、需要復用、云端存儲
- Base64更適合:文件<1MB、臨時處理、需要加密
- 性能優化:
- 圖片:調整分辨率(推薦512px)、選擇WEBP格式
- 音頻:采樣率控制在16kHz、單聲道
- 視頻:關鍵幀間隔建議10-30秒
- 上下文管理:
# 多模態消息組合示例
combined_message = [HumanMessage(content=[...]), # 用戶問題AIMessage(content=[...]), # 歷史回答HumanMessage(content=[...]) # 新輸入
]
- 錯誤處理:
try:response = chat(messages)
except openai.BadRequestError as e:if "content" in str(e):print("建議:1. 降低圖片分辨率 2. 減少多模態元素數量 3. 簡化文字指令")
注意事項:
- 模型支持驗證:確認使用的模型版本支持多模態輸入
- 成本控制:圖片high detail模式消耗更多token(base image: 85 tokens + 85*detail)
- 隱私保護:敏感信息建議使用臨時URL+訪問控制
完整示例(圖片分析+音頻轉寫):
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage# 初始化多模態模型
multimodal_chat = ChatOpenAI(model="gpt-4-vision-preview",max_tokens=4096
)# 構建多模態消息
message = HumanMessage(content=[{"type": "text","text": "請分析這張設計圖,并根據錄音反饋改進建議"},{"type": "image_url","image_url": {"url": "https://example.com/design.jpg","detail": "high"}},{"type": "audio_url","audio_url": {"url": "data:audio/mp3;base64,...","transcription_format": "text"}}]
)# 獲取響應
response = multimodal_chat([message])
print(response.content)
```<think>
嗯,用戶正在學習LangChain,想了解如何在大模型交互中處理多模態輸入,包括音頻、視頻、圖片,以及URL和Base64兩種形式,還有多張圖片的情況。需要根據實際使用優先級來介紹,并提供最佳實踐代碼。首先,我需要整理多模態輸入的優先級。通常,圖片處理在LangChain中最常見,其次是音頻,視頻處理可能較少,因為大模型對視頻的支持可能有限。不過,還是需要涵蓋所有類型。然后,關于輸入形式,URL和Base64都是常用的。URL更節省資源,適合大文件,而Base64適合小文件或需要內聯數據的場景。需要分別說明這兩種方法。對于多張圖片的情況,可能需要考慮模型是否支持多圖輸入,或者需要將多圖信息合并處理,比如用CLIP模型提取特征再綜合。接下來需要分步驟詳細講解每種模態的處理方式,包括代碼示例。代碼部分需要使用最新的LangChain庫,可能需要結合OpenAI的API或其他多模態模型,如GPT-4V。需要注意用戶可能對如何構造消息、處理不同格式不太熟悉,所以代碼要盡量詳細,并解釋關鍵步驟。同時,要提醒用戶注意費用和模型支持情況,比如GPT-4V是否可用,或者是否需要其他模型如CLIP。可能需要提到使用PIL庫處理圖片,requests庫獲取URL內容,base64編碼轉換,以及如何處理音頻和視頻的轉錄或摘要。對于視頻,可能建議提取關鍵幀或使用Whisper進行音頻轉錄,再結合處理。還要檢查是否有遺漏的部分,比如多張圖片的最佳實踐,是否要分別編碼還是合并,可能需要舉例說明。另外,確保代碼示例正確,特別是base64編碼的部分,可能需要演示如何正確讀取文件并編碼。最后,總結最佳實踐,強調根據模型能力選擇輸入方式,優先使用URL減少傳輸負擔,處理大文件時注意分塊或使用適合的模型。可能還要提醒用戶處理多模態時的上下文管理,比如在LangChain中使用不同的消息類型組合。可能還需要驗證代碼示例是否正確,比如使用OpenAI的ChatCompletion時,是否正確傳遞了multipart/form-data的數據結構,或者是否需要使用特定的LangChain模塊,如ChatOpenAI和HumanMessage、AIMessage等。總之,結構應該是先介紹圖片處理(URL和Base64),然后是音頻,視頻,多張圖片的情況,每個部分都給出代碼示例,并解釋注意事項和最佳實踐。確保回答全面,覆蓋用戶的問題點,同時提供可操作的代碼。</think>以下是關于在LangChain中實現大模型多模態輸入的實踐指南,按實際使用優先級排序:1. 圖片處理(最高頻)
1.1 URL形式(推薦大文件)
```python
from langchain.schema import HumanMessage
from langchain.chat_models import ChatOpenAIchat = ChatOpenAI(model="gpt-4-vision-preview")message = HumanMessage(content=[{"type": "text", "text": "描述這張圖片"},{"type": "image_url","image_url": {"url": "https://example.com/image.jpg","detail": "auto" # 可選參數:low/high(控制token消耗)}}]
)
response = chat([message])
1.2 Base64形式(推薦小文件)
import base64
from PIL import Image
import io# 圖像預處理
def process_image(image_path, max_size=512):img = Image.open(image_path)img.thumbnail((max_size, max_size))buffered = io.BytesIO()img.save(buffered, format="JPEG")return base64.b64encode(buffered.getvalue()).decode('utf-8')base64_image = process_image("local_image.jpg")message = HumanMessage(content=[{"type": "text", "text": "分析這張醫學影像"},{"type": "image_url","image_url": {"url": f"data:image/jpeg;base64,{base64_image}","detail": "high"}}]
)
- 音頻處理
2.1 URL形式(推薦長音頻)
audio_message = HumanMessage(content=[{"type": "text", "text": "總結這段會議錄音"},{"type": "audio_url","audio_url": {"url": "https://example.com/meeting.mp3","transcription_format": "srt" # 可選字幕格式}}]
)
2.2 Base64形式(推薦短音頻)
with open("audio.mp3", "rb") as audio_file:base64_audio = base64.b64encode(audio_file.read()).decode('utf-8')audio_message = HumanMessage(content=[{"type": "text", "text": "轉寫這段語音"},{"type": "audio_url","audio_url": {"url": f"data:audio/mp3;base64,{base64_audio}","duration": 30 # 單位:秒(幫助模型優化處理)}}]
)
- 視頻處理(需分幀處理)
3.1 關鍵幀提取方案
from moviepy.editor import VideoFileClip
import numpy as npdef extract_key_frames(video_path, interval=10):clip = VideoFileClip(video_path)frames = []for t in np.arange(0, clip.duration, interval):frame = clip.get_frame(t)# 轉換為base64(參考圖片處理部分)return frameskey_frames = extract_key_frames("video.mp4")video_message = HumanMessage(content=[{"type": "text", "text": "分析這段教學視頻"},*[{"type": "image_url", "image_url": frame} for frame in key_frames],{"type": "text","text": f"視頻總時長:{clip.duration}秒,采樣間隔:{interval}秒"}]
)
- 多圖處理策略
multi_image_message = HumanMessage(content=[{"type": "text", "text": "比較這三款產品的設計差異"},*[{"type": "image_url","image_url": {"url": url,"detail": "low" # 多圖時建議使用low模式}} for url in ["https://example.com/product1.jpg","data:image/png;base64,...","https://example.com/product3.jpg"]],{"type": "text","text": "從左到右依次是產品A、B、C,請重點分析色彩搭配"}]
)
最佳實踐建議:
- 格式選擇:
- URL更適合:文件>1MB、需要復用、云端存儲
- Base64更適合:文件<1MB、臨時處理、需要加密
- 性能優化:
- 圖片:調整分辨率(推薦512px)、選擇WEBP格式
- 音頻:采樣率控制在16kHz、單聲道
- 視頻:關鍵幀間隔建議10-30秒
- 上下文管理:
# 多模態消息組合示例
combined_message = [HumanMessage(content=[...]), # 用戶問題AIMessage(content=[...]), # 歷史回答HumanMessage(content=[...]) # 新輸入
]
- 錯誤處理:
try:response = chat(messages)
except openai.BadRequestError as e:if "content" in str(e):print("建議:1. 降低圖片分辨率 2. 減少多模態元素數量 3. 簡化文字指令")
注意事項:
- 模型支持驗證:確認使用的模型版本支持多模態輸入
- 成本控制:圖片high detail模式消耗更多token(base image: 85 tokens + 85*detail)
- 隱私保護:敏感信息建議使用臨時URL+訪問控制
完整示例(圖片分析+音頻轉寫):
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage# 初始化多模態模型
multimodal_chat = ChatOpenAI(model="gpt-4-vision-preview",max_tokens=4096
)# 構建多模態消息
message = HumanMessage(content=[{"type": "text","text": "請分析這張設計圖,并根據錄音反饋改進建議"},{"type": "image_url","image_url": {"url": "https://example.com/design.jpg","detail": "high"}},{"type": "audio_url","audio_url": {"url": "data:audio/mp3;base64,...","transcription_format": "text"}}]
)# 獲取響應
response = multimodal_chat([message])
print(response.content)
格式化輸出(Parser)
1. PydanticOutputParser (最高優先級)
適用場景:結構化數據輸出、API交互、數據庫存儲
優勢:類型安全、自動驗證、支持復雜嵌套結構
實現代碼:
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import PromptTemplate
from langchain.chat_models import ChatOpenAI
from pydantic import BaseModel, Field# 定義數據結構模型
class MovieInfo(BaseModel):title: str = Field(description="電影名稱")year: int = Field(description="上映年份", gt=1900)genres: list[str] = Field(description="類型列表")rating: float = Field(description="豆瓣評分", ge=0, le=10)directors: list[str] = Field(description="導演列表")# 創建解析器
parser = PydanticOutputParser(pydantic_object=MovieInfo)# 構建提示詞模板
prompt = PromptTemplate(template="""根據用戶輸入提取電影信息,嚴格使用如下格式:
{format_instructions}
用戶輸入:{query}""",input_variables=["query"],partial_variables={"format_instructions": parser.get_format_instructions()}
)# 完整調用流程
model = ChatOpenAI(temperature=0)
chain = prompt | model | parser# 執行調用
result = chain.invoke({"query": "諾蘭導演的星際穿越2014年上映,科幻題材,評分9.4"
})print(result.dict())
# 輸出:{'title': '星際穿越', 'year': 2014, 'genres': ['科幻'], 'rating': 9.4, 'directors': ['諾蘭']}
高級技巧:
- 字段約束增強:
# 在Pydantic模型中使用Field添加約束
rating: float = Field(..., ge=0, le=10, description="必須保留一位小數")
- 自定義驗證:
from pydantic import validator@validator('genres')
def validate_genres(cls, v):allowed = ['科幻', '動作', '愛情']if not all(g in allowed for g in v):raise ValueError(f"包含非法類型,允許值:{allowed}")return v
2. JsonOutputParser (第二優先級)
適用場景:靈活JSON輸出、快速原型開發
優勢:無需預定義結構、適合動態字段
實現代碼:
from langchain.output_parsers import JsonOutputParser
from langchain.prompts import PromptTemplate
from langchain.chat_models import ChatOpenAI# 自定義JSON Schema提示
json_schema = {"type": "object","properties": {"name": {"type": "string"},"ingredients": {"type": "array", "items": {"type": "string"}},"cooking_time": {"type": "number"}},"required": ["name", "ingredients"]
}parser = JsonOutputParser()prompt = PromptTemplate(template="""將菜譜轉換為JSON格式,遵循以下schema:
{schema}
當前輸入:{input}
只返回JSON,不要額外說明""",input_variables=["input"],partial_variables={"schema": str(json_schema)}
)chain = prompt | ChatOpenAI() | parserresult = chain.invoke({"input": "魚香肉絲需要豬肉、木耳、胡蘿卜,烹飪時間20分鐘"
})print(result)
# 輸出:{'name': '魚香肉絲', 'ingredients': ['豬肉', '木耳', '胡蘿卜'], 'cooking_time': 20.0}
增強方案:
- JSON Schema驗證:
from jsonschema import validate# 結果驗證
schema = {"type": "object","properties": {"name": {"type": "string", "minLength": 2},"steps": {"type": "array", "minItems": 3}}
}try:validate(instance=result, schema=schema)
except Exception as e:print(f"驗證失敗:{e}")
3. XMLOutputParser (第三優先級)
適用場景:企業遺留系統對接、特定標簽格式要求
優勢:標簽結構清晰、支持命名空間
實現代碼:
from langchain.output_parsers import XMLOutputParser
from langchain.prompts import PromptTemplate# 自定義標簽配置
xml_template = """生成產品數據XML,要求:
1. 根標簽為<Products>
2. 每個產品用<Product>包裹
3. 必須包含字段:ID(屬性)、名稱、價格、庫存
輸入:{input}"""parser = XMLOutputParser(tags=["Products", "Product"])prompt = PromptTemplate(template=xml_template + "\n{format_instructions}",input_variables=["input"],partial_variables={"format_instructions": parser.get_format_instructions()}
)chain = prompt | ChatOpenAI() | parserresult = chain.invoke({"input": "蘋果手機庫存50臺,售價6999;華為筆記本庫存30臺,售價5999"
})print(result)
# 輸出:
"""
<Products><Product ID="1"><名稱>蘋果手機</名稱><價格>6999</價格><庫存>50</庫存></Product><Product ID="2"><名稱>華為筆記本</名稱><價格>5999</價格><庫存>30</庫存></Product>
</Products>
"""
高級用法:
- XPath解析:
from lxml import etreexml_data = etree.fromstring(result)
# 提取所有價格
prices = xml_data.xpath("//Product/價格/text()")
print(prices) # 輸出:['6999', '5999']
- 命名空間支持:
xml_template += """
<Products xmlns:ns="http://example.com/ns"><ns:Product ns:id="1">...</ns:Product>
</Products>"""
4. YAMLOutputParser (特殊場景)
適用場景:配置文件生成、DevOps工具鏈集成
實現代碼:
from langchain.output_parsers import YamlOutputParser
from langchain.prompts import PromptTemplateyaml_template = """生成Kubernetes部署配置:
{format_instructions}
應用需求:
- 名稱:frontend
- 副本數:3
- 鏡像:nginx:1.25
- 端口:80"""parser = YamlOutputParser()prompt = PromptTemplate(template=yaml_template,input_variables=[],partial_variables={"format_instructions": parser.get_format_instructions()}
)chain = prompt | ChatOpenAI() | parserresult = chain.invoke({})
print(result)
# 輸出:
"""
apiVersion: apps/v1
kind: Deployment
metadata:name: frontend
spec:replicas: 3template:spec:containers:- name: nginximage: nginx:1.25ports:- containerPort: 80
"""
最佳實踐技巧
- 混合模式輸出:
# 結合JSON和Pydantic
class BaseResponse(BaseModel):code: intdata: dicterror: str = Nonechain = prompt | model | JsonOutputParser() | BaseResponse.parse_obj
- 錯誤處理機制:
from langchain.schema import OutputParserExceptiontry:result = chain.invoke(...)
except OutputParserException as e:print(f"解析失敗,建議:{e.advice}")# 自動修復流程fixed_result = repair_json(e.llm_output)
- 格式強化提示:
# 在提示詞中加入示例
prompt_template += """
示例:
輸入:今天北京晴轉多云,氣溫25-32度
輸出:{"city": "北京", "weather": "晴轉多云", "temp_range": [25, 32]}"""
- 大模型參數優化:
ChatOpenAI(model="gpt-4",temperature=0.3, # 降低隨機性top_p=0.9,frequency_penalty=0.5 # 減少重復
)
輸出解析器對比表
解析器 | 適用場景 | 優點 | 缺點 |
---|---|---|---|
PydanticOutputParser | API響應/數據庫存儲 | 強類型驗證/自動文檔生成 | 需要預定義模型 |
JsonOutputParser | 快速原型/動態結構 | 靈活/無需預定義 | 無自動驗證 |
XMLOutputParser | 企業系統對接 | 結構清晰/支持復雜層級 | 冗余標簽/解析復雜度高 |
YamlOutputParser | 配置文件/DevOps | 可讀性強/適合配置管理 | 生態工具較少 |
根據實際需求選擇:
- 優先使用Pydantic保證數據可靠性
- 需要靈活結構時使用Json
- 對接傳統系統使用XML
- 生成配置文件使用YAML