Image Analysis Pipeline

1from langchain_openai import ChatOpenAI
2from langchain_core.messages import HumanMessage
3from pydantic import BaseModel, Field
4from typing import List, Literal
5import base64
6
7llm = ChatOpenAI(model="gpt-4o-mini")
8
9class ImageClassification(BaseModel):
10    category: str
11    subcategory: str
12    confidence: float = Field(ge=0, le=1)
13    tags: List[str]
14    is_safe: bool
15
16classifier = llm.with_structured_output(ImageClassification)
17
18def classify_image(image_path, categories):
19    with open(image_path, "rb") as f:
20        b64 = base64.b64encode(f.read()).decode()
21    
22    result = classifier.invoke([
23        HumanMessage(content=[
24            {"type": "text", "text": f"Classify image vao categories: {categories}"},
25            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
26        ])
27    ])
28    return result

1class ImageTags(BaseModel):
2    descriptive_tags: List[str] = Field(description="Mo ta noi dung")
3    style_tags: List[str] = Field(description="Style, mood, color")
4    technical_tags: List[str] = Field(description="Camera, lighting")
5    seo_tags: List[str] = Field(description="SEO-friendly tags")
6
7tagger = llm.with_structured_output(ImageTags)
8
9def auto_tag(image_path):
10    with open(image_path, "rb") as f:
11        b64 = base64.b64encode(f.read()).decode()
12    
13    return tagger.invoke([
14        HumanMessage(content=[
15            {"type": "text", "text": "Generate comprehensive tags cho image."},
16            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
17        ])
18    ])

1import asyncio
2from pathlib import Path
3import pandas as pd
4
5async def batch_analyze(image_dir, max_concurrent=5):
6    results = []
7    images = list(Path(image_dir).glob("*.png")) + list(Path(image_dir).glob("*.jpg"))
8    
9    semaphore = asyncio.Semaphore(max_concurrent)
10    
11    async def process_one(img_path):
12        async with semaphore:
13            try:
14                with open(img_path, "rb") as f:
15                    b64 = base64.b64encode(f.read()).decode()
16                
17                result = await classifier.ainvoke([
18                    HumanMessage(content=[
19                        {"type": "text", "text": "Classify and tag this image."},
20                        {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
21                    ])
22                ])
23                return {"file": img_path.name, **result.model_dump()}
24            except Exception as e:
25                return {"file": img_path.name, "error": str(e)}
26    
27    tasks = [process_one(img) for img in images]
28    results = await asyncio.gather(*tasks)
29    
30    return pd.DataFrame(results)
31
32# Run
33df = asyncio.run(batch_analyze("images/"))
34print(df.to_string())

1class QualityAssessment(BaseModel):
2    sharpness: int = Field(ge=1, le=10)
3    composition: int = Field(ge=1, le=10)
4    lighting: int = Field(ge=1, le=10)
5    color_balance: int = Field(ge=1, le=10)
6    overall: int = Field(ge=1, le=10)
7    issues: List[str]
8    suggestions: List[str]
9
10quality_checker = llm.with_structured_output(QualityAssessment)
11
12def check_quality(image_path):
13    with open(image_path, "rb") as f:
14        b64 = base64.b64encode(f.read()).decode()
15    
16    return quality_checker.invoke([
17        HumanMessage(content=[
18            {"type": "text", "text": "Danh gia chat luong anh: sharpness, composition, lighting, color."},
19            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
20        ])
21    ])

1from openai import OpenAI
2
3client = OpenAI()
4
5def get_image_embedding(image_path):
6    # Use CLIP-like embedding
7    with open(image_path, "rb") as f:
8        b64 = base64.b64encode(f.read()).decode()
9    
10    # Describe image -> embed description
11    desc = llm.invoke([
12        HumanMessage(content=[
13            {"type": "text", "text": "Mo ta ngan gon 1 cau."},
14            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
15        ])
16    ])
17    
18    embedding = client.embeddings.create(
19        model="text-embedding-3-small",
20        input=desc.content
21    )
22    return embedding.data[0].embedding
23
24# Compare similarity via cosine distance
25import numpy as np
26
27def cosine_similarity(a, b):
28    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))