9장: 멀티모달 에이전트 구축

import anthropic
 
client = anthropic.Anthropic()
 
# Computer Use 도구 정의
tools = [
    {
        "type": "computer_20250124",
        "name": "computer",
        "display_width_px": 1920,
        "display_height_px": 1080,
        "display_number": 1,
    },
    {
        "type": "text_editor_20250124",
        "name": "str_replace_editor",
    },
    {
        "type": "bash_20250124",
        "name": "bash",
    },
]
 
# 에이전트 루프
messages = [{
    "role": "user",
    "content": "Chrome 브라우저를 열고 'AI agent patterns'를 검색해주세요.",
}]
 
while True:
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        tools=tools,
        messages=messages,
    )
 
    # 도구 호출이 없으면 종료
    if response.stop_reason == "end_turn":
        break
 
    # 도구 호출 처리
    for block in response.content:
        if block.type == "tool_use":
            # 도구 실행 (스크린샷 캡처, 클릭, 타이핑 등)
            result = execute_tool(block)
 
            # 결과를 메시지에 추가
            messages.append({"role": "assistant", "content": response.content})
            messages.append({
                "role": "user",
                "content": [{
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                }],
            })

def execute_computer_action(action: dict) -> dict:
    """Computer Use 행동 실행"""
    action_type = action["action"]
 
    if action_type == "screenshot":
        # 현재 화면 캡처
        screenshot = capture_screenshot()
        return {
            "type": "image",
            "source": {
                "type": "base64",
                "media_type": "image/png",
                "data": encode_screenshot(screenshot),
            },
        }
 
    elif action_type == "click":
        # 마우스 클릭
        x, y = action["coordinate"]
        pyautogui.click(x, y)
        return {"type": "text", "text": f"Clicked at ({x}, {y})"}
 
    elif action_type == "type":
        # 텍스트 입력
        pyautogui.typewrite(action["text"], interval=0.05)
        return {"type": "text", "text": f"Typed: {action['text']}"}
 
    elif action_type == "key":
        # 키 조합
        pyautogui.hotkey(*action["key"].split("+"))
        return {"type": "text", "text": f"Pressed: {action['key']}"}
 
    elif action_type == "scroll":
        # 스크롤
        direction = action["coordinate"][1]
        pyautogui.scroll(direction)
        return {"type": "text", "text": f"Scrolled: {direction}"}

tools = [
    {
        "name": "capture_screenshot",
        "description": "현재 화면의 스크린샷을 캡처합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "region": {
                    "type": "string",
                    "description": "캡처 영역: full, active_window, 또는 x,y,w,h 좌표",
                },
            },
        },
    },
    {
        "name": "analyze_image",
        "description": "이미지를 분석하여 내용을 설명합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "image_path": {"type": "string"},
                "question": {"type": "string"},
            },
            "required": ["image_path"],
        },
    },
    {
        "name": "extract_text_from_image",
        "description": "이미지에서 텍스트를 추출합니다 (OCR).",
        "input_schema": {
            "type": "object",
            "properties": {
                "image_path": {"type": "string"},
                "language": {"type": "string", "default": "ko"},
            },
            "required": ["image_path"],
        },
    },
    {
        "name": "search_similar_images",
        "description": "텍스트 설명으로 유사한 이미지를 검색합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "top_k": {"type": "integer", "default": 5},
            },
            "required": ["query"],
        },
    },
]

class MultimodalAgent:
    def __init__(self, client, tools, max_iterations: int = 10):
        self.client = client
        self.tools = tools
        self.max_iterations = max_iterations
        self.tool_handlers = {}
 
    def register_handler(self, name: str, handler):
        self.tool_handlers[name] = handler
 
    async def run(self, user_message: str, images: list[dict] | None = None) -> str:
        """에이전트 실행"""
        # 초기 메시지 구성
        content = []
        if images:
            content.extend(images)
        content.append({"type": "text", "text": user_message})
 
        messages = [{"role": "user", "content": content}]
 
        for _ in range(self.max_iterations):
            response = self.client.messages.create(
                model="claude-sonnet-4-20250514",
                max_tokens=4096,
                tools=self.tools,
                messages=messages,
            )
 
            # 최종 응답
            if response.stop_reason == "end_turn":
                return self._extract_text(response.content)
 
            # 도구 호출 처리
            messages.append({"role": "assistant", "content": response.content})
            tool_results = []
 
            for block in response.content:
                if block.type == "tool_use":
                    handler = self.tool_handlers.get(block.name)
                    if handler:
                        result = await handler(block.input)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": result,
                        })
 
            messages.append({"role": "user", "content": tool_results})
 
        return "최대 반복 횟수에 도달했습니다."
 
    def _extract_text(self, content) -> str:
        return "\n".join(
            block.text for block in content if hasattr(block, "text")
        )

# 사용자가 이미지를 업로드하고 질문하면,
# 에이전트가 이미지를 분석하고 추가 도구를 활용하여 답변
 
agent = MultimodalAgent(client, tools)
 
# 도구 핸들러 등록
agent.register_handler("analyze_image", analyze_image_handler)
agent.register_handler("search_similar_images", search_handler)
agent.register_handler("extract_text_from_image", ocr_handler)
 
# 이미지와 함께 질문
result = await agent.run(
    user_message="이 아키텍처 다이어그램에서 데이터베이스 병목 지점을 찾고, AWS에서 사용할 수 있는 대안 서비스를 추천해주세요.",
    images=[{
        "type": "image",
        "source": {"type": "base64", "media_type": "image/png", "data": diagram_b64},
    }],
)

document_tools = [
    {"name": "extract_tables", "description": "PDF에서 표를 추출합니다."},
    {"name": "extract_charts", "description": "PDF에서 차트 데이터를 추출합니다."},
    {"name": "search_document", "description": "문서 내용을 검색합니다."},
    {"name": "compare_documents", "description": "두 문서를 비교합니다."},
]
 
# 사용 예시
result = await agent.run(
    "이 두 분기 보고서를 비교하여 매출 변화를 분석하고, "
    "가장 큰 변화가 있는 부문을 차트 데이터와 함께 설명해주세요.",
    images=[q3_report_image, q4_report_image],
)

web_tools = [
    {"name": "navigate", "description": "URL로 이동합니다."},
    {"name": "screenshot", "description": "현재 페이지 스크린샷을 캡처합니다."},
    {"name": "click_element", "description": "페이지의 특정 요소를 클릭합니다."},
    {"name": "fill_form", "description": "폼 필드에 값을 입력합니다."},
    {"name": "extract_content", "description": "페이지 콘텐츠를 추출합니다."},
]
 
# 에이전트가 스크린샷을 보고 다음 행동을 결정
# 1. 페이지 로드 → 스크린샷
# 2. 스크린샷 분석 → 클릭할 요소 결정
# 3. 클릭 → 새 스크린샷
# 4. 반복...

class AgentGuardrails:
    def __init__(self):
        self.allowed_domains = ["*.example.com"]
        self.blocked_actions = ["delete", "format", "shutdown"]
        self.max_cost_per_run = 1.0  # USD
 
    def validate_action(self, action: dict) -> bool:
        """행동 사전 검증"""
        # URL 허용 목록 확인
        if "url" in action:
            if not self._is_allowed_url(action["url"]):
                return False
 
        # 위험한 행동 차단
        if action.get("type") in self.blocked_actions:
            return False
 
        return True
 
    def validate_visual_output(self, screenshot: bytes) -> bool:
        """스크린샷에 민감 정보가 포함되어 있는지 확인"""
        # 개인정보, 비밀번호 필드 등 감지
        return True
 
    def check_cost(self, current_cost: float) -> bool:
        """비용 한도 확인"""
        return current_cost < self.max_cost_per_run