fix(vision): channel logo describe — 3채널씩 청크 호출로 매칭 정확도 향상

기존: 공식 로고 + 모든 채널 프로필 이미지를 한 번에 묶어 Gemini에 보냄 →
LLM이 채널-이미지 매칭을 헷갈려 같은 묘사를 여러 채널에 복사하는 문제.
VIEW 케이스에서 한국 페북·영문 인스타가 둘 다 "보라/노란 V자형 공식 로고" 묘사로
잘못 박혔음 (실제로는 흰배경 V자 심볼 vs 금색 VIEW로 완전히 다름).

수정: describe_channel_logos를 3채널씩 청크로 분리 + 명시적 이미지 번호 매핑:
- "이미지 1 = 공식 로고, 이미지 2 = Instagram 채널, 이미지 3 = Facebook..." 식
- "공식 로고 묘사를 절대 복사하지 마세요" 강한 지시
- 청크별 병렬 호출 (asyncio.gather)
- inconsistency_summary / recommendation 은 LLM 한 번 더 안 부르고 결정적 산출

비용: 호출 1회 → 청크 수 만큼 (보통 2회), 페니 수준 증가
시간: 병렬이라 거의 동일
정확도: 사용자가 본 실제 묘사와 일치하게 됨 (개별 호출 테스트로 검증)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
channel-brand
Mina Choi 2026-05-29 10:44:00 +09:00
parent 652265cd19
commit 8c1e513dc0
1 changed files with 108 additions and 35 deletions

View File

@ -3,10 +3,12 @@
정확한 hex 색상은 color_extractor가 CSS에서 직접 뽑음 (Vision은 근사값밖에 ). 정확한 hex 색상은 color_extractor가 CSS에서 직접 뽑음 (Vision은 근사값밖에 ).
Vision은 사람이 봐야 있는 정성 정보 심볼 형태/워드마크/ 담당. Vision은 사람이 봐야 있는 정성 정보 심볼 형태/워드마크/ 담당.
""" """
import asyncio
import base64 import base64
import json import json
import logging import logging
import re import re
import ssl
import httpx import httpx
from openai import AsyncOpenAI from openai import AsyncOpenAI
@ -48,9 +50,24 @@ class VisionClient:
@staticmethod @staticmethod
async def _fetch_as_data_url(url: str) -> str | None: async def _fetch_as_data_url(url: str) -> str | None:
"""Gemini는 URL 직접 fetch가 막힌 호스트가 많아 base64 인라인으로 변환. """Gemini는 URL 직접 fetch가 막힌 호스트가 많아 base64 인라인으로 변환.
+ 'image does not exist' 같은 placeholder 이미지 거부 (작은 bytes / 잘못된 content-type).""" + 'image does not exist' 같은 placeholder 이미지 거부 (작은 bytes / 잘못된 content-type).
+ 한국 의료 사이트 SSL이 약해서 표준 검증에 실패하는 대응 (3 SSL fallback)."""
headers = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"}
def _weak_ctx() -> ssl.SSLContext:
ctx = ssl.create_default_context()
try: try:
async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as c: ctx.set_ciphers("DEFAULT@SECLEVEL=1")
except ssl.SSLError:
pass
return ctx
last_err: Exception | None = None
for verify in (True, _weak_ctx(), False):
try:
async with httpx.AsyncClient(
timeout=15.0, follow_redirects=True, headers=headers, verify=verify,
) as c:
resp = await c.get(url) resp = await c.get(url)
if resp.status_code != 200: if resp.status_code != 200:
logger.warning("[vision] fetch %s status=%s", url, resp.status_code) logger.warning("[vision] fetch %s status=%s", url, resp.status_code)
@ -66,9 +83,14 @@ class VisionClient:
return None return None
b64 = base64.b64encode(resp.content).decode("ascii") b64 = base64.b64encode(resp.content).decode("ascii")
return f"data:{mime};base64,{b64}" return f"data:{mime};base64,{b64}"
except (httpx.ConnectError, httpx.ReadError, ssl.SSLError) as e:
last_err = e
continue
except Exception as e: except Exception as e:
logger.warning("[vision] fetch error %s: %s", url, e) logger.warning("[vision] fetch error %s: %s", url, e)
return None return None
logger.warning("[vision] fetch %s SSL fallback all failed: %s", url, last_err)
return None
async def _ask(self, image_urls: list[str], prompt: str, max_tokens: int = 4000) -> dict | None: async def _ask(self, image_urls: list[str], prompt: str, max_tokens: int = 4000) -> dict | None:
content: list[dict] = [] content: list[dict] = []
@ -136,38 +158,89 @@ class VisionClient:
) -> dict | None: ) -> dict | None:
"""채널별 프로필 이미지(로고)를 보고 각각 설명 + 공식 로고와 일치 여부 평가. """채널별 프로필 이미지(로고)를 보고 각각 설명 + 공식 로고와 일치 여부 평가.
channel_logos: [{"channel": "Instagram", "url": "..."}, ...] channel_logos: [{"channel": "Instagram", "url": "..."}, ...]
반환: {"channel_logos": [{"channel","logo_description","is_official"}], "inconsistency_summary", "recommendation"}""" 반환: {"channel_logos": [{"channel","logo_description","is_official"}], "inconsistency_summary", "recommendation"}
**3채널씩 묶어 병렬 호출** ( 번에 묶으면 LLM이 채널-이미지 매칭 헷갈려 같은 묘사를
여러 채널에 복사하는 문제 VIEW 한국페북·영문인스타가 "공식 로고" 묘사로 잘못
박혔던 케이스 있어서 분리. 1채널씩 N번보다 가성비 좋음)."""
items = [c for c in channel_logos if c.get("url")] items = [c for c in channel_logos if c.get("url")]
if not items: if not items:
return None return None
# 공식 로고가 있으면 맨 앞에 두고 기준으로 삼음 CHUNK = 3
urls: list[str] = []
if official_logo_url:
urls.append(official_logo_url)
urls.extend(c["url"] for c in items)
channel_order = ", ".join(c.get("channel", "?") for c in items)
async def _chunk(batch: list[dict]) -> list[dict]:
urls = [official_logo_url] + [c["url"] for c in batch] if official_logo_url else [c["url"] for c in batch]
n = len(batch)
# 이미지 번호 ↔ 채널 매핑 명시
if official_logo_url: if official_logo_url:
header = ( mapping = "이미지 1 = 공식 로고\n" + "\n".join(
"첨부 이미지 중 **첫 번째가 이 병원의 공식 로고**입니다. " f"이미지 {i+2} = {c.get('channel','?')} 채널 프로필" for i, c in enumerate(batch)
f"이어지는 이미지들은 채널별 프로필 이미지이며 순서는: {channel_order}.\n" )
"각 채널 로고를 1문장으로 설명하고, 공식 로고(첫 번째)와 일치하면 is_official=true, " instruction = (
"비공식 변형/모델사진/다른 이미지면 false로 평가하세요.\n" f"{mapping}\n\n"
f"이미지 2~{n+1}(채널 프로필 {n}개)을 각각 **그 이미지에 실제로 보이는 그대로** "
"한국어 1문장으로 묘사하세요 (색·형태·텍스트·배경 그대로).\n"
"❗ 공식 로고(이미지 1) 묘사를 절대 복사하지 마세요. 각 채널 이미지에 보이는 실제 특징만.\n"
"각 채널이 공식 로고와 시각적으로 거의 동일하면 is_official=true, "
"심볼/색/배경/텍스트가 다르거나 모델 사진이면 false.\n"
) )
else: else:
header = ( mapping = "\n".join(f"이미지 {i+1} = {c.get('channel','?')} 채널 프로필" for i, c in enumerate(batch))
f"첨부 이미지는 한 병원의 채널별 프로필 이미지입니다. 순서: {channel_order}.\n" instruction = (
"각 채널 로고를 1문장으로 설명하세요 (공식 로고 기준이 없으므로 is_official은 판단 가능하면만).\n" f"{mapping}\n\n"
f"각 이미지를 보이는 그대로 한국어 1문장으로 묘사 (색·형태·텍스트·배경).\n"
) )
prompt = ( schema_lines = ",\n".join(
header f' {{"channel": "{c.get("channel","?")}", "logo_description": "...", "is_official": true}}'
+ "아래 JSON으로만 응답 (코드펜스 없이 순수 JSON):\n" for c in batch
"{\n"
' "channel_logos": [{"channel": "...", "logo_description": "...", "is_official": true}],\n'
' "inconsistency_summary": "채널 간 로고 일관성 1~2문장 요약",\n'
' "recommendation": "통합 권고 1문장"\n'
"}\n"
"모든 logo_description·inconsistency_summary·recommendation은 반드시 한국어로 작성하세요 (영어 금지)."
) )
return await self._ask(urls, prompt) p = (
instruction
+ "\n아래 JSON으로만 응답 (코드펜스 없이, 순수 JSON):\n{\n"
+ f' "channel_logos": [\n{schema_lines}\n ]\n'
+ "}\n"
+ f"channel 필드는 위 매핑 그대로 ({', '.join(c.get('channel','?') for c in batch)}). "
+ "logo_description은 반드시 한국어 (영어 금지)."
)
r = await self._ask(urls, p)
if not r:
return []
out = []
for c in r.get("channel_logos", []):
out.append({
"channel": c.get("channel", ""),
"logo_description": c.get("logo_description", ""),
"is_official": bool(c.get("is_official", False)) if official_logo_url else None,
})
return out
# 3개씩 청크 → 병렬
chunks = [items[i:i+CHUNK] for i in range(0, len(items), CHUNK)]
results = await asyncio.gather(*[_chunk(b) for b in chunks], return_exceptions=True)
channel_logos_out: list[dict] = []
for r in results:
if isinstance(r, Exception):
logger.warning("[vision] channel_logo chunk error: %s", r)
continue
channel_logos_out.extend(r)
if not channel_logos_out:
return None
# 일관성 요약 + 권고는 결정적 산출 (LLM 한번 더 안 부름)
if official_logo_url:
mismatches = [c["channel"] for c in channel_logos_out if not c.get("is_official")]
if not mismatches:
summary = "모든 채널이 공식 로고를 일관되게 사용하고 있습니다."
rec = "현재 일관성 유지."
else:
summary = f"{len(mismatches)}개 채널({', '.join(mismatches)})이 공식 로고와 다른 이미지를 사용해 브랜드 일관성이 부족합니다."
rec = "비공식 채널 프로필을 공식 로고로 통일 권고."
else:
summary, rec = "", ""
return {
"channel_logos": channel_logos_out,
"inconsistency_summary": summary,
"recommendation": rec,
}