15：多模态模型在Agentic产品中的杀手级应用

class MultimodalAgent:
    def __init__(self, model_name="gemini-1.5-pro"):
        self.model = self.load_model(model_name)
        self.image_processor = self.load_image_processor()
        self.audio_processor = self.load_audio_processor()
        self.video_processor = self.load_video_processor()
    
    def load_model(self, model_name):
        # 实际实现会加载相应的多模态模型
        return lambda prompt, images=None, audio=None, video=None: f"模型响应：{prompt}"
    
    def load_image_processor(self):
        # 实际实现会加载图像处理器
        return lambda image: image
    
    def load_audio_processor(self):
        # 实际实现会加载音频处理器
        return lambda audio: audio
    
    def load_video_processor(self):
        # 实际实现会加载视频处理器
        return lambda video: video
    
    def process_text(self, text):
        """处理文本输入"""
        return text
    
    def process_image(self, image_path):
        """处理图像输入"""
        with open(image_path, 'rb') as f:
            image = f.read()
        return self.image_processor(image)
    
    def process_audio(self, audio_path):
        """处理音频输入"""
        with open(audio_path, 'rb') as f:
            audio = f.read()
        return self.audio_processor(audio)
    
    def process_video(self, video_path):
        """处理视频输入"""
        with open(video_path, 'rb') as f:
            video = f.read()
        return self.video_processor(video)
    
    def run(self, text=None, image_paths=None, audio_paths=None, video_paths=None):
        """运行多模态推理"""
        # 处理输入
        processed_text = self.process_text(text) if text else ""
        processed_images = [self.process_image(path) for path in image_paths] if image_paths else []
        processed_audio = [self.process_audio(path) for path in audio_paths] if audio_paths else []
        processed_video = [self.process_video(path) for path in video_paths] if video_paths else []
        
        # 构建提示
        prompt = f"""请基于以下输入提供响应：

文本：{processed_text}

请提供详细、准确的回答。"""
        
        # 调用模型
        response = self.model(
            prompt,
            images=processed_images,
            audio=processed_audio,
            video=processed_video
        )
        
        return response

# 示例使用
if __name__ == "__main__":
    agent = MultimodalAgent()
    response = agent.run(
        text="这张图片里有什么？",
        image_paths=["cat.jpg"]
    )
    print(response)

class MultimodalToolManager:
    def __init__(self):
        self.tools = {
            'image_analysis': self.analyze_image,
            'audio_transcription': self.transcribe_audio,
            'video_analysis': self.analyze_video,
            'text_generation': self.generate_text,
            'image_generation': self.generate_image,
            'audio_generation': self.generate_audio,
            'video_generation': self.generate_video
        }
    
    def analyze_image(self, image_path):
        """分析图像"""
        # 实际实现会调用图像分析API
        return f"图像分析结果：{image_path}"
    
    def transcribe_audio(self, audio_path):
        """转录音频"""
        # 实际实现会调用音频转录API
        return f"音频转录结果：{audio_path}"
    
    def analyze_video(self, video_path):
        """分析视频"""
        # 实际实现会调用视频分析API
        return f"视频分析结果：{video_path}"
    
    def generate_text(self, prompt):
        """生成文本"""
        # 实际实现会调用文本生成API
        return f"生成的文本：{prompt}"
    
    def generate_image(self, prompt):
        """生成图像"""
        # 实际实现会调用图像生成API
        return f"生成的图像：{prompt}"
    
    def generate_audio(self, text):
        """生成音频"""
        # 实际实现会调用音频生成API
        return f"生成的音频：{text}"
    
    def generate_video(self, prompt):
        """生成视频"""
        # 实际实现会调用视频生成API
        return f"生成的视频：{prompt}"
    
    def execute_tool(self, tool_name, **params):
        """执行工具"""
        if tool_name in self.tools:
            return self.tools[tool_name](**params)
        else:
            return f"错误：工具 {tool_name} 不存在"

class MultimodalContextManager:
    def __init__(self, max_context_size=100):
        self.max_context_size = max_context_size
        self.context = []
    
    def add_text(self, text):
        """添加文本到上下文"""
        self.context.append({'type': 'text', 'content': text})
        self._trim_context()
    
    def add_image(self, image_path):
        """添加图像到上下文"""
        self.context.append({'type': 'image', 'content': image_path})
        self._trim_context()
    
    def add_audio(self, audio_path):
        """添加音频到上下文"""
        self.context.append({'type': 'audio', 'content': audio_path})
        self._trim_context()
    
    def add_video(self, video_path):
        """添加视频到上下文"""
        self.context.append({'type': 'video', 'content': video_path})
        self._trim_context()
    
    def get_context(self):
        """获取上下文"""
        return self.context
    
    def _trim_context(self):
        """修剪上下文以保持大小限制"""
        while len(self.context) > self.max_context_size:
            self.context.pop(0)

import os
import time
from typing import List, Dict, Any, Optional

class MultimodalAgent:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.model = self._load_model()
        self.context_manager = MultimodalContextManager(max_context_size=config.get('max_context_size', 100))
        self.tool_manager = MultimodalToolManager()
    
    def _load_model(self):
        """加载多模态模型"""
        model_name = self.config.get('model_name', 'gemini-1.5-pro')
        # 实际实现会加载相应的模型
        return lambda prompt, images=None, audio=None, video=None: f"模型响应：{prompt}"
    
    def process_input(self, input_data: Dict[str, Any]):
        """处理多模态输入"""
        # 处理文本输入
        if 'text' in input_data:
            text = input_data['text']
            self.context_manager.add_text(text)
        
        # 处理图像输入
        if 'images' in input_data:
            for image_path in input_data['images']:
                self.context_manager.add_image(image_path)
        
        # 处理音频输入
        if 'audio' in input_data:
            for audio_path in input_data['audio']:
                self.context_manager.add_audio(audio_path)
        
        # 处理视频输入
        if 'video' in input_data:
            for video_path in input_data['video']:
                self.context_manager.add_video(video_path)
    
    def run(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """运行多模态推理"""
        start_time = time.time()
        
        # 处理输入
        self.process_input(input_data)
        
        # 获取上下文
        context = self.context_manager.get_context()
        
        # 构建提示
        prompt = self._build_prompt(context, input_data.get('text', ''))
        
        # 准备多模态输入
        images = [item['content'] for item in context if item['type'] == 'image']
        audio = [item['content'] for item in context if item['type'] == 'audio']
        video = [item['content'] for item in context if item['type'] == 'video']
        
        # 调用模型
        response = self.model(prompt, images=images, audio=audio, video=video)
        
        # 处理工具调用
        if '```tool_call' in response:
            tool_call = self._parse_tool_call(response)
            tool_result = self.tool_manager.execute_tool(**tool_call)
            
            # 构建工具响应提示
            tool_prompt = self._build_tool_prompt(context, input_data.get('text', ''), tool_result)
            response = self.model(tool_prompt, images=images, audio=audio, video=video)
        
        end_time = time.time()
        
        return {
            'response': response,
            'processing_time': end_time - start_time,
            'timestamp': time.time()
        }
    
    def _build_prompt(self, context: List[Dict[str, Any]], user_input: str) -> str:
        """构建提示"""
        prompt = f"""你是一个智能多模态助手，需要根据用户的输入提供帮助。

用户输入：{user_input}

上下文：
"""
        
        for item in context:
            if item['type'] == 'text':
                prompt += f"文本：{item['content']}\n"
            elif item['type'] == 'image':
                prompt += f"图像：{item['content']}\n"
            elif item['type'] == 'audio':
                prompt += f"音频：{item['content']}\n"
            elif item['type'] == 'video':
                prompt += f"视频：{item['content']}\n"
        
        prompt += """

可用工具：
- image_analysis: 分析图像内容
- audio_transcription: 转录音频内容
- video_analysis: 分析视频内容
- text_generation: 生成文本内容
- image_generation: 根据描述生成图像
- audio_generation: 将文本转换为音频
- video_generation: 根据描述生成视频

如果需要使用工具，请按照以下格式输出：
```tool_call
工具名称
参数1: 值1
参数2: 值2

    return prompt

def _build_tool_prompt(self, context: List[Dict[str, Any]], user_input: str, tool_result: str) -> str:
    """构建工具响应提示"""
    prompt = f"""你是一个智能多模态助手，需要根据用户的输入和工具执行结果提供帮助。

    return prompt

def _parse_tool_call(self, response: str) -> Dict[str, Any]:
    """解析工具调用"""
    # 简单的工具调用解析
    lines = response.split('\n')
    tool_name = None
    params = {}
    
    for i, line in enumerate(lines):
        if line.strip() == '```tool_call':
            tool_name = lines[i+1].strip()
            for j in range(i+2, len(lines)):
                if lines[j].strip() == '```':
                    break
                if ':' in lines[j]:
                    key, value = lines[j].split(':', 1)
                    params[key.strip()] = value.strip()
            break
    
    if tool_name:
        params['tool_name'] = tool_name
    
    return params

agent = MultimodalAgent(config)
response = agent.run({
    'text': '这张图片里有什么？',
    'images': ['cat.jpg']
})
print(response)

### 11.2 企业级多模态系统

```python
class EnterpriseMultimodalSystem:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.agents = {}
        self.init_agents()
        self.logging = self._init_logging()
    
    def init_agents(self):
        """初始化多模态代理"""
        agent_configs = self.config.get('agents', {
            'customer_service': {
                'model_name': 'gemini-1.5-pro',
                'max_context_size': 100
            },
            'product_recommendation': {
                'model_name': 'gemini-1.5-pro',
                'max_context_size': 100
            },
            'technical_support': {
                'model_name': 'gemini-1.5-pro',
                'max_context_size': 100
            }
        })
        
        for agent_name, agent_config in agent_configs.items():
            self.agents[agent_name] = MultimodalAgent(agent_config)
    
    def _init_logging(self):
        """初始化日志系统"""
        return lambda msg: print(msg)
    
    def process_request(self, agent_name: str, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """处理请求"""
        if agent_name not in self.agents:
            return {'error': f'Agent {agent_name} not found'}
        
        try:
            agent = self.agents[agent_name]
            response = agent.run(input_data)
            self.logging(f"Request processed by {agent_name} in {response['processing_time']:.2f}s")
            return response
        except Exception as e:
            self.logging(f"Error processing request: {str(e)}")
            return {'error': str(e)}
    
    def batch_process(self, agent_name: str, requests: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """批量处理请求"""
        results = []
        for request in requests:
            result = self.process_request(agent_name, request)
            results.append(result)
        return results

# 示例使用
if __name__ == "__main__":
    config = {
        'agents': {
            'customer_service': {
                'model_name': 'gemini-1.5-pro',
                'max_context_size': 100
            }
        }
    }
    
    system = EnterpriseMultimodalSystem(config)
    response = system.process_request('customer_service', {
        'text': '我收到的商品有损坏，怎么办？',
        'images': ['damaged_product.jpg']
    })
    print(response)