Multimodal Pipelines

1from langchain_openai import ChatOpenAI
2from langchain_core.prompts import ChatPromptTemplate
3from langchain_core.output_parsers import StrOutputParser
4from openai import OpenAI
5from pydantic import BaseModel, Field
6from typing import List
7
8llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
9image_client = OpenAI()
10
11class BlogOutline(BaseModel):
12    title: str
13    sections: List[str]
14    image_prompts: List[str] = Field(description="DALL-E prompts cho moi section")
15
16# Step 1: Generate outline voi image prompts
17outline_chain = (
18    ChatPromptTemplate.from_messages([
19        ("system", """Tao blog outline voi 4 sections.
20        Cho moi section, tao image prompt chi tiet cho DALL-E."""),
21        ("human", "Topic: {topic}")
22    ])
23    | llm.with_structured_output(BlogOutline)
24)
25
26# Step 2: Generate text
27write_chain = (
28    ChatPromptTemplate.from_messages([
29        ("system", "Viet section chi tiet, 200 tu."),
30        ("human", "Section: {section}\nContext: {context}")
31    ])
32    | llm
33    | StrOutputParser()
34)
35
36# Step 3: Generate images
37def generate_image(prompt):
38    response = image_client.images.generate(
39        model="dall-e-3",
40        prompt=prompt,
41        size="1792x1024",
42        quality="standard"
43    )
44    return response.data[0].url
45
46# Full pipeline
47async def create_blog_with_images(topic):
48    outline = outline_chain.invoke({"topic": topic})
49    
50    blog = {"title": outline.title, "sections": []}
51    
52    for section, img_prompt in zip(outline.sections, outline.image_prompts):
53        text = write_chain.invoke({
54            "section": section,
55            "context": topic
56        })
57        image_url = generate_image(img_prompt)
58        
59        blog["sections"].append({
60            "heading": section,
61            "text": text,
62            "image": image_url
63        })
64    
65    return blog

1import base64
2
3class ProductListing(BaseModel):
4    title: str
5    description: str
6    features: List[str]
7    seo_tags: List[str]
8
9# Analyze product image -> generate listing
10def create_product_listing(image_path):
11    with open(image_path, "rb") as f:
12        b64 = base64.b64encode(f.read()).decode()
13    
14    # Step 1: Analyze image
15    listing_llm = llm.with_structured_output(ProductListing)
16    
17    listing = listing_llm.invoke([{
18        "role": "user",
19        "content": [
20            {"type": "text", "text": "Tao product listing tu hinh san pham nay."},
21            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
22        ]
23    }])
24    
25    # Step 2: Generate variations
26    variations = image_client.images.generate(
27        model="dall-e-3",
28        prompt=f"Product photo of {listing.title}, white background, studio lighting",
29        n=1,
30        size="1024x1024"
31    )
32    
33    return {
34        "listing": listing,
35        "additional_images": [v.url for v in variations.data]
36    }

1class SocialPack(BaseModel):
2    caption: str
3    hashtags: List[str]
4    image_prompt: str
5
6# Image -> complete social media pack
7def create_social_pack(image_path, platform="instagram"):
8    with open(image_path, "rb") as f:
9        b64 = base64.b64encode(f.read()).decode()
10    
11    pack_llm = llm.with_structured_output(SocialPack)
12    
13    pack = pack_llm.invoke([{
14        "role": "user",
15        "content": [
16            {"type": "text", "text": f"Tao {platform} post tu image. Include caption, hashtags, va suggested image prompt for additional content."},
17            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
18        ]
19    }])
20    
21    return pack

1class AnalysisReport(BaseModel):
2    image_description: str
3    text_summary: str
4    combined_insights: List[str]
5    recommendations: List[str]
6
7def multimodal_report(image_path, text_data):
8    with open(image_path, "rb") as f:
9        b64 = base64.b64encode(f.read()).decode()
10    
11    report_llm = llm.with_structured_output(AnalysisReport)
12    
13    return report_llm.invoke([{
14        "role": "user",
15        "content": [
16            {"type": "text", "text": f"Phan tich ket hop image va text data:\n\n{text_data}\n\nTao insights va recommendations."},
17            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
18        ]
19    }])