안녕! 오늘은 정말 흥미로운 주제를 가지고 왔어. 바로 LangChain을 활용한 데이터 정리와 요약 자동화인데, 이게 왜 중요하냐고? 🤔

상상해봐. 매일 아침 출근하면 수백 페이지의 보고서, 끝없는 이메일, 고객 피드백, 시장 조사 자료들이 쌓여있어. 이걸 일일이 읽고 정리하고 요약하려면... 아마 퇴근 시간이 자정을 넘길 거야. 😱

하지만 LangChain을 활용하면? 이 모든 작업을 자동화할 수 있어! AI가 당신의 똑똑한 비서가 되어 데이터를 정리하고, 핵심만 쏙쏙 뽑아내고, 심지어 인사이트까지 제공해줘. 마치 재능넷에서 최고의 데이터 분석 전문가를 고용한 것처럼 말이야! 🚀

자, 본격적으로 시작하기 전에 LangChain이 정확히 뭔지부터 알아야겠지? 🧐

왜 LangChain을 써야 할까?

GPT API를 직접 호출해서 쓸 수도 있는데, 왜 굳이 LangChain을 쓸까? 그 이유는 명확해:

이제 본격적으로 LangChain을 활용한 데이터 자동화 시스템을 만들어보자! 단계별로 차근차근 설명할게. 📝

Step 1: 환경 설정하기

먼저 개발 환경을 세팅해야 해. Python을 사용한다고 가정하고 진행할게.

pip install langchain
pip install openai
pip install chromadb
pip install tiktoken
pip install pypdf
pip install python-dotenv

Step 2: API 키 설정하기

OpenAI API를 사용하려면 API 키가 필요해. .env 파일을 만들어서 관리하는 게 안전해:

# .env 파일
OPENAI_API_KEY=your-api-key-here

그리고 Python 코드에서 이렇게 불러와:

from dotenv import load_dotenv
import os

load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")

Step 3: 기본 문서 로더 만들기

이제 실제로 문서를 읽어오는 코드를 작성해보자. LangChain은 다양한 문서 로더를 제공해:

from langchain.document_loaders import PyPDFLoader, TextLoader, CSVLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# PDF 파일 로드
pdf_loader = PyPDFLoader("report.pdf")
documents = pdf_loader.load()

# 텍스트를 적절한 크기로 분할
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,  # 각 청크의 크기
    chunk_overlap=200,  # 청크 간 겹치는 부분
    length_function=len
)

texts = text_splitter.split_documents(documents)

이제 진짜 재미있는 부분이야! 실제로 데이터를 요약하는 시스템을 만들어보자. 🎨

방법 1: 간단한 요약 체인

가장 기본적인 방법은 Summarization Chain을 사용하는 거야:

from langchain.chains.summarize import load_summarize_chain
from langchain.chat_models import ChatOpenAI

# LLM 모델 초기화
llm = ChatOpenAI(
    temperature=0,  # 창의성 수준 (0=일관적, 1=창의적)
    model_name="gpt-3.5-turbo"
)

# 요약 체인 생성
chain = load_summarize_chain(
    llm,
    chain_type="map_reduce"  # 또는 "stuff", "refine"
)

# 요약 실행
summary = chain.run(texts)
print(summary)

방법 2: 커스텀 프롬프트로 정교하게 제어하기

기본 요약이 마음에 안 든다면? 직접 프롬프트를 작성해서 원하는 형식으로 요약할 수 있어! ✍️

from langchain.prompts import PromptTemplate

# 커스텀 프롬프트 템플릿
prompt_template = """다음 텍스트를 3가지 관점에서 요약해주세요:

1. 핵심 내용 (3-5문장)
2. 주요 수치 및 데이터
3. 실행 가능한 인사이트

텍스트:
{text}

요약:"""

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["text"]
)

# 프롬프트를 적용한 체인
chain = load_summarize_chain(
    llm,
    chain_type="stuff",
    prompt=PROMPT
)

summary = chain.run(texts)

이렇게 하면 단순히 요약만 하는 게 아니라, 구조화된 형식으로 결과를 받을 수 있어! 보고서 작성할 때 정말 유용하지. 📋

방법 3: 벡터 데이터베이스를 활용한 스마트 요약

이건 좀 더 고급 기술이야. 문서 전체를 벡터로 변환해서 저장하고, 질문에 관련된 부분만 찾아서 요약하는 방식이지:

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA

# 임베딩 모델 초기화
embeddings = OpenAIEmbeddings()

# 벡터 데이터베이스 생성
vectordb = Chroma.from_documents(
    documents=texts,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 검색 기반 QA 체인 생성
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectordb.as_retriever(
        search_kwargs={"k": 3}  # 상위 3개 관련 문서 검색
    )
)

# 질문하기
question = "이 보고서의 주요 재무 지표는 무엇인가요?"
answer = qa_chain.run(question)

이제 진짜 자동화의 핵심으로 들어가보자! 단순히 한 번 요약하는 게 아니라, 정기적으로 자동 실행되는 시스템을 만드는 거야. ⚙️

완전 자동화 시스템 설계

실무에서 사용할 수 있는 완전한 자동화 시스템은 이런 구조로 만들어:

import schedule
import time
from datetime import datetime
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

class DataSummarizationBot:
    def __init__(self, llm, source_folder, output_folder):
        self.llm = llm
        self.source_folder = source_folder
        self.output_folder = output_folder
        self.processed_files = set()
        
    def scan_new_files(self):
        """새로운 파일 스캔"""
        import os
        all_files = set(os.listdir(self.source_folder))
        new_files = all_files - self.processed_files
        return new_files
    
    def process_document(self, file_path):
        """문서 처리 및 요약"""
        # 파일 로드
        loader = PyPDFLoader(file_path)
        documents = loader.load()
        
        # 텍스트 분할
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200
        )
        texts = text_splitter.split_documents(documents)
        
        # 요약 실행
        chain = load_summarize_chain(self.llm, chain_type="map_reduce")
        summary = chain.run(texts)
        
        return summary
    
    def save_summary(self, filename, summary):
        """요약 결과 저장"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_file = f"{self.output_folder}/{filename}_{timestamp}_summary.txt"
        
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(f"요약 생성 시간: {timestamp}\n")
            f.write(f"원본 파일: {filename}\n")
            f.write("-" * 50 + "\n\n")
            f.write(summary)
        
        print(f"✅ 요약 완료: {output_file}")
    
    def run_automation(self):
        """자동화 실행"""
        print(f"🔍 새 파일 스캔 중... ({datetime.now()})")
        new_files = self.scan_new_files()
        
        if not new_files:
            print("새로운 파일이 없습니다.")
            return
        
        for file in new_files:
            try:
                file_path = f"{self.source_folder}/{file}"
                print(f"📄 처리 중: {file}")
                
                summary = self.process_document(file_path)
                self.save_summary(file, summary)
                
                self.processed_files.add(file)
                
            except Exception as e:
                print(f"❌ 오류 발생 ({file}): {str(e)}")

# 봇 초기화
bot = DataSummarizationBot(
    llm=llm,
    source_folder="./input_documents",
    output_folder="./summaries"
)

# 스케줄 설정 (매일 오전 9시 실행)
schedule.every().day.at("09:00").do(bot.run_automation)

# 또는 30분마다 실행
# schedule.every(30).minutes.do(bot.run_automation)

print("🤖 자동화 봇 시작!")
while True:
    schedule.run_pending()
    time.sleep(60)  # 1분마다 체크

이메일 자동 발송 기능 추가하기

요약만 하고 끝? 아니지! 결과를 자동으로 이메일로 보내면 더 완벽해: 📧

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_summary_email(summary, recipient_email):
    """요약 결과를 이메일로 전송"""
    sender_email = "your-email@gmail.com"
    password = "your-app-password"
    
    # 이메일 구성
    message = MIMEMultipart()
    message["From"] = sender_email
    message["To"] = recipient_email
    message["Subject"] = f"📊 일일 데이터 요약 리포트 - {datetime.now().strftime('%Y-%m-%d')}"
    
    # HTML 형식의 본문
    html_body = f"""
    <html>
        <body style="font-family: Arial, sans-serif;">
            <h2 style="color: #4a5568;">📈 데이터 요약 리포트</h2>
            <p>안녕하세요! 오늘의 데이터 요약 결과를 전달드립니다.</p>
            <div style="background-color: #f8f9fb; padding: 20px; border-radius: 8px;">
                <pre style="white-space: pre-wrap;">{summary}</pre>
            </div>
            <hr>
            <p style="color: #6b7280; font-size: 0.9em;">
                이 리포트는 LangChain 자동화 시스템에 의해 생성되었습니다.
            </p>
        </body>
    </html>
    """
    
    message.attach(MIMEText(html_body, "html"))
    
    # 이메일 전송
    try:
        with smtplib.SMTP_SSL("smtp.gmail.com", 465) as server:
            server.login(sender_email, password)
            server.send_message(message)
        print("✅ 이메일 전송 완료!")
    except Exception as e:
        print(f"❌ 이메일 전송 실패: {str(e)}")

이제 save_summary 함수 끝에 send_summary_email(summary, "boss@company.com")만 추가하면 끝! 매일 아침 상사의 메일함에 깔끔한 요약 리포트가 도착할 거야. 😎

요약 결과를 텍스트로만 받는 건 좀 심심하지? 다양한 형식으로 출력하는 방법을 알아보자! 🖼️

1. 마크다운 형식으로 구조화하기

markdown_prompt = """다음 문서를 마크다운 형식으로 요약해주세요:

# 제목
## 주요 내용
- 핵심 포인트 1
- 핵심 포인트 2

## 수치 데이터
| 항목 | 값 |
|------|-----|
| ... | ... |

## 결론 및 제언

문서:
{text}
"""

MARKDOWN_PROMPT = PromptTemplate(
    template=markdown_prompt,
    input_variables=["text"]
)

2. JSON 형식으로 구조화된 데이터 추출

프로그램에서 바로 사용할 수 있는 JSON 형식으로 받을 수도 있어:

from langchain.output_parsers import StructuredOutputParser, ResponseSchema

# 출력 스키마 정의
response_schemas = [
    ResponseSchema(name="title", description="문서의 제목"),
    ResponseSchema(name="summary", description="3-5문장의 요약"),
    ResponseSchema(name="key_points", description="주요 포인트 리스트"),
    ResponseSchema(name="numbers", description="중요한 수치 데이터"),
    ResponseSchema(name="action_items", description="실행 가능한 액션 아이템")
]

output_parser = StructuredOutputParser.from_response_schemas(response_schemas)
format_instructions = output_parser.get_format_instructions()

json_prompt = PromptTemplate(
    template="다음 문서를 분석하고 JSON 형식으로 출력해주세요.\n{format_instructions}\n\n문서:\n{text}",
    input_variables=["text"],
    partial_variables={"format_instructions": format_instructions}
)

chain = LLMChain(llm=llm, prompt=json_prompt)
result = chain.run(text=document_text)
parsed_result = output_parser.parse(result)

print(parsed_result)
# {'title': '...', 'summary': '...', 'key_points': [...], ...}

3. 엑셀 리포트 자동 생성

비즈니스 환경에서는 엑셀 리포트가 필수지. 이것도 자동화할 수 있어:

import pandas as pd
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment

def create_excel_report(summary_data, filename):
    """엑셀 리포트 생성"""
    wb = Workbook()
    ws = wb.active
    ws.title = "요약 리포트"
    
    # 헤더 스타일
    header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
    header_font = Font(bold=True, color="FFFFFF")
    
    # 제목
    ws['A1'] = "데이터 요약 리포트"
    ws['A1'].font = Font(size=16, bold=True)
    ws.merge_cells('A1:D1')
    
    # 생성 일시
    ws['A2'] = f"생성 일시: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
    
    # 요약 내용
    ws['A4'] = "요약"
    ws['A4'].font = header_font
    ws['A4'].fill = header_fill
    ws['A5'] = summary_data['summary']
    ws.merge_cells('A5:D5')
    
    # 주요 포인트
    ws['A7'] = "주요 포인트"
    ws['A7'].font = header_font
    ws['A7'].fill = header_fill
    
    row = 8
    for point in summary_data['key_points']:
        ws[f'A{row}'] = f"• {point}"
        row += 1
    
    # 열 너비 조정
    ws.column_dimensions['A'].width = 50
    
    wb.save(filename)
    print(f"✅ 엑셀 리포트 생성 완료: {filename}")

LangChain을 실무에서 사용하다 보면 성능과 비용 문제에 부딪히게 돼. 여기 몇 가지 꿀팁을 공유할게! 💰

1. 토큰 사용량 최적화

OpenAI API는 토큰 단위로 과금돼. 불필요한 토큰 사용을 줄이는 게 중요해:

import tiktoken

def count_tokens(text, model="gpt-3.5-turbo"):
    """토큰 수 계산"""
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

def optimize_text_length(text, max_tokens=3000):
    """텍스트 길이 최적화"""
    current_tokens = count_tokens(text)
    
    if current_tokens <= max_tokens:
        return text
    
    # 토큰 수에 맞게 텍스트 자르기
    encoding = tiktoken.encoding_for_model("gpt-3.5-turbo")
    tokens = encoding.encode(text)
    truncated_tokens = tokens[:max_tokens]
    return encoding.decode(truncated_tokens)

# 사용 예시
optimized_text = optimize_text_length(long_document)
print(f"원본: {count_tokens(long_document)} 토큰")
print(f"최적화: {count_tokens(optimized_text)} 토큰")

2. 캐싱 시스템 구현

import hashlib
import json
import os

class SummaryCache:
    def __init__(self, cache_dir="./cache"):
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)
    
    def get_cache_key(self, text):
        """텍스트의 해시값을 캐시 키로 사용"""
        return hashlib.md5(text.encode()).hexdigest()
    
    def get(self, text):
        """캐시에서 요약 가져오기"""
        cache_key = self.get_cache_key(text)
        cache_file = f"{self.cache_dir}/{cache_key}.json"
        
        if os.path.exists(cache_file):
            with open(cache_file, 'r', encoding='utf-8') as f:
                cached_data = json.load(f)
                print("✅ 캐시에서 불러옴!")
                return cached_data['summary']
        return None
    
    def set(self, text, summary):
        """요약 결과를 캐시에 저장"""
        cache_key = self.get_cache_key(text)
        cache_file = f"{self.cache_dir}/{cache_key}.json"
        
        cache_data = {
            'timestamp': datetime.now().isoformat(),
            'summary': summary
        }
        
        with open(cache_file, 'w', encoding='utf-8') as f:
            json.dump(cache_data, f, ensure_ascii=False, indent=2)

# 사용 예시
cache = SummaryCache()

def summarize_with_cache(text):
    # 캐시 확인
    cached_summary = cache.get(text)
    if cached_summary:
        return cached_summary
    
    # 캐시에 없으면 새로 생성
    summary = chain.run(text)
    cache.set(text, summary)
    return summary

3. 병렬 처리로 속도 향상

여러 문서를 동시에 처리하면 시간을 크게 줄일 수 있어:

from concurrent.futures import ThreadPoolExecutor, as_completed

def process_multiple_documents(file_paths, max_workers=5):
    """여러 문서를 병렬로 처리"""
    results = {}
    
    def process_single(file_path):
        try:
            summary = process_document(file_path)
            return file_path, summary
        except Exception as e:
            return file_path, f"오류: {str(e)}"
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(process_single, fp): fp for fp in file_paths}
        
        for future in as_completed(futures):
            file_path, result = future.result()
            results[file_path] = result
            print(f"✅ 완료: {file_path}")
    
    return results

# 사용 예시
files = ["doc1.pdf", "doc2.pdf", "doc3.pdf", "doc4.pdf"]
summaries = process_multiple_documents(files)

for file, summary in summaries.items():
    print(f"\n📄 {file}:")
    print(summary[:200] + "...")

이론은 충분히 배웠으니, 이제 실제로 어떻게 활용할 수 있는지 구체적인 사례를 살펴보자! 🎯

사례 1: 고객 피드백 자동 분석 시스템

매일 수백 개의 고객 리뷰가 쌓이는 서비스가 있다고 가정해보자:

class CustomerFeedbackAnalyzer:
    def __init__(self, llm):
        self.llm = llm
        
        # 감성 분석 프롬프트
        self.sentiment_prompt = PromptTemplate(
            template="""다음 고객 피드백들을 분석해서 JSON 형식으로 출력해주세요:

피드백:
{feedback}

다음 형식으로 출력:
{{
    "overall_sentiment": "긍정/부정/중립",
    "positive_points": ["긍정적 의견1", "긍정적 의견2"],
    "negative_points": ["부정적 의견1", "부정적 의견2"],
    "common_issues": ["공통 이슈1", "공통 이슈2"],
    "action_items": ["개선 사항1", "개선 사항2"]
}}
""",
            input_variables=["feedback"]
        )
    
    def analyze_batch(self, feedbacks):
        """피드백 배치 분석"""
        combined_feedback = "\n\n".join([f"[피드백 {i+1}]\n{fb}" for i, fb in enumerate(feedbacks)])
        
        chain = LLMChain(llm=self.llm, prompt=self.sentiment_prompt)
        result = chain.run(feedback=combined_feedback)
        
        return json.loads(result)
    
    def generate_daily_report(self, feedbacks):
        """일일 리포트 생성"""
        analysis = self.analyze_batch(feedbacks)
        
        report = f"""
📊 고객 피드백 일일 리포트
{'='*50}

📈 전체 감성: {analysis['overall_sentiment']}
총 피드백 수: {len(feedbacks)}개

✅ 긍정적 의견:
{chr(10).join([f"  • {point}" for point in analysis['positive_points']])}

⚠️ 부정적 의견:
{chr(10).join([f"  • {point}" for point in analysis['negative_points']])}

🔍 공통 이슈:
{chr(10).join([f"  • {issue}" for issue in analysis['common_issues']])}

💡 개선 제안:
{chr(10).join([f"  • {item}" for item in analysis['action_items']])}
"""
        return report

# 사용 예시
analyzer = CustomerFeedbackAnalyzer(llm)

daily_feedbacks = [
    "배송이 정말 빨랐어요! 다만 포장이 조금 아쉬웠습니다.",
    "제품 품질은 좋은데 가격이 비싼 것 같아요.",
    "고객센터 응대가 친절했습니다. 만족스러워요!",
    # ... 더 많은 피드백
]

report = analyzer.generate_daily_report(daily_feedbacks)
print(report)

사례 2: 뉴스 기사 자동 큐레이션

여러 뉴스 사이트에서 기사를 수집해서 요약하고 카테고리별로 분류하는 시스템:

from langchain.document_loaders import WebBaseLoader

class NewsAggregator:
    def __init__(self, llm):
        self.llm = llm
        
    def fetch_articles(self, urls):
        """웹에서 기사 수집"""
        articles = []
        for url in urls:
            try:
                loader = WebBaseLoader(url)
                docs = loader.load()
                articles.append({
                    'url': url,
                    'content': docs[0].page_content
                })
            except Exception as e:
                print(f"❌ 오류 ({url}): {str(e)}")
        return articles
    
    def summarize_and_categorize(self, article):
        """기사 요약 및 카테고리 분류"""
        prompt = PromptTemplate(
            template="""다음 뉴스 기사를 분석해주세요:

기사:
{content}

다음 형식으로 출력:
{{
    "category": "정치/경제/사회/문화/IT/스포츠 중 하나",
    "headline": "한 줄 헤드라인",
    "summary": "3문장 요약",
    "keywords": ["키워드1", "키워드2", "키워드3"]
}}
""",
            input_variables=["content"]
        )
        
        chain = LLMChain(llm=self.llm, prompt=prompt)
        result = chain.run(content=article['content'][:2000])  # 토큰 절약
        
        return json.loads(result)
    
    def create_newsletter(self, articles):
        """뉴스레터 생성"""
        categorized = {}
        
        for article in articles:
            analysis = self.summarize_and_categorize(article)
            category = analysis['category']
            
            if category not in categorized:
                categorized[category] = []
            
            categorized[category].append({
                'url': article['url'],
                'headline': analysis['headline'],
                'summary': analysis['summary'],
                'keywords': analysis['keywords']
            })
        
        # HTML 뉴스레터 생성
        newsletter = "<html><body style='font-family: Arial;'>"
        newsletter += f"<h1>📰 오늘의 뉴스 다이제스트 - {datetime.now().strftime('%Y-%m-%d')}</h1>"
        
        for category, items in categorized.items():
            newsletter += f"<h2>📌 {category}</h2>"
            for item in items:
                newsletter += f"""
                <div style='margin: 20px 0; padding: 15px; background: #f8f9fb; border-radius: 8px;'>
                    <h3>{item['headline']}</h3>
                    <p>{item['summary']}</p>
                    <p><small>🏷️ {', '.join(item['keywords'])}</small></p>
                    <a href='{item['url']}'>원문 보기 →</a>
                </div>
                """
        
        newsletter += "</body></html>"
        return newsletter

# 사용 예시
aggregator = NewsAggregator(llm)

news_urls = [
    "https://example.com/news1",
    "https://example.com/news2",
    # ... 더 많은 URL
]

articles = aggregator.fetch_articles(news_urls)
newsletter = aggregator.create_newsletter(articles)

# 이메일로 발송하거나 웹에 게시
with open("newsletter.html", "w", encoding="utf-8") as f:
    f.write(newsletter)

사례 3: 회의록 자동 생성 시스템

음성 회의를 텍스트로 변환한 후 자동으로 회의록을 작성하는 시스템:

class MeetingMinutesGenerator:
    def __init__(self, llm):
        self.llm = llm
        
    def generate_minutes(self, transcript):
        """회의록 생성"""
        prompt = PromptTemplate(
            template="""다음 회의 내용을 바탕으로 공식 회의록을 작성해주세요:

회의 내용:
{transcript}

다음 형식으로 작성:

# 회의록

## 1. 회의 개요
- 참석자:
- 안건:

## 2. 주요 논의 사항
(핵심 내용을 정리)

## 3. 결정 사항
(구체적인 결정 내용)

## 4. 액션 아이템
- [ ] 담당자: 할 일 (기한)

## 5. 다음 회의
- 일시:
- 안건:
""",
            input_variables=["transcript"]
        )
        
        chain = LLMChain(llm=self.llm, prompt=prompt)
        minutes = chain.run(transcript=transcript)
        
        return minutes
    
    def extract_action_items(self, minutes):
        """액션 아이템 추출"""
        prompt = PromptTemplate(
            template="""다음 회의록에서 액션 아이템만 추출해서 JSON 형식으로 출력해주세요:

회의록:
{minutes}

형식:
{{
    "action_items": [
        {{
            "task": "할 일",
            "assignee": "담당자",
            "deadline": "기한",
            "priority": "높음/중간/낮음"
        }}
    ]
}}
""",
            input_variables=["minutes"]
        )
        
        chain = LLMChain(llm=self.llm, prompt=prompt)
        result = chain.run(minutes=minutes)
        
        return json.loads(result)
    
    def send_to_participants(self, minutes, action_items, emails):
        """참석자들에게 회의록 발송"""
        for email in emails:
            # 개인별 액션 아이템 필터링
            personal_actions = [
                item for item in action_items['action_items']
                if email in item['assignee']
            ]
            
            # 이메일 본문 작성
            body = f"""
            <h2>회의록</h2>
            {minutes}
            
            <h2>🎯 당신의 액션 아이템</h2>
            """
            
            for action in personal_actions:
                priority_emoji = "🔴" if action['priority'] == "높음" else "🟡" if action['priority'] == "중간" else "🟢"
                body += f"""
                <div style='margin: 10px 0; padding: 10px; background: #f0f4f8; border-radius: 5px;'>
                    {priority_emoji} <strong>{action['task']}</strong><br>
                    📅 기한: {action['deadline']}
                </div>
                """
            
            # 이메일 발송 (이전에 만든 send_email 함수 활용)
            send_summary_email(body, email)

# 사용 예시
generator = MeetingMinutesGenerator(llm)

# 회의 음성을 텍스트로 변환한 결과
transcript = """
[김대리] 오늘은 신제품 출시 일정에 대해 논의하겠습니다.
[박과장] 개발은 90% 완료되었고, 다음 주까지 QA를 마칠 예정입니다.
[이부장] 좋습니다. 마케팅팀은 준비 상황이 어떤가요?
[최팀장] 홍보 자료는 완성되었고, 광고 캠페인은 다음 달 1일부터 시작합니다.
...
"""

minutes = generator.generate_minutes(transcript)
action_items = generator.extract_action_items(minutes)

print(minutes)
print("\n액션 아이템:")
print(json.dumps(action_items, indent=2, ensure_ascii=False))

LangChain을 실무에 적용할 때 꼭 알아야 할 주의사항들을 정리해봤어. ⚠️

1. 데이터 보안과 프라이버시

import re

def mask_sensitive_data(text):
    """민감한 정보 마스킹"""
    # 이메일 마스킹
    text = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 
                  '***@***.***', text)
    
    # 전화번호 마스킹
    text = re.sub(r'\d{2,3}-\d{3,4}-\d{4}', '***-****-****', text)
    
    # 주민등록번호 마스킹
    text = re.sub(r'\d{6}-\d{7}', '******-*******', text)
    
    return text

2. 에러 핸들링과 재시도 로직

API 호출은 언제든 실패할 수 있어. 견고한 에러 핸들링이 필수야:

from tenacity import retry, stop_after_attempt, wait_exponential

class RobustSummarizer:
    def __init__(self, llm):
        self.llm = llm
    
    @retry(
        stop=stop_after_attempt(3),  # 최대 3번 재시도
        wait=wait_exponential(multiplier=1, min=4, max=10)  # 지수 백오프
    )
    def summarize_with_retry(self, text):
        """재시도 로직이 있는 요약"""
        try:
            chain = load_summarize_chain(self.llm, chain_type="stuff")
            return chain.run(text)
        except Exception as e:
            print(f"⚠️ 오류 발생: {str(e)}")
            raise  # 재시도를 위해 예외를 다시 발생
    
    def safe_summarize(self, text, fallback="요약 생성 실패"):
        """안전한 요약 (실패 시 대체 텍스트 반환)"""
        try:
            return self.summarize_with_retry(text)
        except Exception as e:
            print(f"❌ 최종 실패: {str(e)}")
            return fallback

3. 프롬프트 인젝션 방지

사용자 입력을 그대로 프롬프트에 넣으면 위험할 수 있어:

def sanitize_input(user_input):
    """사용자 입력 정제"""
    # 프롬프트 인젝션 시도 패턴 제거
    dangerous_patterns = [
        r'ignore previous instructions',
        r'disregard all',
        r'forget everything',
        # ... 더 많은 패턴
    ]
    
    for pattern in dangerous_patterns:
        user_input = re.sub(pattern, '', user_input, flags=re.IGNORECASE)
    
    # 길이 제한
    max_length = 5000
    if len(user_input) > max_length:
        user_input = user_input[:max_length]
    
    return user_input

# 사용 예시
user_text = sanitize_input(request.get('text'))
summary = chain.run(user_text)

4. 모니터링과 로깅

프로덕션 환경에서는 모든 것을 로깅하고 모니터링해야 해:

import logging
from datetime import datetime

# 로깅 설정
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('langchain_automation.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

class MonitoredSummarizer:
    def __init__(self, llm):
        self.llm = llm
        self.stats = {
            'total_requests': 0,
            'successful': 0,
            'failed': 0,
            'total_tokens': 0
        }
    
    def summarize(self, text):
        """모니터링이 포함된 요약"""
        start_time = datetime.now()
        self.stats['total_requests'] += 1
        
        try:
            logger.info(f"요약 시작 - 텍스트 길이: {len(text)}")
            
            # 토큰 수 계산
            tokens = count_tokens(text)
            self.stats['total_tokens'] += tokens
            
            # 요약 실행
            summary = chain.run(text)
            
            # 성공 기록
            self.stats['successful'] += 1
            elapsed = (datetime.now() - start_time).total_seconds()
            
            logger.info(f"요약 완료 - 소요 시간: {elapsed:.2f}초, 토큰: {tokens}")
            
            return summary
            
        except Exception as e:
            self.stats['failed'] += 1
            logger.error(f"요약 실패: {str(e)}")
            raise
    
    def get_stats(self):
        """통계 조회"""
        success_rate = (self.stats['successful'] / self.stats['total_requests'] * 100) if self.stats['total_requests'] > 0 else 0
        
        return f"""
📊 시스템 통계
{'='*40}
총 요청: {self.stats['total_requests']}
성공: {self.stats['successful']}
실패: {self.stats['failed']}
성공률: {success_rate:.1f}%
총 토큰 사용: {self.stats['total_tokens']:,}
        """

기본을 넘어서 더 고급 기능들을 알아보자! 🚀

1. 멀티모달 처리: 이미지와 텍스트 함께 분석

GPT-4 Vision을 활용하면 이미지도 함께 분석할 수 있어:

from langchain.chat_models import ChatOpenAI
from langchain.schema.messages import HumanMessage

def analyze_document_with_images(text, image_paths):
    """텍스트와 이미지를 함께 분석"""
    llm = ChatOpenAI(model="gpt-4-vision-preview", max_tokens=1024)
    
    # 이미지를 base64로 인코딩
    import base64
    
    def encode_image(image_path):
        with open(image_path, "rb") as image_file:
            return base64.b64encode(image_file.read()).decode('utf-8')
    
    # 메시지 구성
    content = [
        {"type": "text", "text": f"다음 문서와 이미지를 분석해주세요:\n\n{text}"}
    ]
    
    for img_path in image_paths:
        base64_image = encode_image(img_path)
        content.append({
            "type": "image_url",
            "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}
        })
    
    message = HumanMessage(content=content)
    response = llm([message])
    
    return response.content

# 사용 예시
result = analyze_document_with_images(
    text="이 보고서는 2024년 1분기 실적을 다룹니다.",
    image_paths=["chart1.png", "graph2.png"]
)
print(result)

2. 에이전트 활용: AI가 스스로 판단하고 실행

에이전트는 LangChain의 가장 강력한 기능 중 하나야. AI가 스스로 필요한 도구를 선택해서 작업을 수행해:

from langchain.agents import initialize_agent, Tool
from langchain.agents import AgentType
from langchain.utilities import GoogleSearchAPIWrapper

# 도구 정의
search = GoogleSearchAPIWrapper()

tools = [
    Tool(
        name="Search",
        func=search.run,
        description="최신 정보를 검색할 때 유용합니다."
    ),
    Tool(
        name="Calculator",
        func=lambda x: eval(x),
        description="수학 계산이 필요할 때 사용합니다."
    )
]

# 에이전트 초기화
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 에이전트 실행
result = agent.run("""
2024년 AI 시장 규모를 조사하고, 
2025년 예상 성장률을 적용해서 
2025년 시장 규모를 계산해주세요.
""")

print(result)

3. 커스텀 체인 만들기

복잡한 워크플로우는 커스텀 체인으로 만들 수 있어:

from langchain.chains import SequentialChain

class DocumentProcessingPipeline:
    def __init__(self, llm):
        self.llm = llm
        
        # 1단계: 문서 분류
        self.classify_chain = LLMChain(
            llm=llm,
            prompt=PromptTemplate(
                template="다음 문서의 카테고리를 분류하세요: {document}\n카테고리:",
                input_variables=["document"]
            ),
            output_key="category"
        )
        
        # 2단계: 요약
        self.summarize_chain = LLMChain(
            llm=llm,
            prompt=PromptTemplate(
                template="다음 {category} 문서를 요약하세요: {document}\n요약:",
                input_variables=["document", "category"]
            ),
            output_key="summary"
        )
        
        # 3단계: 액션 아이템 추출
        self.action_chain = LLMChain(
            llm=llm,
            prompt=PromptTemplate(
                template="다음 요약에서 액션 아이템을 추출하세요: {summary}\n액션 아이템:",
                input_variables=["summary"]
            ),
            output_key="actions"
        )
        
        # 체인 연결
        self.pipeline = SequentialChain(
            chains=[self.classify_chain, self.summarize_chain, self.action_chain],
            input_variables=["document"],
            output_variables=["category", "summary", "actions"],
            verbose=True
        )
    
    def process(self, document):
        """전체 파이프라인 실행"""
        return self.pipeline({"document": document})

# 사용 예시
pipeline = DocumentProcessingPipeline(llm)
result = pipeline.process(long_document)

print(f"카테고리: {result['category']}")
print(f"요약: {result['summary']}")
print(f"액션 아이템: {result['actions']}")

여기까지 읽었다면 이미 LangChain 중급자는 된 거야! 이제 이 지식을 어떻게 활용할 수 있을까? 💡

성공적인 프로젝트를 위한 팁

가격 책정 가이드

재능넷에서 서비스를 제공할 때 가격은 이렇게 책정할 수 있어:

물론 프로젝트 규모와 복잡도에 따라 조정이 필요해! 💰

LangChain과 AI 자동화의 미래는 어떻게 될까? 🌟

주목해야 할 트렌드

지금 준비해야 할 것들

이 분야에서 전문가가 되려면:

와, 여기까지 정말 긴 여정이었어! 🎉 LangChain을 활용한 데이터 정리와 요약 자동화에 대해 정말 많은 걸 다뤘지?

핵심을 다시 한 번 정리하면:

이제 당신도 LangChain 전문가가 될 준비가 되었어! 🚀 재능넷에서 이 지식을 활용해서 멋진 서비스를 제공할 수도 있고, 자신의 업무를 자동화해서 생산성을 높일 수도 있지.

기억해, AI는 도구일 뿐이야. 중요한 건 어떤 문제를 해결할 것인가를 정의하는 거야. 기술은 그 문제를 해결하는 수단일 뿐이지. 🎯

자, 이제 당신의 차례야! 오늘 배운 내용을 바탕으로 첫 번째 자동화 프로젝트를 시작해보는 건 어때? 작은 것부터 시작해서 점점 확장해나가면 돼. 💪

이 글이 도움이 되었다면, 당신의 자동화 프로젝트 성공 스토리를 공유해줘!
다음에 또 만나자! 👋