多模态输入
- 图片处理(最高频)
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