15. LangChain多模态应用开发：融合文本、图像与语音

引言：当AI学会"看听说想"

2025年某智慧医院的多模态问诊系统，通过同时分析患者CT影像、语音描述和电子病历，将误诊率降低42%。本文将基于LangChain多模态框架与Deepseek-R1，手把手构建能理解复合信息的智能系统。

一、多模态技术栈全景

1.1 核心组件选型建议

模态	推荐模型	延迟	适用场景
文本	Deepseek-R1	350ms	病历分析/报告生成
图像	CLIP	520ms	医学影像识别
语音	Whisper	1200ms	医患对话转写
多模态	LLaVA	2100ms	跨模态推理

1.2 数据流架构

二、四步构建多模态问诊系统

2.1 多模态数据加载

from langchain_community.document_loaders import UnstructuredPDFLoader
from langchain_community.document_loaders import (
    ImageCaptionLoader      # 图像
)
from langchain.docstore.document import Document
import whisper

# 加载多源数据
text_docs = UnstructuredPDFLoader("patient_records.pdf").load()
image_docs = ImageCaptionLoader("ct_scan.jpg").load()

model = whisper.load_model("base")

# 识别音频（支持 MP3/WAV）
result = model.transcribe("symptoms_description.mp3",
    language="zh",  # 强制识别为中文
    initial_prompt="以下是普通话内容",  # 提升识别准确率
)
audio_docs = [Document(page_content=result["text"])]

2.2 特征提取与对齐

import numpy as np
from PIL import Image
from langchain_huggingface import HuggingFaceEmbeddings

# 文本嵌入 - 使用HuggingFace模型
text_embedder = HuggingFaceEmbeddings(model_name="BAAI/bge-base-zh-v1.5")
text_emb = text_embedder.embed_documents([doc.page_content for doc in text_docs])

# 图像嵌入 - 使用CLIP模型
# 需要先安装transformers和torch
from transformers import CLIPProcessor, CLIPModel
model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

def embed_image(image_path):
    image = Image.open(image_path)
    inputs = processor(images=image, return_tensors="pt")
    image_features = model.get_image_features(**inputs)
    return image_features.detach().numpy()

image_emb = embed_image("ct_scan.jpg")

# 特征融合
multimodal_emb = np.concatenate([text_emb, image_emb], axis=1)

2.3 多模态提示工程

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", """
    你是一名资深放射科医生，需要综合以下信息进行分析：
    1. 患者主诉：{audio_text}
    2. CT影像描述：{image_caption}
    3. 病史摘要：{medical_history}
    """),
    ("human", "请给出诊断意见和进一步检查建议")
])

2.4 混合生成与验证

from langchain_ollama import ChatOllama
from langchain_community.tools import TavilySearchResults
import os
# 修改为自己的key
os.environ["TAVILY_API_KEY"] = "tvly-dev-xxxxxx"

# 多模态生成
llm = ChatOllama(model="deepseek-r1")
diagnosis_chain = prompt | llm

# 事实核查工具
def verify_diagnosis(diagnosis):
    tool = TavilySearchResults(
        max_results=5,
        search_depth="advanced",
        include_answer=True,
    )
    return tool.invoke(f"医学验证：{diagnosis}")

# 执行管道
diagnosis = diagnosis_chain.invoke({
    "audio_text": audio_docs[0].page_content,
    "image_caption": image_docs[0].page_content,
    "medical_history": text_docs[0].page_content
})
verification = verify_diagnosis(diagnosis)
print(diagnosis.content)

输出为：

<think>
好，我现在要分析这位患者的病情。首先，患者是女性，44岁，主诉右下腹疼痛24小时，发热到38.5℃，既往史有高血压病史5年，还有青霉素过敏史。

CT影像描述显示的是一个“黑色的黑洞在中间”，这可能意味着肝脏的病变。考虑到肝癌的可能性较高，但需要结合其他检查结果来看。

接下来是分析主诉：疼痛从下午开始，最初在右下腹右侧，后来转移到右边，现在 pain score 7分左右，到了晚上达到10分，并且有恶心感，但没有剧烈疼痛或呕吐。体温37.8℃，没有发烧。这可能提示疼痛不是特别严重，但也并非完全无害。

CT结果中的“黑色黑洞”很可能是肝脏问题。肝癌的典型表现包括右下腹疼痛、发热和腹部检查发现肠鸣音减弱。陈女士的情况符合这些症状，尤其是结合她的既往史高血压，增加了患肝癌的风险。

不过，我也要考虑其他可能性。比如胆囊或胃的问题，但CT显示是黑色中间的黑洞，更倾向于肝脏问题。另外，考虑到她有青霉素过敏，如果有轻微过敏反应，可能会导致类似的疼痛和发热。

接下来，建议进一步检查：首先肯定是等待肝癌的标准流程，包括肝穿刺活检、影像学检查（如超声或MRI）来确认病变性质。此外，血常规、电解质和肝功能检查也很重要，特别是肝细胞癌的相关指标。

另外，考虑到她的症状较轻，可能还需要观察一段时间，看看疼痛是否加重或其他症状变化。如果出现发热明显，或者其他腹部症状如 full blood picture 降低等，就需要及时就医。

最后，建议她定期复查肝功能和影像学检查，以确保病情稳定。
</think>

### 诊断意见：
患者44岁女性，主诉右下腹疼痛伴发热，CT结果显示肝脏内有“黑色的黑洞”。结合既往史高血压及青霉素过敏，初步考虑为肝癌可能性较大。

### 进一步检查建议：
1. **肝穿刺活检**：以确认是否为肝癌。
2. **影像学检查**：包括超声或MRI进一步评估肝脏病变性质。
3. **血液检查**：血常规、电解质及肝功能评估，包括肝细胞癌相关指标（如CA-19-9）。
4. **观察病情变化**：密切监测疼痛和发热情况，必要时进行相应治疗方案调整。

2.4 测试文件生成

def generate_ct_scan(output_path="ct_scan.jpg"):
    # 创建模拟人体横截面（灰度值范围对应HU单位）
    image = np.zeros((512, 512), dtype=np.int16)

    # 模拟脊柱（高密度）
    cv2.circle(image, (256, 350), 30, 1000, -1)

    # 模拟内脏器官（中等密度）
    cv2.ellipse(image, (256, 200), (120, 80), 0, 0, 360, 40, -1)

    # 模拟病变区域（低密度）
    cv2.circle(image, (300, 220), 15, -100, -1)

    # 调整动态范围并保存
    normalized = ((image - image.min()) / (image.max() - image.min()) * 255).astype(np.uint8)
    cv2.imwrite(output_path, normalized)

generate_ct_scan()

from gtts import gTTS

# 典型中文主诉内容
symptoms = """
医生您好，我从昨天下午开始出现右下腹剧烈疼痛，
最开始是肚脐周围疼，后来转移到右边。
现在疼痛评分大概7分（满分10分），
伴有恶心但还没呕吐，
体温测量37.8度，没有腹泻。
"""

# 生成中文语音（需联网）
tts = gTTS(
    text=symptoms,
    lang='zh-cn',  # 中文普通话
    slow=False,
    tld='com'     
)
tts.save("symptoms_description.mp3")

from faker import Faker
from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas

fake = Faker("zh_CN")  # 指定中文数据生成

def generate_medical_record(output_path="patient_records.pdf"):
    c = canvas.Canvas(output_path, pagesize=A4)
    
    # 生成符合中文医疗文书规范的内容
    record = f"""
    电子病历
    姓名：{fake.name()}
    性别：{fake.random_element(("男","女"))}
    年龄：{fake.random_int(18, 80)}岁
    病历号：{fake.random_number(digits=8)}
    就诊时间：{fake.date(pattern="%Y年%m月%d日")}
    
    主诉：
    - {fake.random_element(("持续性咳嗽2周", "右下腹疼痛24小时", "头痛伴视力模糊3天"))}
    - {fake.random_element(("发热最高38.5℃", "恶心未呕吐", "乏力食欲减退"))}
    
    既往史：
    - 过敏史：{fake.random_element(("无", "青霉素过敏", "花粉过敏"))}
    - 慢性病：{fake.random_element(("高血压病史5年", "II型糖尿病", "支气管哮喘"))}
    
    体格检查：
    - 血压：{fake.random_int(90, 140)}/{fake.random_int(60, 90)}mmHg
    - 心率：{fake.random_int(60, 100)}次/分
    - 腹部查体：{fake.random_element(("麦氏点压痛阳性", "肠鸣音减弱", "无反跳痛"))}
    """
    
    # 设置中文字体（需系统支持）
    from reportlab.pdfbase import pdfmetrics
    from reportlab.pdfbase.ttfonts import TTFont
    pdfmetrics.registerFont(TTFont('SimSun', 'SimSun.ttf'))  # 使用宋体
    
    c.setFont('SimSun', 12)
    y_position = 750
    for line in record.split('\n'):
        c.drawString(50, y_position, line)
        y_position -= 20
    c.save()

generate_medical_record()

三、性能优化三大策略

3.1 异步并行处理

from langchain_core.runnables import RunnableParallel, RunnableLambda

async_pipeline = RunnableParallel(
    text_embedding=RunnableLambda(lambda x: text_embedder.embed_text(x)),
    image_embedding=RunnableLambda(lambda x: image_embedder.embed_image(x))
)

async def call():
    results = await async_pipeline.ainvoke({
        "text": "右下腹持续性疼痛",
        "image": "ct_scan.jpg"
    })

3.2 分级特征提取

# 根据紧急程度选择模型
def adaptive_feature_extraction(data):
    if data["priority"] == "high":
        return clip_large.embed_image(data["image"])
    return clip_fast.embed_image(data["image"])