LangChain学习笔记05——多模态开发与工具使用

一、多模态集成开发

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  # 网络图片直接使用URLelse: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 RunnableLambdadef 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 文档