【带你 langchain 双排系列教程】6.LangChain多模态输入与自定义输出实战指南

一、为什么需要多模态交互？

在真实业务场景中，数据从来都不是单一形式的。想象一个智能客服系统需要同时分析用户的文字描述、上传的产品图片和语音留言，或者一个内容审核系统需要检查文本、图像和视频的组合内容。传统单一模态的处理方式已经无法满足这些需求。

二、环境准备

pip install langchain-openai Pillow python-multipart librosa

三、多模态输入处理实战

1. 基础文本处理

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAIprompt = ChatPromptTemplate.from_template("分析这段话的情感倾向：{text}")
model = ChatOpenAI(model="gpt-4")chain = prompt | model
response = chain.invoke({"text": "虽然产品功能不错，但客服响应太慢了"})
print(response.content)

2. 图像处理进阶

from PIL import Image
import base64def image_to_base64(image_path):with open(image_path, "rb") as image_file:return base64.b64encode(image_file.read()).decode('utf-8')image_analysis_prompt = """
请分析这张图片：
<img src="data:image/png;base64,{image}" width=300/>
重点说明：
1. 主要视觉元素
2. 颜色搭配特点
3. 可能的商业应用场景
"""prompt = ChatPromptTemplate.from_template(image_analysis_prompt)
chain = prompt | modelresponse = chain.invoke({"image": image_to_base64("product_design.png")
})
print(response.content)

图片url方式多模态，模型可用视觉相关的模型，有些对话模型或不支持多模态。
在这里插入图片描述

import base64
import httpx
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
import osimage_url = "https://scpic.chinaz.net/files/pic/pic9/201602/apic19010.jpg"
image_data = base64.b64encode(httpx.get(image_url).content).decode("utf-8")
model = ChatOpenAI(model="deepseek-ai/deepseek-vl2",temperature=0.8,max_tokens=None,timeout=None,max_retries=2,api_key=os.getenv("silicon-key"),base_url="https://api.siliconflow.cn/v1"# organization="...",# other params...
)
message = HumanMessage(content=[{"type": "text", "text": "用中文描述这张图片中的天气"},{"type": "image_url", "image_url": {"url": image_url}},], )
response = model.invoke([message])
print(response.content)

在这里插入图片描述
多张图片的情况：

import base64
import httpx
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
import osimage_url = "https://scpic.chinaz.net/files/pic/pic9/201602/apic19010.jpg"
image_url2 = "https://pic.nximg.cn/file/20140920/18021896_181234816000_2.jpg"
image_data = base64.b64encode(httpx.get(image_url).content).decode("utf-8")
model = ChatOpenAI(model="deepseek-ai/deepseek-vl2",temperature=0.8,max_tokens=None,timeout=None,max_retries=2,api_key=os.getenv("silicon-key"),base_url="https://api.siliconflow.cn/v1"# organization="...",# other params...
)
message = HumanMessage(content=[{"type": "text", "text": "用中文描述这张2图片中的天气是不是一样的"},{"type": "image_url", "image_url": {"url": image_url}},{"type": "image_url", "image_url": {"url": image_url2}},], )
response = model.invoke([message])
print(response.content)

在这里插入图片描述
工具调用方式多模态处理：

import base64
import httpx
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
import os
from langchain_core.tools import tool
from typing import Literal@tool
def weather_tool(weather: Literal["晴朗的", "多云的", "多雨的", "下雪的"]) -> None:""" Description the weather"""passimage_url = "https://scpic.chinaz.net/files/pic/pic9/201602/apic19010.jpg"
image_url2 = "https://pic.nximg.cn/file/20140920/18021896_181234816000_2.jpg"
image_data = base64.b64encode(httpx.get(image_url).content).decode("utf-8")
model = ChatOpenAI(model="deepseek-ai/deepseek-vl2",temperature=0.8,max_tokens=None,timeout=None,max_retries=2,api_key=os.getenv("silicon-key"),base_url="https://api.siliconflow.cn/v1"# organization="...",# other params...
)model.bind_tools([weather_tool])message = HumanMessage(content=[{"type": "text", "text": "用中文描述这张2图片中的天气是不是一样的"},{"type": "image_url", "image_url": {"url": image_url}},{"type": "image_url", "image_url": {"url": image_url2}},], )
response = model.invoke([message])
print(response.content)

在这里插入图片描述

3. 音频处理技巧

import librosa
import numpy as npdef audio_feature_extraction(file_path):y, sr = librosa.load(file_path)return {"duration": len(y)/sr,"tempo": librosa.beat.tempo(y=y, sr=sr)[0],"spectral_centroid": np.mean(librosa.feature.spectral_centroid(y=y, sr=sr))}audio_features = audio_feature_extraction("customer_feedback.wav")analysis_prompt = """
根据以下音频特征生成报告：
- 时长：{duration:.2f}秒
- 节奏：{tempo:.1f}BPM
- 频谱重心：{spectral_centroid:.1f}Hz结合语音内容："{transcript}"
分析客户情绪状态"""

四、自定义输出深度解析

1. 结构化输出实践

from pydantic import BaseModel
from langchain_core.output_parsers import PydanticOutputParserclass MarketAnalysis(BaseModel):target_audience: strcompetitive_advantages: list[str]risk_factors: list[str]market_score: float parser = PydanticOutputParser(pydantic_object=MarketAnalysis)prompt_template = """
执行深度市场分析：
{instruction}
{format_instructions}
"""
prompt = ChatPromptTemplate.from_template(prompt_template).partial(format_instructions=parser.get_format_instructions()
)chain = prompt | model | parser
result = chain.invoke({"instruction": "分析智能手表市场现状"
})print(f"目标人群：{result.target_audience}")
print(f"风险因素：{result.risk_factors[:2]}")

2. 动态格式转换

from langchain_core.output_parsers import JsonOutputParserdynamic_prompt = """
根据用户查询生成响应：
查询内容：{query}
要求格式：{format}
"""chain = ChatPromptTemplate.from_template(dynamic_prompt) | model | JsonOutputParser()response = chain.invoke({"query": "对比Python和JavaScript的优缺点","format": """{"comparison_points": ["语法", "生态系统", "性能"],"summary": "综合建议"}"""
})print(response["summary"])

五、综合实战案例：智能内容审核系统

class ContentAuditResult(BaseModel):violation_type: Optional[str]confidence: floatproblematic_elements: list[str]recommendation: strdef multimodal_audit(text: str, image: Image, audio: bytes):# 图像处理image_analysis = chain_image.invoke(...)# 音频转录audio_text = transcribe_audio(audio)# 综合审核combined_prompt = """执行内容审核：文本内容：{text}图片分析：{image_analysis}语音转录：{audio_text}输出要求：{format_instructions}"""chain = (ChatPromptTemplate.from_template(combined_prompt).partial(format_instructions=parser.get_format_instructions())| model| parser)return chain.invoke(...)# 使用示例
result = multimodal_audit(text="敏感文本内容",image=Image.open("user_image.jpg"),audio=audio_bytes
)

六、性能优化技巧

并行处理：使用async/await实现多模态并行处理
缓存策略：对静态内容（如产品图片）实施特征缓存
分级处理：先快速筛查再深度分析的分级处理机制

async def parallel_processing(text, image, audio):image_task = chain_image.ainvoke(...)audio_task = chain_audio.ainvoke(...)results = await asyncio.gather(image_task, audio_task)return combine_results(text, *results)