一、多模态集成开发

1.1. LangChain 多模态实现目标

我们这里以图文问答为例：

• 输入：图片 + 用户问题
• 输出：基于图片内容的自然语言回答

多模态的关键在于：

• 图像向量化（使用模型如 BLIP2 / CLIP / MiniGPT-4）
• 文本提问嵌入
• 多模态融合模型执行推理

1.2. 实现代码：图文问答链

1.2.1. 步骤说明

1. 读取图片，传入视觉模型（如 BLIP2）获取图像描述
2. 构建文本提示模板
3. 用qwen-vl-plus模型生成答案

1.2.2. 完整示例代码：

import base64  # 用于将图片编码为 base64 格式
import requests  # 用于发送 HTTP 请求调用 DeepSeek 接口
from PIL import Image  # 用于读取和处理图片
from io import BytesIO  # 用于创建内存缓冲区
from langchain_core.runnables import RunnableLambda  # LangChain 中用于链式封装的工具

DASHSCOPE_API_KEY = "你的api key" # 通义千问模型
DASHSCOPE_API_URL = "https://dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation"

# ✅ 将图片转为 base64 格式，供 API 使用
def image_to_base64(image_path: str) -> str:
    # 打开图像，并转换为 RGB 模式（去除透明通道，统一格式）
    img = Image.open(image_path).convert("RGB")
    # 创建内存缓冲区，用于存储图像二进制数据
    buffered = BytesIO()
    # 将图像保存为 JPEG 格式写入内存
    img.save(buffered, format="JPEG")
    img_str = base64.b64encode(buffered.getvalue()).decode("utf-8")
    # 将图像数据编码为 base64 字符串并返回
    return f"data:image/jpeg;base64,{img_str}"  # 必须包含MIME前缀

def ask_qwen_multimodal(inputs: dict) -> str:
    """调用通义千问多模态模型处理图文问答"""
    image_source = inputs["image_url"]
    question = inputs["question"]

    # 处理图片源：本地路径转base64，URL直接使用
    if image_source.startswith(("http://", "https://")):
        image_data = image_source  # 网络图片直接使用URL
    else:
        image_data = image_to_base64(image_source)  # 本地图片转base64

    # 设置请求头
    headers = {
        "Authorization": f"Bearer {DASHSCOPE_API_KEY}",
        "Content-Type": "application/json"
    }

    # 构造通义千问API请求体（符合DashScope规范）
    payload = {
        "model": "qwen-vl-plus",  # 可选：qwen-vl-plus/qwen-vl-max
        "input": {
            "messages": [
                {
                    "role": "user",
                    "content": [
                        {"image": image_data},  # 图片数据
                        {"text": question}  # 问题文本
                    ]
                }
            ]
        },
        "parameters": {
            "temperature": 0.6,
            "top_p": 0.8
        }
    }

    # 发送API请求
    response = requests.post(DASHSCOPE_API_URL, headers=headers, json=payload)

    # 错误处理
    try:
        response.raise_for_status()
    except requests.exceptions.HTTPError as e:
        error_detail = response.json().get("message", "未知错误")
        raise Exception(f"API请求失败 [{response.status_code}]: {error_detail}") from e

    # 解析响应（通义千问返回结构）
    response_data = response.json()
    return response_data["output"]["choices"][0]["message"]["content"]


# ✅ 使用 LangChain 的 Runnable 封装成链式调用单元
multimodal_chain = RunnableLambda(ask_qwen_multimodal)

# ✅ 测试用例（支持本地路径和URL）
if __name__ == "__main__":
    # 测试配置（根据需要修改）
    test_inputs = {
        "image_url": "https://gitee.com/ACERT6/langchain/raw/master/true.jpg",
        "question": "图片中有什么动物？它们在做什么？"
    }

    # 执行调用
    try:
        result = multimodal_chain.invoke(test_inputs)
        print("\n✅ 通义千问回答：")
        print("-" * 50)
        print(result)
        print("-" * 50)
    except Exception as e:
        print(f"\n❌ 调用失败: {str(e)}")

1.3. 总结

项目	描述
多模态核心	图像 + 文本融合建模
LangChain 作用	提供链式调用与结构化逻辑处理
LECL 亮点	极大提升多模态处理链的组合性和可读性

二、工具

LCEL 工具使用指南

LCEL 允许通过管道符 | 声明式组合链式组件，简化复杂工作流的构建。

1. 基础工具调用

from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 定义自定义工具
@tool
def calculate_length(text: str) -> int:
    """计算输入字符串的长度"""
    return len(text)

# 创建 LCEL 链
chain = (
    ChatPromptTemplate.from_template("分析内容: {input}") 
    | ChatOpenAI(model="gpt-3.5-turbo") 
    | calculate_length  # 直接调用工具
)

result = chain.invoke({"input": "Hello, world!"})
print(result)  # 输出: 13

2. 动态工具选择 (Tool Routing)

使用 bind_tools 让模型智能选择工具：

from langchain_core.utils.function_calling import convert_to_openai_tool

# 定义多个工具
tools = [calculate_length, ...]  # 添加其他工具

# 创建支持工具调用的模型
model = ChatOpenAI().bind_tools(tools)

# 构建决策链
chain = (
    {"input": lambda x: x["input"]} 
    | ChatPromptTemplate.from_template("处理请求: {input}") 
    | model
    | (lambda msg: tool_map[msg.tool_calls[0]['name']](msg.tool_calls[0]['args']))
)

3. 组合工具流

from langchain_core.runnables import RunnablePassthrough

# 定义工具预处理链
preprocess = (
    RunnablePassthrough.assign(
        cleaned_input=lambda x: x["input"].strip().lower()
    ) 
    | {"data": RunnablePassthrough()}
)

# 完整工作流
full_chain = (
    preprocess 
    | calculate_length 
    | {"original": RunnablePassthrough(), "length": RunnablePassthrough()}
)

full_chain.invoke({"input": "  HELLO  "}) 
# 输出: {'original': '  HELLO  ', 'length': 5}

4. 错误处理

from langchain_core.runnables import RunnableLambda

def safe_tool_call(args):
    try:
        return calculate_length(args)
    except Exception as e:
        return f"Tool error: {str(e)}"

resilient_chain = (
    RunnablePassthrough() 
    | RunnableLambda(safe_tool_call)
)

5. 流式输出

# 流式返回工具结果
async for chunk in chain.astream({"input": "Stream me!"}):
    print(chunk, end="", flush=True)

关键技巧

1. 工具绑定

model_with_tools = ChatOpenAI().bind_tools([tool1, tool2])

1. 参数解析

from langchain_core.output_parsers import JsonOutputParser
parser = JsonOutputParser()
chain = model | parser

1. 调试模式

chain = ... | calculate_length
chain.get_graph().print_ascii()  # 打印拓扑图

1. 工具组合

multi_tool_chain = tool1 | tool2  # 前一个工具输出作为后一个输入

最佳实践

• 命名规范：工具函数名需清晰（如 get_weather_data）
• 类型提示：严格定义工具输入类型（如 text: str）
• 错误处理：在工具内部捕获异常，返回结构化错误信息
• 性能优化：对耗时工具使用 RunnableLambda 异步化

官方资源：

• LCEL 文档