多模态集成
多模态应用在人工智能中变得越来越重要，能够实现更丰富的交互和更复杂的任务。Model Context Protocol (MCP) 提供了一个构建多模态应用的框架，能够处理各种类型的数据，如文本、图像和音频。
MCP 不仅支持基于文本的交互，还支持多模态功能，使模型能够处理图像、音频及其他数据类型。
介绍
在本课中，你将学习如何构建一个多模态应用。
学习目标
完成本课后，你将能够：
- 理解多模态的选择
- 实现一个多模态应用
多模态支持的架构
多模态 MCP 实现通常包括：
- 特定模态解析器：将不同媒体类型转换为模型可处理格式的组件。
- 特定模态工具：专门处理特定模态（图像分析、音频处理）的工具。
- 统一上下文管理：跨不同模态维护上下文的系统。
- 响应生成：能够生成包含多种模态的响应。
多模态示例：图像分析
下面的示例中，我们将分析一张图像并提取信息。
C# 实现
using ModelContextProtocol.SDK.Server;
using ModelContextProtocol.SDK.Server.Tools;
using ModelContextProtocol.SDK.Server.Content;
using System.Text.Json;
using System.IO;
using System.Threading.Tasks;
using System.Collections.Generic;
namespace MultiModalMcpExample
{
// Tool for image analysis
public class ImageAnalysisTool : ITool
{
private readonly IImageAnalysisService _imageService;
public ImageAnalysisTool(IImageAnalysisService imageService)
{
_imageService = imageService;
}
public string Name => "imageAnalysis";
public string Description => "Analyzes image content and extracts information";
public ToolDefinition GetDefinition()
{
return new ToolDefinition
{
Name = Name,
Description = Description,
Parameters = new Dictionary<string, ParameterDefinition>
{
["imageUrl"] = new ParameterDefinition
{
Type = ParameterType.String,
Description = "URL to the image to analyze"
},
["analysisType"] = new ParameterDefinition
{
Type = ParameterType.String,
Description = "Type of analysis to perform",
Enum = new[] { "general", "objects", "text", "faces" },
Default = "general"
}
},
Required = new[] { "imageUrl" }
};
}
public async Task<ToolResponse> ExecuteAsync(IDictionary<string, object> parameters)
{
// Extract parameters
string imageUrl = parameters["imageUrl"].ToString();
string analysisType = parameters.ContainsKey("analysisType")
? parameters["analysisType"].ToString()
: "general";
// Download or access the image
byte[] imageData = await DownloadImageAsync(imageUrl);
// Analyze based on the requested analysis type
var analysisResult = analysisType switch
{
"objects" => await _imageService.DetectObjectsAsync(imageData), "text" => await _imageService.RecognizeTextAsync(imageData),
"faces" => await _imageService.DetectFacesAsync(imageData),
_ => await _imageService.AnalyzeGeneralAsync(imageData) // Default general analysis
};
// Return structured result as a ToolResponse
// Format follows the MCP specification for content structure
var content = new List<ContentItem>
{
new ContentItem
{
Type = ContentType.Text,
Text = JsonSerializer.Serialize(analysisResult)
}
};
return new ToolResponse
{
Content = content,
IsError = false
};
}
private async Task<byte[]> DownloadImageAsync(string url)
{
using var httpClient = new HttpClient();
return await httpClient.GetByteArrayAsync(url);
}
}
// Multi-modal MCP server with image and text processing
public class MultiModalMcpServer
{
public static async Task Main(string[] args)
{
// Create an MCP server
var server = new McpServer(
name: "Multi-Modal MCP Server",
version: "1.0.0"
);
// Configure server for multi-modal support
var serverOptions = new McpServerOptions
{
MaxRequestSize = 10 * 1024 * 1024, // 10MB for larger payloads like images
SupportedContentTypes = new[]
{
"image/jpeg",
"image/png",
"text/plain",
"application/json"
}
};
// Create image analysis service
var imageService = new ComputerVisionService();
// Register image analysis tools
server.AddTool(new ImageAnalysisTool(imageService));
// Register a text-to-image tool
services.AddMcpTool<TextAnalysisTool>();
services.AddMcpTool<ImageAnalysisTool>();
services.AddMcpTool<DocumentGenerationTool>(); // Tool that can generate documents with text and images
}
}
}
在上述示例中，我们：
- 创建了一个
ImageAnalysisTool，使用假设的IImageAnalysisService来分析图像。 - 配置了 MCP 服务器以处理更大请求并支持图像内容类型。
- 将图像分析工具注册到服务器。
- 实现了一个方法，从 URL 下载图像并根据请求类型（对象、文本、面部等）进行分析。
- 返回符合 MCP 规范的结构化结果。
多模态示例：音频处理
音频处理是多模态应用中另一种常见的模态。下面是一个实现音频转录工具的示例，该工具可以处理音频文件并返回转录文本。
Java 实现
package com.example.mcp.multimodal;
import com.mcp.server.McpServer;
import com.mcp.tools.Tool;
import com.mcp.tools.ToolRequest;
import com.mcp.tools.ToolResponse;
import com.mcp.tools.ToolExecutionException;
import com.example.audio.AudioProcessor;
import java.util.Base64;
import java.util.HashMap;
import java.util.Map;
// Audio transcription tool
public class AudioTranscriptionTool implements Tool {
private final AudioProcessor audioProcessor;
public AudioTranscriptionTool(AudioProcessor audioProcessor) {
this.audioProcessor = audioProcessor;
}
@Override
public String getName() {
return "audioTranscription";
}
@Override
public String getDescription() {
return "Transcribes speech from audio files to text";
}
@Override
public Object getSchema() {
Map<String, Object> schema = new HashMap<>();
schema.put("type", "object");
Map<String, Object> properties = new HashMap<>();
Map<String, Object> audioUrl = new HashMap<>();
audioUrl.put("type", "string");
audioUrl.put("description", "URL to the audio file to transcribe");
Map<String, Object> audioData = new HashMap<>();
audioData.put("type", "string");
audioData.put("description", "Base64-encoded audio data (alternative to URL)");
Map<String, Object> language = new HashMap<>();
language.put("type", "string");
language.put("description", "Language code (e.g., 'en-US', 'es-ES')");
language.put("default", "en-US");
properties.put("audioUrl", audioUrl);
properties.put("audioData", audioData);
properties.put("language", language);
schema.put("properties", properties);
schema.put("required", Arrays.asList("audioUrl"));
return schema;
}
@Override
public ToolResponse execute(ToolRequest request) {
try {
byte[] audioData;
String language = request.getParameters().has("language") ?
request.getParameters().get("language").asText() : "en-US";
// Get audio either from URL or direct data
if (request.getParameters().has("audioUrl")) {
String audioUrl = request.getParameters().get("audioUrl").asText();
audioData = downloadAudio(audioUrl);
} else if (request.getParameters().has("audioData")) {
String base64Audio = request.getParameters().get("audioData").asText();
audioData = Base64.getDecoder().decode(base64Audio);
} else {
throw new ToolExecutionException("Either audioUrl or audioData must be provided");
}
// Process audio and transcribe
Map<String, Object> transcriptionResult = audioProcessor.transcribe(audioData, language);
// Return transcription result
return new ToolResponse.Builder()
.setResult(transcriptionResult)
.build();
} catch (Exception ex) {
throw new ToolExecutionException("Audio transcription failed: " + ex.getMessage(), ex);
}
}
private byte[] downloadAudio(String url) {
// Implementation for downloading audio from URL
// ...
return new byte[0]; // Placeholder
}
}
// Main application with audio and other modalities
public class MultiModalApplication {
public static void main(String[] args) {
// Configure services
AudioProcessor audioProcessor = new AudioProcessor();
ImageProcessor imageProcessor = new ImageProcessor();
// Create and configure server
McpServer server = new McpServer.Builder()
.setName("Multi-Modal MCP Server")
.setVersion("1.0.0")
.setPort(5000)
.setMaxRequestSize(20 * 1024 * 1024) // 20MB for audio/video content
.build();
// Register multi-modal tools
server.registerTool(new AudioTranscriptionTool(audioProcessor));
server.registerTool(new ImageAnalysisTool(imageProcessor));
server.registerTool(new VideoProcessingTool());
// Start server
server.start();
System.out.println("Multi-Modal MCP Server started on port 5000");
}
}
在上述示例中，我们：
- 创建了一个
AudioTranscriptionTool，能够转录音频文件。 - 定义了工具的 schema，接受 URL 或 base64 编码的音频数据。
- 实现了
execute方法来处理音频处理和转录。 - 配置了 MCP 服务器以处理多模态请求，包括音频和图像处理。
- 将音频转录工具注册到服务器。
- 实现了一个方法，从 URL 下载音频文件或解码 base64 音频数据。
- 使用
AudioProcessor服务来处理实际的转录逻辑。 - 启动 MCP 服务器以监听请求。
多模态示例：多模态响应生成
Python 实现
from mcp_server import McpServer
from mcp_tools import Tool, ToolRequest, ToolResponse, ToolExecutionException
import base64
from PIL import Image
import io
import requests
import json
from typing import Dict, Any, List, Optional
# Image generation tool
class ImageGenerationTool(Tool):
def get_name(self):
return "imageGeneration"
def get_description(self):
return "Generates images based on text descriptions"
def get_schema(self):
return {
"type": "object",
"properties": {
"prompt": {
"type": "string",
"description": "Text description of the image to generate"
},
"style": {
"type": "string",
"enum": ["realistic", "artistic", "cartoon", "sketch"],
"default": "realistic"
},
"width": {
"type": "integer",
"default": 512
},
"height": {
"type": "integer",
"default": 512
}
},
"required": ["prompt"]
}
async def execute_async(self, request: ToolRequest) -> ToolResponse:
try:
# Extract parameters
prompt = request.parameters.get("prompt")
style = request.parameters.get("style", "realistic")
width = request.parameters.get("width", 512)
height = request.parameters.get("height", 512)
# Generate image using external service (example implementation)
image_data = await self._generate_image(prompt, style, width, height)
# Convert image to base64 for response
buffered = io.BytesIO()
image_data.save(buffered, format="PNG")
img_str = base64.b64encode(buffered.getvalue()).decode()
# Return result with both the image and metadata
return ToolResponse(
result={
"imageBase64": img_str,
"format": "image/png",
"width": width,
"height": height,
"generationPrompt": prompt,
"style": style
}
)
except Exception as e:
raise ToolExecutionException(f"Image generation failed: {str(e)}")
async def _generate_image(self, prompt: str, style: str, width: int, height: int) -> Image.Image:
"""
This would call an actual image generation API
Simplified placeholder implementation
"""
# Return a placeholder image or call actual image generation API
# For this example, we'll create a simple colored image
image = Image.new('RGB', (width, height), color=(73, 109, 137))
return image
# Multi-modal response handler
class MultiModalResponseHandler:
"""Handler for creating responses that combine text, images, and other modalities"""
def __init__(self, mcp_client):
self.client = mcp_client
async def create_multi_modal_response(self,
text_content: str,
generate_images: bool = False,
image_prompts: Optional[List[str]] = None) -> Dict[str, Any]:
"""
Creates a response that may include generated images alongside text
"""
response = {
"text": text_content,
"images": []
}
# Generate images if requested
if generate_images and image_prompts:
for prompt in image_prompts:
image_result = await self.client.execute_tool(
"imageGeneration",
{
"prompt": prompt,
"style": "realistic",
"width": 512,
"height": 512
}
)
response["images"].append({
"imageData": image_result.result["imageBase64"],
"format": image_result.result["format"],
"prompt": prompt
})
return response
# Main application
async def main():
# Create server
server = McpServer(
name="Multi-Modal MCP Server",
version="1.0.0",
port=5000
)
# Register multi-modal tools
server.register_tool(ImageGenerationTool())
server.register_tool(AudioAnalysisTool())
server.register_tool(VideoFrameExtractionTool())
# Start server
await server.start()
print("Multi-Modal MCP Server running on port 5000")
if __name__ == "__main__":
import asyncio
asyncio.run(main())
接下来
免责声明：
本文件使用 AI 翻译服务 Co-op Translator 进行翻译。虽然我们力求准确，但请注意，自动翻译可能包含错误或不准确之处。原始文件的母语版本应被视为权威来源。对于重要信息，建议使用专业人工翻译。我们不对因使用本翻译而产生的任何误解或误释承担责任。