2026年マルチモーダルAI Agent入門完全ガイド

# 仮想環境の作成
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# コア依存関係のインストール
pip install torch>=2.1.0
pip install transformers>=4.36.0
pip install langchain>=0.1.0
pip install anthropic>=0.18.0
pip install openai>=1.12.0
pip install python-dotenv>=1.0.0

from typing import List, Dict, Union
from dataclasses import dataclass
from enum import Enum

class MessageType(Enum):
    TEXT = "text"
    IMAGE = "image"
    AUDIO = "audio"

@dataclass
class Message:
    type: MessageType
    content: str
    metadata: Dict = None

class MultimodalAgent:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.conversation_history: List[Message] = []

    async def process(self, message: Message) -> str:
        """ユーザーメッセージを処理して応答を返す"""
        self.conversation_history.append(message)

        if message.type == MessageType.IMAGE:
            return await self._process_image(message)
        elif message.type == MessageType.TEXT:
            return await self._process_text(message)
        else:
            return "Unsupported message type"

    async def _process_text(self, message: Message) -> str:
        """テキストメッセージを処理"""
        response = await self._call_llm(message.content)
        return response

    async def _process_image(self, message: Message) -> str:
        """画像メッセージを処理"""
        description = await self._analyze_image(message.content)
        return f"画像分析結果: {description}"

agent = MultimodalAgent(api_key="your-api-key")

import base64
from io import BytesIO
from PIL import Image

class VisionAgent(MultimodalAgent):
    def __init__(self, api_key: str):
        super().__init__(api_key)
        self.client = OpenAI()

    async def _analyze_image(self, image_path: str) -> str:
        """GPT-4 Visionを使用して画像を分析"""
        with Image.open(image_path) as img:
            max_size = (2048, 2048)
            img.thumbnail(max_size, Image.Resampling.LANCZOS)

            buffered = BytesIO()
            img.save(buffered, format="PNG")
            img_str = base64.b64encode(buffered.getvalue()).decode()

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": [
                    {"type": "text", "text": "この画像の内容を説明してください。"},
                    {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_str}"}}
                ]
            }]
        )
        return response.choices[0].message.content

from typing import Callable, Any
import json

class Tool:
    def __init__(self, name: str, description: str, func: Callable):
        self.name = name
        self.description = description
        self.func = func

    def to_dict(self) -> Dict:
        return {
            "name": self.name,
            "description": self.description
        }

class ToolCallingAgent(VisionAgent):
    def __init__(self, api_key: str):
        super().__init__(api_key)
        self.tools: List[Tool] = []
        self._register_default_tools()

    def _register_default_tools(self):
        """デフォルトツールを登録"""
        self.register_tool(Tool(
            name="web_search",
            description="インターネット検索で最新情報を取得",
            func=self._web_search
        ))
        self.register_tool(Tool(
            name="calculator",
            description="数学計算を実行",
            func=self._calculator
        ))

    def register_tool(self, tool: Tool):
        self.tools.append(tool)

    async def execute_with_tools(self, user_message: str) -> str:
        """ツールを使用してタスクを実行"""
        tools_json = json.dumps([t.to_dict() for t in self.tools])

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": user_message}],
            tools=[{"type": "function", "function": t.to_dict()} for t in self.tools]
        )

        tool_calls = response.choices[0].message.tool_calls
        if tool_calls:
            for call in tool_calls:
                tool_name = call.function.name
                args = json.loads(call.function.arguments)
                result = await self._execute_tool(tool_name, args)
                return result

        return response.choices[0].message.content

from datetime import datetime
from typing import List, Tuple

class Memory:
    def __init__(self, max_size: int = 100):
        self.short_term: List[Tuple[str, datetime]] = []
        self.long_term: List[str] = []
        self.max_size = max_size

    def add(self, content: str, memory_type: str = "short"):
        """記憶を追加"""
        if memory_type == "short":
            self.short_term.append((content, datetime.now()))
            if len(self.short_term) > self.max_size:
                oldest = self.short_term.pop(0)
                self.long_term.append(oldest[0])
        else:
            self.long_term.append(content)

    def get_recent(self, n: int = 5) -> List[str]:
        """最近の記憶を取得"""
        return [item[0] for item in self.short_term[-n:]]

    def get_context(self) -> str:
        """すべての記憶をコンテキストとして取得"""
        context = "Recent memories:\n"
        context += "\n".join(self.get_recent(5))
        if self.long_term:
            context += "\n\nImportant memories:\n"
            context += "\n".join(self.long_term[-5:])
        return context

# FastAPIを使用してAgentをデプロイ
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn

app = FastAPI(title="Multimodal Agent API")

class ChatRequest(BaseModel):
    message: str
    image_url: str = None

@app.post("/api/chat")
async def chat(request: ChatRequest):
    try:
        agent = AgentWithMemory(api_key=settings.OPENAI_API_KEY)
        response = await agent.chat(request.message)
        return {"response": response}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    return {"status": "healthy"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)