Claude Sonnet 4.5

AI 기반 시맨틱 검색 엔진 구축하기: 하이브리드 검색 완전 가이드

## 키워드 검색 vs 시맨틱 검색 | 구분 | 키워드(BM25) | 시맨틱(Vector) | 하이브리드 | |------|-------------|---------------|-----------| | 검색 방식 | 단어 일치 | 의미 유사도 | 둘 다 | | "파이썬 배우기" → "Python 강좌" | ❌ 불일치 | ✅ 유사 | ✅ | | 정확한 키워드 검색 | ✅ 우수 | ❌ 약함 | ✅ | | 오타 처리 | ❌ 약함 | ✅ 강함 | ✅ | | 속도 | 매우 빠름 | 느림 | 중간 | --- ## 아키텍처 ```mermaid flowchart TD Query[사용자 쿼리] --> Parallel{병렬 검색} Parallel --> BM25[BM25 키워드 검색<br/>Elasticsearch] Parallel --> Vec[벡터 시맨틱 검색<br/>pgvector] BM25 --> Merge[결과 병합<br/>RRF 알고리즘] Vec --> Merge Merge --> Rerank[Cohere Rerank<br/>재정렬] Rerank --> Results[최종 검색 결과] ``` --- ## pgvector + BM25 하이브리드 검색 ```python import psycopg2 from openai import OpenAI from rank_bm25 import BM25Okapi import numpy as np client = OpenAI() class HybridSearchEngine: def __init__(self, conn): self.conn = conn self.documents = [] self.bm25 = None def add_documents(self, docs: list[dict]): self.documents = docs # BM25 인덱스 구축 tokenized = [doc["content"].split() for doc in docs] self.bm25 = BM25Okapi(tokenized) # 벡터 임베딩 저장 contents = [doc["content"] for doc in docs] response = client.embeddings.create( input=contents, model="text-embedding-3-small" ) with self.conn.cursor() as cur: for i, (doc, emb_data) in enumerate(zip(docs, response.data)): cur.execute( '''INSERT INTO search_index (id, content, metadata, embedding) VALUES (%s, %s, %s, %s) ON CONFLICT (id) DO UPDATE SET content=EXCLUDED.content, embedding=EXCLUDED.embedding''', (doc["id"], doc["content"], doc.get("metadata", {}), emb_data.embedding) ) self.conn.commit() def search(self, query: str, top_k: int = 10, bm25_weight: float = 0.3) -> list: # 1. BM25 점수 bm25_scores = self.bm25.get_scores(query.split()) bm25_scores = (bm25_scores - bm25_scores.min()) / (bm25_scores.max() - bm25_scores.min() + 1e-8) # 2. 벡터 유사도 query_vec = client.embeddings.create( input=[query], model="text-embedding-3-small" ).data[0].embedding with self.conn.cursor() as cur: cur.execute( '''SELECT id, 1 - (embedding <=> %s::vector) AS score FROM search_index ORDER BY embedding <=> %s::vector LIMIT %s''', (query_vec, query_vec, top_k * 2) ) vec_results = {row[0]: row[1] for row in cur.fetchall()} # 3. 점수 융합 (가중 합산) combined = [] for i, doc in enumerate(self.documents): vec_score = vec_results.get(doc["id"], 0) combined_score = bm25_weight * bm25_scores[i] + (1 - bm25_weight) * vec_score combined.append((doc, combined_score)) combined.sort(key=lambda x: x[1], reverse=True) return combined[:top_k] ``` --- ## RRF (Reciprocal Rank Fusion) 두 결과 목록을 더 공정하게 합치는 알고리즘: ```python def reciprocal_rank_fusion(rankings: list[list], k: int = 60) -> list: ''' rankings: 각 검색 방법의 문서 ID 순위 리스트 k: 완화 상수 (기본값 60) ''' scores = {} for ranking in rankings: for rank, doc_id in enumerate(ranking): scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1) return sorted(scores.items(), key=lambda x: x[1], reverse=True) # 사용 bm25_ranking = ["doc3", "doc1", "doc7", "doc2"] vec_ranking = ["doc1", "doc5", "doc3", "doc8"] fused = reciprocal_rank_fusion([bm25_ranking, vec_ranking]) print(fused[:3]) # [('doc3', 0.0317...), ('doc1', 0.0313...), ...] ``` --- ## Typesense: 빠른 오픈소스 대안 ```python import typesense client = typesense.Client({ "api_key": "your-api-key", "nodes": [{"host": "localhost", "port": "8108", "protocol": "http"}], "connection_timeout_seconds": 2 }) # 스키마 생성 client.collections.create({ "name": "articles", "fields": [ {"name": "id", "type": "string"}, {"name": "title", "type": "string"}, {"name": "content", "type": "string"}, {"name": "embedding", "type": "float[]", "num_dim": 1536} # 벡터 필드 ] }) # 하이브리드 검색 (키워드 + 벡터 동시) results = client.collections["articles"].documents.search({ "q": "파이썬 머신러닝", "query_by": "title,content", "vector_query": f"embedding:([{','.join(map(str, query_embedding))}], k:10)", "hybrid_search_weight": 0.7, # 벡터 검색 가중치 "per_page": 10 }) ``` --- ## Cohere Rerank으로 최종 정렬 ```python import cohere co = cohere.Client("your-api-key") def rerank_results(query: str, documents: list[str], top_n: int = 5) -> list: results = co.rerank( model="rerank-multilingual-v3.0", query=query, documents=documents, top_n=top_n, ) return [(results.results[i].index, results.results[i].relevance_score) for i in range(len(results.results))] # 하이브리드 검색 후 리랭킹 candidates = hybrid_search(query, top_k=20) doc_texts = [c["content"] for c, _ in candidates] reranked = rerank_results(query, doc_texts, top_n=5) final_results = [candidates[idx] for idx, score in reranked] ``` --- ## 검색 품질 평가 ```python # NDCG (Normalized Discounted Cumulative Gain) 계산 def ndcg_at_k(relevance_scores: list, k: int) -> float: dcg = sum(rel / np.log2(i + 2) for i, rel in enumerate(relevance_scores[:k])) ideal = sum(s / np.log2(i + 2) for i, s in enumerate(sorted(relevance_scores, reverse=True)[:k])) return dcg / ideal if ideal > 0 else 0 # A/B 테스트 프레임워크 results_a = keyword_search(query) results_b = hybrid_search(query) ndcg_a = ndcg_at_k([1 if doc in relevant_docs else 0 for doc, _ in results_a], k=5) ndcg_b = ndcg_at_k([1 if doc in relevant_docs else 0 for doc, _ in results_b], k=5) print(f"키워드 검색 NDCG@5: {ndcg_a:.4f}") print(f"하이브리드 검색 NDCG@5: {ndcg_b:.4f}") ``` 실제 서비스에서는 클릭 데이터, 체류 시간, 전환율을 오프라인 지표와 함께 측정하세요.

멀티모달 AI 개발 가이드: 이미지+텍스트 분석 실전 구현

## 멀티모달 AI란? 멀티모달 AI는 텍스트뿐 아니라 이미지, 오디오, 비디오 등 여러 형태의 데이터를 동시에 처리합니다. 이미지를 설명하거나, 차트를 해석하거나, 문서에서 정보를 추출하는 작업이 가능합니다. ```mermaid flowchart LR Image["이미지/PDF"] --> Encode[인코딩<br/>Base64/URL] Text["텍스트 질문"] --> API[멀티모달 LLM API] Encode --> API API --> GPT["GPT-4o<br/>Vision"] API --> Claude["Claude 3.5<br/>Sonnet"] API --> Gemini["Gemini 2.5<br/>Pro"] GPT --> Result["분석 결과<br/>텍스트"] Claude --> Result Gemini --> Result ``` --- ## GPT-4o Vision API ```python import base64 from openai import OpenAI from pathlib import Path client = OpenAI() def encode_image(image_path: str) -> str: with open(image_path, "rb") as f: return base64.b64encode(f.read()).decode("utf-8") def analyze_image(image_path: str, question: str) -> str: base64_image = encode_image(image_path) ext = Path(image_path).suffix.lower() media_type = {"jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".gif": "image/gif", ".webp": "image/webp"}.get(ext, "image/jpeg") response = client.chat.completions.create( model="gpt-4o", messages=[{ "role": "user", "content": [ { "type": "image_url", "image_url": { "url": f"data:{media_type};base64,{base64_image}", "detail": "high" # low/high/auto } }, {"type": "text", "text": question} ] }], max_tokens=1024 ) return response.choices[0].message.content # URL로 직접 이미지 분석 def analyze_image_url(url: str, question: str) -> str: response = client.chat.completions.create( model="gpt-4o", messages=[{ "role": "user", "content": [ {"type": "image_url", "image_url": {"url": url}}, {"type": "text", "text": question} ] }] ) return response.choices[0].message.content # 사용 예시 result = analyze_image("chart.png", "이 차트에서 가장 높은 값은 무엇이고 어떤 트렌드를 보이나요?") print(result) ``` --- ## Claude Vision API Claude는 특히 긴 문서와 복잡한 레이아웃 분석에 강합니다. ```python import anthropic import base64 import httpx client = anthropic.Anthropic() def analyze_with_claude(image_path: str, question: str) -> str: with open(image_path, "rb") as f: image_data = base64.standard_b64encode(f.read()).decode("utf-8") ext = image_path.rsplit(".", 1)[-1].lower() media_type_map = {"jpg": "image/jpeg", "jpeg": "image/jpeg", "png": "image/png", "gif": "image/gif", "webp": "image/webp"} media_type = media_type_map.get(ext, "image/jpeg") response = client.messages.create( model="claude-sonnet-4-5-20251001", max_tokens=2048, messages=[{ "role": "user", "content": [ { "type": "image", "source": { "type": "base64", "media_type": media_type, "data": image_data, }, }, {"type": "text", "text": question} ] }] ) return response.content[0].text # URL에서 다운로드 후 분석 def analyze_url_with_claude(url: str, question: str) -> str: image_data = base64.standard_b64encode(httpx.get(url).content).decode("utf-8") response = client.messages.create( model="claude-sonnet-4-5-20251001", max_tokens=2048, messages=[{ "role": "user", "content": [ { "type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": image_data}, }, {"type": "text", "text": question} ] }] ) return response.content[0].text ``` --- ## PDF 문서 분석 (멀티페이지) ```python import fitz # PyMuPDF import base64 from anthropic import Anthropic client = Anthropic() def pdf_to_images(pdf_path: str, dpi: int = 150) -> list[str]: doc = fitz.open(pdf_path) images = [] for page in doc: mat = fitz.Matrix(dpi/72, dpi/72) pix = page.get_pixmap(matrix=mat) img_bytes = pix.tobytes("png") images.append(base64.standard_b64encode(img_bytes).decode("utf-8")) doc.close() return images def analyze_pdf(pdf_path: str, question: str, max_pages: int = 10) -> str: images = pdf_to_images(pdf_path)[:max_pages] content = [] for i, img_data in enumerate(images): content.append({ "type": "image", "source": {"type": "base64", "media_type": "image/png", "data": img_data} }) content.append({"type": "text", "text": question}) response = client.messages.create( model="claude-sonnet-4-5-20251001", max_tokens=4096, messages=[{"role": "user", "content": content}] ) return response.content[0].text # 계약서 핵심 조항 추출 result = analyze_pdf( "contract.pdf", "이 계약서에서 위약금, 계약 기간, 비밀유지 조항을 표 형태로 정리해줘" ) ``` --- ## 실전 활용 사례 ### 영수증/청구서 OCR ```python def extract_invoice_data(image_path: str) -> dict: prompt = '''이 영수증/청구서에서 다음 정보를 JSON으로 추출하세요: { "vendor": "가게/회사명", "date": "날짜 (YYYY-MM-DD)", "total": 총금액(숫자), "tax": 세금(숫자), "items": [{"name": "상품명", "qty": 수량, "price": 가격}] }''' result = analyze_with_claude(image_path, prompt) import json return json.loads(result.replace("```json", "").replace("```", "").strip()) # 상품 사진 → 상세 설명 생성 def generate_product_description(image_url: str) -> str: return analyze_image_url( image_url, "이 상품의 상세 설명을 한국어로 작성해줘. 특징, 소재, 활용 방법을 포함해서 200자 이내로." ) ``` ### 차트 데이터 추출 ```python def extract_chart_data(chart_path: str) -> dict: prompt = '''이 차트에서 다음을 JSON으로 반환하세요: { "chart_type": "차트 유형", "title": "제목", "x_axis": "X축 레이블", "y_axis": "Y축 레이블", "data_points": [{"label": "라벨", "value": 값}], "trend": "전반적인 트렌드 설명" }''' import json result = analyze_with_claude(chart_path, prompt) return json.loads(result.replace("```json", "").replace("```", "").strip()) ``` --- ## 모델 멀티모달 기능 비교 | 모델 | 이미지/요청 | 해상도 | PDF | 동영상 | 강점 | |------|-----------|------|-----|------|------| | GPT-4o | 최대 20개 | 최대 2048px | ❌ (이미지 변환 필요) | ❌ | 다이어그램, UI 분석 | | Claude 3.5 Sonnet | 최대 20개 | 최대 8000px | ✅ 네이티브 | ❌ | 긴 문서, 복잡한 레이아웃 | | Gemini 2.5 Pro | 최대 16개 | 제한 없음 | ✅ | ✅ | 동영상, 초대형 문서 | | Gemini 2.5 Flash | 최대 16개 | 제한 없음 | ✅ | ✅ | 가성비 최고 | --- ## 기술 심층 분석 ### Vision 모델 내부 동작 GPT-4V/Claude Vision은 CLIP 같은 인코더로 이미지를 패치(보통 224×224px 타일) 단위로 분할하고, 각 패치를 임베딩 벡터로 변환해 텍스트 토큰과 함께 트랜스포머에 입력합니다. `detail: "high"` 모드(OpenAI)는 이미지를 더 많은 타일로 분할해 세밀한 분석이 가능하지만 토큰 비용도 증가합니다. ### 이미지 토큰 비용 - OpenAI GPT-4o `detail: "low"`: 이미지당 고정 85토큰 - OpenAI GPT-4o `detail: "high"`: 이미지 크기에 따라 170~1360토큰 - Claude: 이미지 크기/해상도에 비례 (1024×1024 ≈ 1600 tokens) ### 멀티모달 프롬프트 팁 1. **구체적 질문**: "이미지 설명해줘" 보다 "왼쪽 그래프의 2023년 매출액은?" 2. **구조화 요청**: JSON, 표, 목록 형식으로 출력 요청 3. **좌표 참조**: "왼쪽 상단의..." 처럼 공간적 참조 활용 4. **체인 분석**: 여러 이미지를 순서대로 제공해 비교 분석 --- ## Footnotes [^1]: GPT-4V 논문: [GPT-4 Technical Report](https://arxiv.org/abs/2303.08774) — OpenAI의 멀티모달 LLM 기술 개요 [^2]: Claude Vision 이미지 크기 권장사항: 최적 품질/비용을 위해 1568px 이하 권장 (Anthropic 공식 문서) [^3]: Gemini 1.5 Pro는 최대 3600개 이미지를 하나의 요청에 포함 가능 (1M 컨텍스트 활용) — 동영상 프레임 분석에 활용

Claude API 완전 가이드: Messages, Vision, Tool Use, Streaming

## Claude API 시작하기 ```bash pip install anthropic ``` ```python import anthropic client = anthropic.Anthropic(api_key="sk-ant-...") # 또는 ANTHROPIC_API_KEY 환경변수 message = client.messages.create( model="claude-sonnet-4-6-20251001", max_tokens=1024, messages=[{"role": "user", "content": "안녕하세요!"}] ) print(message.content[0].text) ``` --- ## 모델 선택 가이드 | 모델 | 특징 | 입력 비용 | 출력 비용 | |------|------|-----------|-----------| | claude-opus-4-6 | 최고 성능, 복잡한 추론 | $15/M | $75/M | | claude-sonnet-4-6 | 균형 잡힌 성능 (권장) | $3/M | $15/M | | claude-haiku-4-5 | 빠르고 저렴 | $0.80/M | $4/M | --- ## System Prompt 활용 ```python response = client.messages.create( model="claude-sonnet-4-6-20251001", max_tokens=2048, system='''당신은 시니어 백엔드 엔지니어입니다. - 코드는 항상 타입 힌트 포함 - 예외 처리 필수 - 한국어로 설명 - 보안 취약점이 있으면 반드시 지적''', messages=[{ "role": "user", "content": "사용자 로그인 API를 FastAPI로 구현해줘" }] ) ``` --- ## 멀티턴 대화 ```python conversation_history = [] def chat(user_message: str) -> str: conversation_history.append({ "role": "user", "content": user_message }) response = client.messages.create( model="claude-sonnet-4-6-20251001", max_tokens=1024, system="당신은 친절한 Python 튜터입니다.", messages=conversation_history ) assistant_message = response.content[0].text conversation_history.append({ "role": "assistant", "content": assistant_message }) return assistant_message # 사용 print(chat("파이썬의 리스트 컴프리헨션을 설명해줘")) print(chat("그럼 딕셔너리 컴프리헨션은?")) print(chat("둘의 차이점을 예제로 보여줘")) ``` --- ## Vision: 이미지 분석 ```python import base64 from pathlib import Path # 파일에서 이미지 읽기 def encode_image(image_path: str) -> str: with open(image_path, "rb") as f: return base64.standard_b64encode(f.read()).decode("utf-8") # 이미지 + 텍스트 질문 response = client.messages.create( model="claude-sonnet-4-6-20251001", max_tokens=1024, messages=[{ "role": "user", "content": [ { "type": "image", "source": { "type": "base64", "media_type": "image/jpeg", "data": encode_image("screenshot.jpg"), }, }, { "type": "text", "text": "이 스크린샷에서 UI 개선점을 3가지 알려줘" } ], }] ) print(response.content[0].text) # URL로 이미지 전달 response = client.messages.create( model="claude-sonnet-4-6-20251001", max_tokens=512, messages=[{ "role": "user", "content": [ {"type": "image", "source": {"type": "url", "url": "https://example.com/chart.png"}}, {"type": "text", "text": "이 차트를 분석해줘"} ] }] ) ``` --- ## Tool Use (함수 호출) ```python import json tools = [ { "name": "get_weather", "description": "특정 도시의 현재 날씨를 가져옵니다", "input_schema": { "type": "object", "properties": { "city": {"type": "string", "description": "도시 이름 (예: Seoul, Tokyo)"}, "unit": {"type": "string", "enum": ["celsius", "fahrenheit"], "default": "celsius"} }, "required": ["city"] } }, { "name": "search_products", "description": "상품 데이터베이스에서 검색합니다", "input_schema": { "type": "object", "properties": { "query": {"type": "string"}, "max_price": {"type": "number"} }, "required": ["query"] } } ] def process_tool_call(tool_name: str, tool_input: dict) -> str: if tool_name == "get_weather": # 실제 날씨 API 호출 return json.dumps({"city": tool_input["city"], "temp": 22, "condition": "맑음"}) elif tool_name == "search_products": return json.dumps([{"name": "노트북", "price": 1200000}]) # 에이전트 루프 def run_agent(user_message: str): messages = [{"role": "user", "content": user_message}] while True: response = client.messages.create( model="claude-sonnet-4-6-20251001", max_tokens=1024, tools=tools, messages=messages ) if response.stop_reason == "end_turn": return response.content[0].text # 도구 호출 처리 tool_use_block = next(b for b in response.content if b.type == "tool_use") tool_result = process_tool_call(tool_use_block.name, tool_use_block.input) # 대화에 추가 messages.append({"role": "assistant", "content": response.content}) messages.append({ "role": "user", "content": [{ "type": "tool_result", "tool_use_id": tool_use_block.id, "content": tool_result }] }) result = run_agent("서울 날씨 알려줘") print(result) ``` --- ## 스트리밍 응답 ```python # 동기 스트리밍 with client.messages.stream( model="claude-sonnet-4-6-20251001", max_tokens=1024, messages=[{"role": "user", "content": "긴 소설의 첫 장을 써줘"}] ) as stream: for text in stream.text_stream: print(text, end="", flush=True) # 비동기 스트리밍 (FastAPI/웹 서버) import asyncio import anthropic async def stream_response(prompt: str): async_client = anthropic.AsyncAnthropic() async with async_client.messages.stream( model="claude-sonnet-4-6-20251001", max_tokens=1024, messages=[{"role": "user", "content": prompt}] ) as stream: async for text in stream.text_stream: yield text ``` --- ## 구조화된 출력 (JSON 강제) ```python import json response = client.messages.create( model="claude-sonnet-4-6-20251001", max_tokens=1024, tools=[{ "name": "extract_info", "description": "텍스트에서 정보를 추출합니다", "input_schema": { "type": "object", "properties": { "name": {"type": "string"}, "company": {"type": "string"}, "email": {"type": "string"}, "phone": {"type": "string"} }, "required": ["name", "company"] } }], tool_choice={"type": "tool", "name": "extract_info"}, messages=[{ "role": "user", "content": "김철수 씨는 네이버 개발팀에 근무하며 이메일은 [email protected]입니다." }] ) info = response.content[0].input print(json.dumps(info, ensure_ascii=False, indent=2)) ``` --- ## 비용 최적화 팁 1. **캐싱 활용**: 동일 시스템 프롬프트는 Prompt Caching으로 90% 비용 절감 2. **Haiku 우선**: 단순 분류/추출 작업은 claude-haiku-4-5로 충분 3. **max_tokens 조절**: 불필요하게 높게 설정하지 않기 4. **배치 처리**: Batch API로 비동기 처리 시 50% 할인 ```python # Prompt Caching (반복 사용되는 긴 시스템 프롬프트) response = client.messages.create( model="claude-sonnet-4-6-20251001", max_tokens=1024, system=[{ "type": "text", "text": "매우 긴 시스템 프롬프트...", "cache_control": {"type": "ephemeral"} # 캐시 적용 }], messages=[{"role": "user", "content": user_input}] ) ``` Anthropic 콘솔(console.anthropic.com)에서 API 키 발급 후 바로 시작할 수 있습니다.