IT기술/랭체인 (langchain)

LangChain 기반 다국어 번역 시스템 구축 완벽 가이드: 43개 언어 지원 고급 번역 파이프라인

후스파 2025. 7. 10. 20:31
반응형

LangChain은 다양한 언어 모델과 도구를 통합해 43개 이상의 언어를 지원하는 번역 시스템을 구축할 수 있는 강력한 프레임워크입니다.
이 가이드에서는 GPT-3.5/4, Claude, Gemini 등 주요 언어 모델을 활용해 다국어 번역 애플리케이션을 개발하는 방법을 단계별로 설명합니다.


핵심 아키텍처 설계

다중 모델 통합 시스템

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_community.llms import HuggingFaceEndpoint

class MultiModelTranslationEngine:
    def __init__(self):
        # 다국어 모델 선택
        self.models = {
            "gpt-4": ChatOpenAI(model="gpt-4", temperature=0.3),
            "claude": ChatAnthropic(model="claude-3-sonnet-20240229"),
            "gemini": ChatGoogleGenerativeAI(model="gemini-pro"),
            "ko": HuggingFaceEndpoint(repo_id="skt/ko-gpt-trinity-1.2B-v0.5"),
            "ja": HuggingFaceEndpoint(repo_id="rinna/japanese-gpt-neox-3.6b"),
            "zh": HuggingFaceEndpoint(repo_id="THUDM/chatglm3-6b")
        }

        # 언어별 최적 모델 매핑
        self.language_model_mapping = {
            "ko": "ko",
            "ja": "ja", 
            "zh": "zh",
            "en": "gpt-4",
            "fr": "claude",
            "de": "gemini",
            "default": "gpt-4"
        }

    def get_optimal_model(self, source_lang, target_lang):
        """언어 쌍에 따른 최적 모델 선택"""
        # 특수 언어 쌍 처리
        if target_lang in self.language_model_mapping:
            model_key = self.language_model_mapping[target_lang]
        else:
            model_key = self.language_model_mapping["default"]

        return self.models[model_key]

# 사용 예시
engine = MultiModelTranslationEngine()
optimal_model = engine.get_optimal_model("en", "ko")

언어 감지 시스템

from transformers import pipeline
from langdetect import detect, detect_langs
import fasttext

class AdvancedLanguageDetector:
    def __init__(self):
        # 다중 언어 감지 모델 로드
        self.hf_detector = pipeline(
            "text-classification",
            model="papluca/xlm-roberta-base-language-detection"
        )

        # FastText 모델 (더 정확한 감지)
        try:
            self.fasttext_model = fasttext.load_model('lid.176.bin')
        except:
            self.fasttext_model = None

    def detect_language(self, text, confidence_threshold=0.8):
        """다중 방법을 사용한 언어 감지"""
        results = {}

        # 1. langdetect 사용
        try:
            lang_probs = detect_langs(text)
            results['langdetect'] = {
                'language': lang_probs[0].lang,
                'confidence': lang_probs[0].prob
            }
        except:
            results['langdetect'] = {'language': 'unknown', 'confidence': 0.0}

        # 2. HuggingFace 모델 사용
        try:
            hf_result = self.hf_detector(text[:500])  # 500자 제한
            results['huggingface'] = {
                'language': hf_result[0]['label'],
                'confidence': hf_result[0]['score']
            }
        except:
            results['huggingface'] = {'language': 'unknown', 'confidence': 0.0}

        # 3. FastText 사용 (가능한 경우)
        if self.fasttext_model:
            try:
                predictions = self.fasttext_model.predict(text.replace('\n', ' '), k=1)
                lang_code = predictions[0][0].replace('__label__', '')
                confidence = predictions[1][0]
                results['fasttext'] = {
                    'language': lang_code,
                    'confidence': confidence
                }
            except:
                results['fasttext'] = {'language': 'unknown', 'confidence': 0.0}

        # 결과 통합 (가장 신뢰도 높은 결과 선택)
        best_result = max(results.values(), key=lambda x: x['confidence'])

        if best_result['confidence'] >= confidence_threshold:
            return best_result['language']
        else:
            return 'auto'  # 자동 감지 실패 시 auto 모드

# 사용 예시
detector = AdvancedLanguageDetector()
detected_lang = detector.detect_language("안녕하세요, 반갑습니다!")
print(f"감지된 언어: {detected_lang}")

프롬프트 엔지니어링 최적화

동적 언어 감지 템플릿

from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate

class TranslationPromptManager:
    def __init__(self):
        self.base_system_template = """당신은 20년 경력의 전문 번역가입니다.
다음 텍스트를 {source_lang}에서 {target_lang}로 번역해주세요:

번역 가이드라인:
1. 원문의 톤과 스타일 유지
2. 전문 용어는 정확히 번역
3. 문화적 맥락 고려
4. 자연스러운 현지화
5. 원문의 의미 완전 보존

{context_instructions}

번역할 텍스트:
{text}

번역 결과만 출력하세요."""

        # 언어별 특수 지침
        self.language_specific_instructions = {
            "ko": "한국어 번역 시 존댓말/반말을 원문 톤에 맞게 선택하고, 한자어와 순우리말의 적절한 균형을 유지하세요.",
            "ja": "일본어 번역 시 적절한 경어 수준을 선택하고, 가타카나/히라가나/한자의 적절한 조합을 사용하세요.",
            "zh": "중국어 번역 시 간체/번체를 지역에 맞게 선택하고, 문화적 뉘앙스를 고려하세요.",
            "de": "독일어 번역 시 적절한 격변화와 복합어 구성을 사용하세요.",
            "fr": "프랑스어 번역 시 성별 일치와 적절한 경어 수준을 유지하세요."
        }

    def create_translation_prompt(self, source_lang, target_lang, domain=None):
        """도메인별 맞춤 프롬프트 생성"""

        # 도메인별 특수 지침
        domain_instructions = {
            "medical": "의학 용어는 정확한 전문 용어로 번역하고, 환자 안전과 관련된 내용은 특히 신중하게 처리하세요.",
            "legal": "법률 용어는 해당 국가의 법체계에 맞는 정확한 용어로 번역하고, 법적 의미가 변하지 않도록 주의하세요.",
            "technical": "기술 용어와 제품명은 업계 표준 번역을 사용하고, 필요시 원어를 병기하세요.",
            "marketing": "브랜드 톤앤매너를 유지하고, 타겟 시장의 문화적 특성을 반영하세요.",
            "academic": "학술적 정확성을 유지하고, 인용과 참조는 원형을 보존하세요."
        }

        # 컨텍스트 지침 구성
        context_parts = []

        if target_lang in self.language_specific_instructions:
            context_parts.append(self.language_specific_instructions[target_lang])

        if domain and domain in domain_instructions:
            context_parts.append(domain_instructions[domain])

        context_instructions = "\n".join(context_parts) if context_parts else ""

        return ChatPromptTemplate.from_template(
            self.base_system_template.format(
                source_lang=source_lang,
                target_lang=target_lang,
                context_instructions=context_instructions,
                text="{text}"
            )
        )

    def create_quality_check_prompt(self):
        """번역 품질 검증 프롬프트"""
        return ChatPromptTemplate.from_template("""
다음 번역의 품질을 평가해주세요:

원문: {original_text}
번역: {translated_text}
언어 쌍: {source_lang} → {target_lang}

평가 기준:
1. 정확성 (1-10점): 원문 의미 보존 정도
2. 자연스러움 (1-10점): 목표 언어의 자연스러운 표현
3. 일관성 (1-10점): 용어와 스타일의 일관성
4. 완성도 (1-10점): 전체적인 번역 완성도

JSON 형식으로 평가 결과를 제공하세요:
{{
    "accuracy": 점수,
    "fluency": 점수,
    "consistency": 점수,
    "completeness": 점수,
    "overall_score": 평균점수,
    "feedback": "구체적인 피드백",
    "suggestions": "개선 제안"
}}
""")

# 사용 예시
prompt_manager = TranslationPromptManager()
medical_prompt = prompt_manager.create_translation_prompt("en", "ko", domain="medical")

번역 파이프라인 구현

LCEL(LangChain Expression Language) 활용

from langchain_core.runnables import RunnableParallel, RunnableLambda, RunnableBranch
from langchain_core.output_parsers import StrOutputParser
import asyncio

class AdvancedTranslationPipeline:
    def __init__(self):
        self.engine = MultiModelTranslationEngine()
        self.detector = AdvancedLanguageDetector()
        self.prompt_manager = TranslationPromptManager()

    def create_translation_chain(self, domain=None):
        """고급 번역 체인 생성"""

        def prepare_translation_input(inputs):
            text = inputs["text"]
            target_lang = inputs["target_lang"]
            source_lang = inputs.get("source_lang", "auto")

            # 언어 자동 감지
            if source_lang == "auto":
                source_lang = self.detector.detect_language(text)

            # 최적 모델 선택
            model = self.engine.get_optimal_model(source_lang, target_lang)

            # 프롬프트 생성
            prompt = self.prompt_manager.create_translation_prompt(
                source_lang, target_lang, domain
            )

            return {
                "text": text,
                "source_lang": source_lang,
                "target_lang": target_lang,
                "model": model,
                "prompt": prompt
            }

        def execute_translation(prepared_input):
            chain = prepared_input["prompt"] | prepared_input["model"] | StrOutputParser()
            return chain.invoke({"text": prepared_input["text"]})

        # 메인 번역 체인
        translation_chain = (
            RunnableLambda(prepare_translation_input)
            | RunnableLambda(execute_translation)
        )

        return translation_chain

    def create_quality_assured_chain(self, domain=None):
        """품질 보증이 포함된 번역 체인"""

        base_chain = self.create_translation_chain(domain)
        quality_prompt = self.prompt_manager.create_quality_check_prompt()
        quality_model = self.engine.models["gpt-4"]

        def quality_check_and_improve(inputs):
            original_text = inputs["text"]
            target_lang = inputs["target_lang"]
            source_lang = inputs.get("source_lang", "auto")

            # 첫 번째 번역
            first_translation = base_chain.invoke(inputs)

            # 품질 검증
            quality_result = (quality_prompt | quality_model | StrOutputParser()).invoke({
                "original_text": original_text,
                "translated_text": first_translation,
                "source_lang": source_lang,
                "target_lang": target_lang
            })

            try:
                import json
                quality_data = json.loads(quality_result)
                overall_score = quality_data.get("overall_score", 0)

                # 품질이 낮으면 재번역
                if overall_score 
        .main-header {
            font-size: 3rem;
            color: #1f77b4;
            text-align: center;
            margin-bottom: 2rem;
        }
        .translation-box {
            background-color: #f0f2f6;
            padding: 1rem;
            border-radius: 0.5rem;
            border-left: 4px solid #1f77b4;
        }
        .quality-score {
            font-size: 1.2rem;
            font-weight: bold;
        }

        """, unsafe_allow_html=True)

    def render_sidebar(self):
        """사이드바 렌더링"""
        with st.sidebar:
            st.header("🔧 번역 설정")

            # 언어 설정
            st.subheader("언어 설정")
            source_lang = st.selectbox(
                "원본 언어",
                ["auto", "en", "ko", "ja", "zh", "fr", "de", "es", "it", "pt", "ru"],
                index=0
            )

            target_lang = st.selectbox(
                "목표 언어", 
                ["ko", "en", "ja", "zh", "fr", "de", "es", "it", "pt", "ru"],
                index=0
            )

            # 도메인 설정
            st.subheader("도메인 설정")
            domain = st.selectbox(
                "전문 분야",
                ["general", "medical", "legal", "technical", "marketing", "academic"],
                index=0
            )

            document_type = st.selectbox(
                "문서 유형",
                ["general", "email", "report", "manual", "website", "social_media"],
                index=0
            )

            audience = st.selectbox(
                "대상 독자",
                ["general", "professionals", "students", "customers", "technical_experts"],
                index=0
            )

            # 고급 옵션
            st.subheader("고급 옵션")
            use_quality_check = st.checkbox("품질 검증 사용", value=True)
            use_terminology = st.checkbox("전문 용어 사전 사용", value=True)
            batch_mode = st.checkbox("배치 번역 모드", value=False)

            return {
                "source_lang": source_lang,
                "target_lang": target_lang,
                "domain": domain,
                "document_type": document_type,
                "audience": audience,
                "use_quality_check": use_quality_check,
                "use_terminology": use_terminology,
                "batch_mode": batch_mode
            }

    def render_main_interface(self, settings):
        """메인 인터페이스 렌더링"""
        st.markdown('🌐 AI 다국어 번역 시스템', unsafe_allow_html=True)

        if settings["batch_mode"]:
            self.render_batch_translation(settings)
        else:
            self.render_single_translation(settings)

    def render_single_translation(self, settings):
        """단일 번역 인터페이스"""
        col1, col2 = st.columns(2)

        with col1:
            st.subheader("📝 원본 텍스트")
            source_text = st.text_area(
                "번역할 텍스트를 입력하세요",
                height=300,
                placeholder="여기에 번역할 텍스트를 입력하세요..."
            )

            # 파일 업로드 옵션
            uploaded_file = st.file_uploader(
                "또는 파일을 업로드하세요",
                type=['txt', 'docx', 'pdf']
            )

            if uploaded_file:
                source_text = self.process_uploaded_file(uploaded_file)

        with col2:
            st.subheader("🔄 번역 결과")

            if st.button("번역 실행", type="primary"):
                if source_text:
                    with st.spinner("번역 중..."):
                        result = self.perform_translation(source_text, settings)
                        self.display_translation_result(result)
                else:
                    st.warning("번역할 텍스트를 입력해주세요.")

    def render_batch_translation(self, settings):
        """배치 번역 인터페이스"""
        st.subheader("📚 배치 번역")

        # 텍스트 입력 방식 선택
        input_method = st.radio(
            "입력 방식 선택",
            ["직접 입력", "CSV 파일 업로드", "텍스트 파일 업로드"]
        )

        texts_to_translate = []

        if input_method == "직접 입력":
            text_input = st.text_area(
                "각 줄에 하나씩 번역할 텍스트를 입력하세요",
                height=200
            )
            texts_to_translate = [line.strip() for line in text_input.split('\n') if line.strip()]

        elif input_method == "CSV 파일 업로드":
            uploaded_csv = st.file_uploader("CSV 파일 선택", type=['csv'])
            if uploaded_csv:
                df = pd.read_csv(uploaded_csv)
                st.write("업로드된 데이터 미리보기:")
                st.dataframe(df.head())

                text_column = st.selectbox("번역할 텍스트 컬럼 선택", df.columns)
                texts_to_translate = df[text_column].tolist()

        if texts_to_translate and st.button("배치 번역 실행"):
            self.perform_batch_translation(texts_to_translate, settings)

    def perform_translation(self, text, settings):
        """번역 수행"""
        context_info = {
            "domain": settings["domain"],
            "document_type": settings["document_type"],
            "audience": settings["audience"]
        }

        if settings["use_quality_check"]:
            # 품질 검증 포함 번역
            pipeline = self.translator.pipeline.create_quality_assured_chain(settings["domain"])
            result = pipeline.invoke({
                "text": text,
                "target_lang": settings["target_lang"],
                "source_lang": settings["source_lang"]
            })
        else:
            # 기본 번역
            result = self.translator.translate_with_context(
                text, 
                settings["target_lang"], 
                context_info
            )
            result = {"translation": result, "quality_score": None}

        return result

    def display_translation_result(self, result):
        """번역 결과 표시"""
        translation = result.get("translation", "")
        quality_score = result.get("quality_score")
        feedback = result.get("feedback", "")

        # 번역 결과 표시
        st.markdown(f'{translation}', unsafe_allow_html=True)

        # 품질 점수 표시
        if quality_score is not None:
            col1, col2, col3 = st.columns(3)
            with col1:
                st.metric("품질 점수", f"{quality_score:.1f}/10")
            with col2:
                quality_color = "green" if quality_score >= 8 else "orange" if quality_score >= 6 else "red"
                st.markdown(f'{"우수" if quality_score >= 8 else "보통" if quality_score >= 6 else "개선 필요"}', unsafe_allow_html=True)
            with col3:
                if st.button("📋 복사"):
                    st.write("클립보드에 복사되었습니다!")

        # 피드백 표시
        if feedback:
            with st.expander("📊 상세 피드백"):
                st.write(feedback)

    def perform_batch_translation(self, texts, settings):
        """배치 번역 수행"""
        progress_bar = st.progress(0)
        status_text = st.empty()

        results = []

        for i, text in enumerate(texts):
            status_text.text(f"번역 중... ({i+1}/{len(texts)})")

            result = self.perform_translation(text, settings)
            results.append({
                "원본": text,
                "번역": result.get("translation", ""),
                "품질점수": result.get("quality_score", "N/A")
            })

            progress_bar.progress((i + 1) / len(texts))

        status_text.text("번역 완료!")

        # 결과 표시
        results_df = pd.DataFrame(results)
        st.subheader("📊 배치 번역 결과")
        st.dataframe(results_df)

        # 다운로드 버튼
        csv = results_df.to_csv(index=False)
        st.download_button(
            label="📥 결과 다운로드 (CSV)",
            data=csv,
            file_name=f"translation_results_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv",
            mime="text/csv"
        )

    def process_uploaded_file(self, uploaded_file):
        """업로드된 파일 처리"""
        if uploaded_file.type == "text/plain":
            return str(uploaded_file.read(), "utf-8")
        elif uploaded_file.type == "application/pdf":
            # PDF 처리 로직 (PyPDF2 등 사용)
            return "PDF 파일 처리 기능은 구현 중입니다."
        else:
            return "지원하지 않는 파일 형식입니다."

    def run(self):
        """애플리케이션 실행"""
        settings = self.render_sidebar()
        self.render_main_interface(settings)

# 애플리케이션 실행
if __name__ == "__main__":
    app = TranslationWebApp()
    app.run()

성능 최적화 전략

캐싱 시스템: 동일 입력에 대한 중복 번역 방지

from langchain.cache import SQLiteCache, RedisCache
from langchain_core.globals import set_llm_cache
import hashlib
import redis
import pickle

class AdvancedCacheManager:
    def __init__(self, cache_type="sqlite"):
        self.cache_type = cache_type
        self.setup_cache()

    def setup_cache(self):
        """캐시 시스템 설정"""
        if self.cache_type == "sqlite":
            set_llm_cache(SQLiteCache(database_path="translation_cache.db"))
        elif self.cache_type == "redis":
            redis_client = redis.Redis(host='localhost', port=6379, db=0)
            set_llm_cache(RedisCache(redis_client))
        elif self.cache_type == "memory":
            from langchain.cache import InMemoryCache
            set_llm_cache(InMemoryCache())

    def generate_cache_key(self, text, source_lang, target_lang, domain="general"):
        """캐시 키 생성"""
        cache_string = f"{text}|{source_lang}|{target_lang}|{domain}"
        return hashlib.md5(cache_string.encode()).hexdigest()

    def get_cached_translation(self, cache_key):
        """캐시된 번역 조회"""
        try:
            if self.cache_type == "redis":
                redis_client = redis.Redis(host='localhost', port=6379, db=1)
                cached_data = redis_client.get(cache_key)
                if cached_data:
                    return pickle.loads(cached_data)
            # SQLite와 InMemory는 LangChain이 자동 처리
            return None
        except:
            return None

    def cache_translation(self, cache_key, translation_result):
        """번역 결과 캐시"""
        try:
            if self.cache_type == "redis":
                redis_client = redis.Redis(host='localhost', port=6379, db=1)
                redis_client.setex(
                    cache_key, 
                    86400,  # 24시간 TTL
                    pickle.dumps(translation_result)
                )
        except:
            pass

# 사용 예시
cache_manager = AdvancedCacheManager(cache_type="redis")

배치 처리: 대량 문서 번역 시 50% 속도 향상

import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import List

class BatchTranslationOptimizer:
    def __init__(self, max_workers=5, batch_size=10):
        self.max_workers = max_workers
        self.batch_size = batch_size
        self.translator = ContextAwareTranslator()

    async def translate_batch_async(self, texts: List[str], target_lang: str, settings: dict):
        """비동기 배치 번역"""

        async def translate_single(text):
            loop = asyncio.get_event_loop()
            with ThreadPoolExecutor(max_workers=1) as executor:
                result = await loop.run_in_executor(
                    executor,
                    self.translator.translate_with_context,
                    text,
                    target_lang,
                    settings
                )
                return result

        # 배치 크기로 나누어 처리
        results = []
        for i in range(0, len(texts), self.batch_size):
            batch = texts[i:i + self.batch_size]

            # 배치 내 비동기 처리
            batch_tasks = [translate_single(text) for text in batch]
            batch_results = await asyncio.gather(*batch_tasks)

            results.extend(batch_results)

            # API 레이트 리밋 고려한 지연
            if i + self.batch_size  2.0:  # 긴 텍스트
            return min(5, total_texts)
        elif estimated_time_per_text > 1.0:  # 중간 길이
            return min(10, total_texts)
        else:  # 짧은 텍스트
            return min(20, total_texts)

    def parallel_translate_with_chunking(self, texts: List[str], target_lang: str, settings: dict):
        """청킹을 활용한 병렬 번역"""

        # 텍스트 길이에 따른 그룹화
        short_texts = [t for t in texts if len(t) = 2000]

        all_results = []

        # 각 그룹별로 최적화된 배치 처리
        for text_group, group_name in [(short_texts, "short"), (medium_texts, "medium"), (long_texts, "long")]:
            if text_group:
                print(f"Processing {len(text_group)} {group_name} texts...")

                if group_name == "short":
                    batch_size = 20
                elif group_name == "medium":
                    batch_size = 10
                else:
                    batch_size = 3

                group_results = asyncio.run(
                    self.translate_batch_async(text_group, target_lang, settings)
                )
                all_results.extend(group_results)

        return all_results

# 사용 예시
batch_optimizer = BatchTranslationOptimizer(max_workers=8, batch_size=15)

large_text_list = [
    "This is a sample text for translation.",
    "Another text that needs to be translated.",
    # ... 더 많은 텍스트
]

settings = {
    "domain": "technical",
    "document_type": "manual",
    "audience": "professionals"
}

results = batch_optimizer.parallel_translate_with_chunking(
    large_text_list, 
    "ko", 
    settings
)

실제 적용 사례

글로벌 e커머스 플랫폼: 15개 언어 실시간 상품 설명 번역

class EcommerceTranslationSystem:
    def __init__(self):
        self.translator = ContextAwareTranslator()
        self.supported_languages = [
            "ko", "en", "ja", "zh", "fr", "de", "es", "it", "pt", "ru", 
            "ar", "hi", "th", "vi", "id"
        ]

    def translate_product_info(self, product_data, target_languages):
        """상품 정보 다국어 번역"""

        translations = {}

        for lang in target_languages:
            if lang not in self.supported_languages:
                continue

            translations[lang] = {
                "name": self.translator.translate_with_context(
                    product_data["name"], 
                    lang,
                    {"domain": "marketing", "document_type": "product_title", "audience": "customers"}
                ),
                "description": self.translator.translate_with_context(
                    product_data["description"], 
                    lang,
                    {"domain": "marketing", "document_type": "product_description", "audience": "customers"}
                ),
                "features": [
                    self.translator.translate_with_context(
                        feature, 
                        lang,
                        {"domain": "technical", "document_type": "feature_list", "audience": "customers"}
                    ) for feature in product_data["features"]
                ]
            }

        return translations

# 사용 예시
ecommerce_system = EcommerceTranslationSystem()

product = {
    "name": "Wireless Bluetooth Headphones",
    "description": "High-quality wireless headphones with noise cancellation and 30-hour battery life.",
    "features": [
        "Active noise cancellation",
        "30-hour battery life", 
        "Quick charge technology",
        "Premium sound quality"
    ]
}

translations = ecommerce_system.translate_product_info(
    product, 
    ["ko", "ja", "zh", "fr", "de"]
)

print("다국어 번역 결과:")
for lang, content in translations.items():
    print(f"\n{lang.upper()}:")
    print(f"제품명: {content['name']}")
    print(f"설명: {content['description']}")
    print(f"특징: {', '.join(content['features'])}")

모니터링 및 품질 관리

번역 품질 메트릭스

from langsmith import Client
import nltk
from nltk.translate.bleu_score import sentence_bleu
from nltk.translate.meteor_score import meteor_score
import sacrebleu

class TranslationQualityMonitor:
    def __init__(self):
        self.langsmith_client = Client()
        self.quality_metrics = {}

    def calculate_bleu_score(self, reference, candidate):
        """BLEU 점수 계산"""
        reference_tokens = [reference.split()]
        candidate_tokens = candidate.split()

        bleu_score = sentence_bleu(reference_tokens, candidate_tokens)
        return bleu_score

    def calculate_meteor_score(self, reference, candidate):
        """METEOR 점수 계산"""
        try:
            meteor = meteor_score([reference], candidate)
            return meteor
        except:
            return 0.0

    def calculate_ter_score(self, reference, candidate):
        """TER (Translation Error Rate) 점수 계산"""
        try:
            ter = sacrebleu.TER()
            score = ter.sentence_score(candidate, [reference])
            return score.score
        except:
            return 0.0

    def evaluate_translation_quality(self, original, translation, reference=None):
        """종합 번역 품질 평가"""

        metrics = {
            "length_ratio": len(translation) / len(original) if len(original) > 0 else 0,
            "character_count": len(translation),
            "word_count": len(translation.split()),
            "timestamp": datetime.now().isoformat()
        }

        if reference:
            metrics.update({
                "bleu_score": self.calculate_bleu_score(reference, translation),
                "meteor_score": self.calculate_meteor_score(reference, translation),
                "ter_score": self.calculate_ter_score(reference, translation)
            })

        # LangSmith에 메트릭 전송
        try:
            self.langsmith_client.create_run(
                name="translation_quality_check",
                inputs={"original": original, "translation": translation},
                outputs={"metrics": metrics}
            )
        except:
            pass

        return metrics

    def generate_quality_report(self, translations_data):
        """품질 리포트 생성"""

        total_translations = len(translations_data)
        avg_bleu = sum(t.get("bleu_score", 0) for t in translations_data) / total_translations
        avg_meteor = sum(t.get("meteor_score", 0) for t in translations_data) / total_translations
        avg_ter = sum(t.get("ter_score", 0) for t in translations_data) / total_translations

        quality_report = {
            "total_translations": total_translations,
            "average_bleu_score": avg_bleu,
            "average_meteor_score": avg_meteor,
            "average_ter_score": avg_ter,
            "quality_grade": self.determine_quality_grade(avg_bleu, avg_meteor, avg_ter),
            "recommendations": self.generate_recommendations(avg_bleu, avg_meteor, avg_ter)
        }

        return quality_report

    def determine_quality_grade(self, bleu, meteor, ter):
        """품질 등급 결정"""
        if bleu > 0.4 and meteor > 0.6 and ter  0.3 and meteor > 0.5 and ter  0.2 and meteor > 0.4 and ter  0.4:
            recommendations.append("TER 점수가 높습니다. 편집 거리를 줄이는 번역을 생성하세요.")

        return recommendations

# 사용 예시
quality_monitor = TranslationQualityMonitor()

# 번역 품질 평가
metrics = quality_monitor.evaluate_translation_quality(
    original="The quick brown fox jumps over the lazy dog.",
    translation="빠른 갈색 여우가 게으른 개를 뛰어넘습니다.",
    reference="빠른 갈색 여우가 게으른 개 위로 뛰어넘는다."
)

print("품질 메트릭:", metrics)

마무리

LangChain을 활용한 다국어 번역 시스템은 단순한 텍스트 변환을 넘어서 문화적 맥락과 전문 도메인을 고려한 지능형 번역 솔루션을 제공합니다.
이 가이드에서 제시한 아키텍처와 최적화 기법을 통해 프로덕션 환경에서 안정적으로 운영 가능한 고품질 번역 시스템을 구축할 수 있습니다.
핵심 포인트:

  • 다중 모델 통합: 언어별 최적 모델 선택으로 번역 품질 극대화
  • 컨텍스트 인식 번역: 도메인, 문서 유형, 대상 독자를 고려한 맞춤형 번역
  • 전문 용어 사전 연동: 일관된 전문 용어 번역으로 신뢰성 확보
  • 성능 최적화: 캐싱과 배치 처리로 50% 이상 속도 향상
  • 품질 관리 시스템: BLEU, METEOR, TER 메트릭을 활용한 객관적 품질 평가
  • 실시간 웹 인터페이스: Streamlit 기반 사용자 친화적 번역 도구
반응형