多模态 Agent：让 AI 也能「看」世界

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

# 带图像的对话
from langchain.schema import HumanMessage
from langchain_core.messages import HumanMessage
from base64 import b64encode

# 读取图像
with open("image.jpg", "rb") as f:
    image_data = b64encode(f.read()).decode()

# 构建消息
message = HumanMessage(
    content=[
        {"type": "text", "text": "描述这张图片"},
        {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_data}"}}
    ]
)

# 对话
response = llm.invoke([message])
print(response.content)

from transformers import AutoProcessor, LlavaForConditionalGeneration
import torch

# 加载模型
model = LlavaForConditionalGeneration.from_pretrained("llava-hf/llava-1.5-7b-hf")
processor = AutoProcessor.from_pretrained("llava-hf/llava-1.5-7b-hf")

# 图像输入
prompt = "描述这张图片"
image = Image.open("image.jpg")

inputs = processor(prompt, image, return_tensors="pt")
output = model.generate(**inputs)
description = processor.decode(output[0], skip_special_tokens=True)

def describe_image(image_path: str) -> str:
    """图像描述"""
    message = HumanMessage(
        content=[
            {"type": "text", "text": "用一句话描述这张图片"},
            {"type": "image_url", "image_url": {"url": image_path}}
        ]
    )
    return llm.invoke([message]).content

def visual_question_answer(image_path: str, question: str) -> str:
    """视觉问答"""
    message = HumanMessage(
        content=[
            {"type": "text", "text": question},
            {"type": "image_url", "image_url": {"url": image_path}}
        ]
    )
    return llm.invoke([message]).content

# 使用
answer = visual_question_answer(
    "chart.png",
    "这张图表显示的趋势是什么？"
)

def extract_from_image(image_path: str, schema: dict) -> dict:
    """从图像中提取结构化信息"""
    prompt = f"""看这张图片，提取以下信息：
    {schema}
    
    以 JSON 格式返回。"""
    
    message = HumanMessage(
        content=[
            {"type": "text", "text": prompt},
            {"type": "image_url", "image_url": {"url": image_path}}
        ]
    )
    
    # 解析 JSON
    import json
    result = llm.invoke([message]).content
    return json.loads(result)

用户输入（图像/文本）
    ↓
[多模态理解层] → 提取特征/理解内容
    ↓
[Agent 核心] → 理解意图、规划行动
    ↓
[工具层] → 搜索、分析、生成
    ↓
[输出层] → 文本/图像/语音

from langgraph.graph import StateGraph, END
from typing import TypedDict

class MultimodalState(TypedDict):
    user_input: str  # 可能是文本或图像 URL
    input_type: str  # "text" 或 "image"
    understanding: str  # 理解后的内容
    action: str  # 决定的动作
    result: str  # 最终结果

def understand_input(state: MultimodalState) -> MultimodalState:
    """理解用户输入"""
    if state["input_type"] == "image":
        # 理解图像
        understanding = llm.invoke([
            HumanMessage(content=[
                {"type": "text", "text": "详细描述这张图片"},
                {"type": "image_url", "image_url": {"url": state["user_input"]}}
            ])
        ]).content
    else:
        understanding = state["user_input"]
    
    return {"understanding": understanding}

def decide_action(state: MultimodalState) -> MultimodalState:
    """决定动作"""
    prompt = f"""根据用户意图决定下一步动作：
    
    用户输入：{state['understanding']}
    
    可用动作：search, analyze, generate_text, generate_image, answer
    
    返回动作名称："""
    
    action = llm.invoke(prompt).content.strip()
    return {"action": action}

# 构建图
workflow = StateGraph(MultimodalState)
workflow.add_node("understand", understand_input)
workflow.add_node("decide", decide_action)
workflow.set_entry_point("understand")
workflow.add_edge("understand", "decide")
workflow.add_edge("decide", END)

agent = workflow.compile()

from langchain.tools import tool

@tool
def analyze_screenshot(screenshot: str) -> str:
    """
    分析屏幕截图，提取 UI 元素和信息。
    
    Args:
        screenshot: 截图的 URL 或 base64 编码
    """
    prompt = """分析这个截图，列出：
    1. 页面主要元素
    2. 可能的交互点（按钮、输入框等）
    3. 页面类型和目的"""
    
    return llm.invoke([
        HumanMessage(content=[
            {"type": "text", "text": prompt},
            {"type": "image_url", "image_url": {"url": screenshot}}
        ])
    ]).content

@tool
def read_document_image(image_path: str) -> str:
    """
    读取文档图片，提取文字。
    
    Args:
        image_path: 文档图片路径
    """
    # 使用 OCR + LLM
    import pytesseract
    text = pytesseract.image_to_string(image_path)
    return text

@tool
def describe_chart(chart_image: str) -> dict:
    """
    分析图表，提取数据 insights。
    
    Args:
        chart_image: 图表图片
    """
    prompt = """分析这个图表：
    1. 图表类型
    2. 主要数据趋势
    3. 关键数据点
    4. 可能的 insights"""
    
    result = llm.invoke([
        HumanMessage(content=[
            {"type": "text", "text": prompt},
            {"type": "image_url", "image_url": {"url": chart_image}}
        ])
    ]).content
    
    return {"analysis": result}

from langchain_community.retrievers import TavilySearchAPIRetriever
from langchain_openai import OpenAIEmbeddings

# CLIP 模型
from transformers import CLIPProcessor, CLIPModel

model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

def search_images_by_text(query: str, image_paths: list[str], top_k: int = 3) -> list:
    """文本搜图像"""
    # 编码文本
    text_inputs = processor(text=[query], return_tensors="pt", padding=True)
    text_features = model.get_text_features(**text_inputs)
    
    # 编码图像
    images = [Image.open(path) for path in image_paths]
    image_inputs = processor(images=images, return_tensors="pt", padding=True)
    image_features = model.get_image_features(**image_inputs)
    
    # 计算相似度
    similarities = (text_features @ image_features.T).squeeze()
    
    # 排序
    top_indices = similarities.argsort(descending()[:top_k]
    
    return [(image_paths[i], similarities[i].item()) for i in top_indices]

def search_similar_images(query_image: str, image_paths: list[str], top_k: int = 5) -> list:
    """图像搜图像"""
    # 编码查询图像
    query_img = Image.open(query_image)
    query_inputs = processor(images=[query_img], return_tensors="pt")
    query_features = model.get_image_features(**query_inputs)
    
    # 编码候选图像
    candidate_imgs = [Image.open(path) for path in image_paths]
    candidate_inputs = processor(images=candidate_imgs, return_tensors="pt", padding=True)
    candidate_features = model.get_image_features(**candidate_inputs)
    
    # 计算相似度
    similarities = (query_features @ candidate_features.T).squeeze()
    
    top_indices = similarities.argsort(descending()[:top_k]
    
    return [(image_paths[i], similarities[i].item()) for i in top_indices]

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

# 使用 DALL-E 3 生成图像
response = llm.invoke([
    HumanMessage(content=[
        {"type": "text", "text": "生成一张图片：一只可爱的橘猫坐在窗台上"},
        {"type": "image_url", "image_url": {"url": "https://api.openai.com/v1/images/generation"}}
    ])
])

# 注意：实际使用 openai.images.generate
from openai import OpenAI
client = OpenAI()

image_response = client.images.generate(
    model="dall-e-3",
    prompt="一只可爱的橘猫坐在窗台上，阳光从窗外洒进来",
    size="1024x1024",
    quality="standard"
)

image_url = image_response.data[0].url

# 使用 DALL-E 2 进行图像编辑
image_response = client.images.edit(
    image=open("original.png", "rb"),
    prompt="把背景改成冬天雪景",
    n=1,
    size="1024x1024"
)

@tool
def analyze_video(video_path: str, time_ranges: list[tuple] = None) -> str:
    """
    分析视频内容。
    
    Args:
        video_path: 视频文件路径
        time_ranges: 要分析的时间段 [(start, end), ...]
    """
    # 提取关键帧
    import cv2
    
    cap = cv2.VideoCapture(video_path)
    frames = []
    
    if time_ranges:
        for start, end in time_ranges:
            cap.set(cv2.CAP_PROP_POS_MSEC, start * 1000)
            ret, frame = cap.read()
            if ret:
                frames.append(frame)
    else:
        # 均匀采样
        total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
        for i in range(0, total_frames, total_frames // 5):
            cap.set(cv2.CAP_PROP_POS_FRAMES, i)
            ret, frame = cap.read()
            if ret:
                frames.append(frame)
    
    # 分析每帧
    descriptions = []
    for frame in frames:
        # 转 base64
        import base64
        _, buffer = cv2.imencode('.jpg', frame)
        b64 = base64.b64encode(buffer).decode()
        
        description = llm.invoke([
            HumanMessage(content=[
                {"type": "text", "text": "简短视频这一帧的内容"},
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
            ])
        ]).content
        descriptions.append(description)
    
    return "\n".join(descriptions)

def analyze_customer_screenshot(screenshot: str, question: str) -> str:
    """分析用户发送的截图，回答问题"""
    # 1. 理解截图内容
    screen_description = llm.invoke([
        HumanMessage(content=[
            {"type": "text", "text": "详细描述这个界面"},
            {"type": "image_url", "image_url": {"url": screenshot}}
        ])
    ]).content
    
    # 2. 结合问题回答
    answer = llm.invoke(f"""截图描述：{screen_description}
    
    用户问题：{question}
    
    基于截图回答用户问题：""")
    
    return answer.content

def process_document(document: str) -> dict:
    """处理各类文档（PDF、图片）"""
    # 判断文档类型
    if document.endswith(('.png', '.jpg', '.jpeg')):
        # 提取文字
        text = extract_text_from_image(document)
    else:
        # PDF 处理
        text = extract_text_from_pdf(document)
    
    # 分析内容
    analysis = llm.invoke(f"""分析这个文档：
    {text}
    
    提取：标题、关键信息、总结""")
    
    return {"text": text, "analysis": analysis.content}

def summarize_video(video_path: str) -> str:
    """生成视频摘要"""
    # 提取关键帧
    key_frames = extract_key_frames(video_path)
    
    # 描述每帧
    frame_descriptions = []
    for frame in key_frames:
        desc = llm.invoke([
            HumanMessage(content=[
                {"type": "text", "text": "简洁描述"},
                {"type": "image_url", "image_url": {"url": frame}}
            ])
        ]).content
        frame_descriptions.append(desc)
    
    # 生成摘要
    summary = llm.invoke(f"""根据以下视频关键帧描述，生成视频摘要：
    
    {" -> ".join(frame_descriptions)}""")
    
    return summary.content