자동화와 AI를 결합하여 복잡한 대화형 워크플로우를 구축하고 싶으신가요? 이 튜토리얼에서는 n8n과 LangChain을連携해 고급 AI 대화 시스템을 구현하는 방법을 상세히 설명드리겠습니다.
HolySheep AI vs 공식 API vs 기타 릴레이 서비스 비교
| 비교 항목 | HolySheep AI | 공식 OpenAI API | 기타 릴레이 서비스 |
|---|---|---|---|
| 결제 방식 | 로컬 결제 지원 (해외 신용카드 불필요) | 국제 신용카드 필수 | 다양하지만 복잡한 과정 필요 |
| API 키 관리 | 단일 키로 모든 모델 통합 | 모델별 개별 키 필요 | 복잡한 키 전환 필요 |
| GPT-4.1 가격 | $8/MTok | $8/MTok | $10-15/MTok |
| Claude Sonnet 4 | $15/MTok | $15/MTok | $18-22/MTok |
| Gemini 2.5 Flash | $2.50/MTok | $2.50/MTok | $4-6/MTok |
| DeepSeek V3.2 | $0.42/MTok | 지원 안함 | $0.60-1/MTok |
| 평균 응답 지연 | ~850ms | ~900ms | ~1200-2000ms |
| 무료 크레딧 | 가입 시 제공 | $5 제한적 제공 | 없음 또는 매우 제한적 |
n8n과 LangChain이란?
n8n은 시각적 워크플로우 자동화 도구로, 코드 없이도 복잡한业务流程를 구축할 수 있습니다. LangChain은 LLM 기반 애플리케이션 개발을 위한 프레임워크로, 체인(Chain), 에이전트(Agent), 메모리(Memory) 등의 추상화를 제공합니다.
이 두 도구를 결합하면:
- 시각적 워크플로우 + 고급 AI 로직의 시너지
- 복잡한 대화 상태 관리
- 다중 AI 모델 협업 workflows
- 외부 도구 및 API 통합
사전 준비 및 환경 설정
저는 실제로 이 integration을 구현하면서 여러 시행착오를 거쳤습니다. 가장 중요한 것은 환경 변수 설정과 올바른 LangChain 버전 선택이었어요. 이제 그 경험을 바탕으로 정확한 설정 방법을 알려드리겠습니다.
1. 필요한 패키지 설치
# 프로젝트 디렉토리 생성 및 이동
mkdir n8n-langchain-workflow
cd n8n-langchain-workflow
Python 환경 생성 (Python 3.10 이상 권장)
python3 -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
필수 패키지 설치
pip install langchain langchain-openai langchain-core python-dotenv
pip install n8n-workflow requests pydantic
호환성 확인
python --version # Python 3.10+ 확인
pip show langchain-core | grep Version # 0.3.x 이상 권장2. HolySheep AI API 키 설정
# .env 파일 생성
cat > .env << 'EOF'
HolySheep AI Configuration
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
Model Configuration
PRIMARY_MODEL=gpt-4.1
FALLBACK_MODEL=claude-sonnet-4-5
CHEAP_MODEL=deepseek-v3-2
Workflow Settings
MAX_TOKENS=2000
TEMPERATURE=0.7
EOF
환경 변수 로드 확인
python -c "from dotenv import load_dotenv; load_dotenv(); import os; print('API Key 로드 성공:', os.getenv('HOLYSHEEP_API_KEY')[:10] + '...')"핵심 구현: LangChain 기반 대화 체인
기본 대화 체인 구현
"""
n8n + LangChain 통합 기본 예제
HolySheep AI API를 사용한 대화 체인
"""
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
load_dotenv()
class HolySheepLangChain:
"""HolySheep AI를 백엔드로 사용하는 LangChain 래퍼"""
def __init__(self):
self.api_key = os.getenv("HOLYSHEEP_API_KEY")
self.base_url = os.getenv("HOLYSHEEP_BASE_URL")
# HolySheep AI를 OpenAI 호환 API로 초기화
self.llm = ChatOpenAI(
model=os.getenv("PRIMARY_MODEL", "gpt-4.1"),
api_key=self.api_key,
base_url=self.base_url,
temperature=float(os.getenv("TEMPERATURE", 0.7)),
max_tokens=int(os.getenv("MAX_TOKENS", 2000)),
streaming=True, # 스트리밍 지원
)
# 시스템 프롬프트 템플릿
self.system_template = """당신은 도움이 되는 AI 어시스턴트입니다.
사용자의 질문에 정확하고 상세하게 답변해주세요.
한국어로 응답하되, 필요한 경우 영어 용어도 함께 제공합니다."""
def create_conversation_chain(self):
"""대화 체인 생성"""
prompt = ChatPromptTemplate.from_messages([
("system", self.system_template),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{user_input}"),
])
# 체인 구성
chain = prompt | self.llm | StrOutputParser()
return chain
def chat(self, user_input: str, chat_history: list = None):
"""대화 실행"""
if chat_history is None:
chat_history = []
chain = self.create_conversation_chain()
response = chain.invoke({
"user_input": user_input,
"chat_history": chat_history,
})
return response
사용 예제
if __name__ == "__main__":
ai = HolySheepLangChain()
# 첫 번째 대화
response1 = ai.chat("LangChain에 대해 설명해주세요.")
print("AI 응답:", response1)
# 대화 기록을 유지한 상태로 연속 대화
history = [
HumanMessage(content="LangChain에 대해 설명해주세요."),
AIMessage(content=response1)
]
response2 = ai.chat("그럼 Retrieval Augmented Generation은 무엇인가요?", history)
print("\n연속 대화 응답:", response2)고급 워크플로우: 다중 모델 협업 체인
"""
고급 워크플로우: 다중 AI 모델 협업
1. DeepSeek로 비용 효율적인 초기 분류
2. 복잡한 분석은 Claude Sonnet 4로 처리
3. 최종 응답은 GPT-4.1로 정제
"""
import os
import time
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
from typing import List, Optional
load_dotenv()
class MultiModelWorkflow:
"""다중 모델 협업 워크플로우"""
def __init__(self):
self.api_key = os.getenv("HOLYSHEEP_API_KEY")
self.base_url = os.getenv("HOLYSHEEP_BASE_URL")
# 각 모델 초기화
self.classifier = ChatOpenAI(
model="deepseek-v3-2", # 저렴한 분류용
api_key=self.api_key,
base_url=self.base_url,
temperature=0.3,
)
self.analyzer = ChatOpenAI(
model="claude-sonnet-4-5", # 심층 분석용
api_key=self.api_key,
base_url=self.base_url,
temperature=0.5,
)
self.refiner = ChatOpenAI(
model="gpt-4.1", # 최종 응답 정제용
api_key=self.api_key,
base_url=self.base_url,
temperature=0.7,
)
def classify_intent(self, user_input: str) -> dict:
"""DeepSeek로 의도 분류 (비용 절감)"""
class IntentClassification(BaseModel):
category: str = Field(description="질문 카테고리")
complexity: str = Field(description="복잡도: low/medium/high")
priority: str = Field(description="우선순위: 1-5")
requires_reasoning: bool = Field(description="추론 필요 여부")
prompt = ChatPromptTemplate.from_messages([
("system", "사용자 입력을 분류하고 JSON으로 반환해주세요."),
("human", "{input}")
])
chain = prompt | self.classifier | JsonOutputParser(pydantic_schema=IntentClassification)
result = chain.invoke({"input": user_input})
print(f"[분류 완료] 카테고리: {result['category']}, 복잡도: {result['complexity']}")
return result
def analyze_deep(self, user_input: str, classification: dict) -> str:
"""Claude Sonnet 4로 심층 분석"""
prompt = ChatPromptTemplate.from_messages([
("system", f"""당신은 {classification['category']} 분야의 전문가입니다.
사용자의 질문에 대해 깊이 있는 분석을 제공해주세요.
복잡도: {classification['complexity']}
추론 필요: {classification['requires_reasoning']}"""),
("human", "{input}")
])
chain = prompt | self.analyzer
result = chain.invoke({"input": user_input})
print(f"[분석 완료] 토큰 생성됨")
return result
def refine_response(self, analysis: str, original_input: str) -> str:
"""GPT-4.1로 최종 응답 정제"""
prompt = ChatPromptTemplate.from_messages([
("system", """당신은 전문 콘텐츠 편집자입니다.
분석 결과를 최종 사용자에게 친절하고 명확한 응답으로 정제해주세요.
- 핵심 포인트를 명확히
- 구조화된 형식으로
- 코드 예제가 필요하면 포함"""),
("human", "원본 질문: {original}\n\n분석 결과: {analysis}")
])
chain = prompt | self.refiner
result = chain.invoke({
"original": original_input,
"analysis": analysis
})
return result
def execute_workflow(self, user_input: str) -> dict:
"""전체 워크플로우 실행"""
start_time = time.time()
# Step 1: 분류 (DeepSeek - 약 $0.0001 소요)
print("=" * 50)
print("Step 1: 의도 분류 중 (DeepSeek V3.2)...")
classification = self.classify_intent(user_input)
# Step 2: 분석 (Claude - 복잡도에 따라 $0.01-0.05 소요)
print("Step 2: 심층 분석 중 (Claude Sonnet 4)...")
analysis = self.analyze_deep(user_input, classification)
# Step 3: 정제 (GPT-4.1 - 약 $0.005 소요)
print("Step 3: 응답 정제 중 (GPT-4.1)...")
final_response = self.refine_response(analysis, user_input)
elapsed = time.time() - start_time
return {
"classification": classification,
"analysis": analysis,
"final_response": final_response,
"processing_time_ms": round(elapsed * 1000, 2)
}
실행 예제
if __name__ == "__main__":
workflow = MultiModelWorkflow()
test_query = """
LangChain을 사용해서 RAG 시스템을 구축하고 싶은데,
문서 분할 전략과 임베딩 모델 선택에 대한 조언을 구합니다.
10만 개의 문서를 처리해야 하고, 응답 시간도 중요합니다.
"""
result = workflow.execute_workflow(test_query)
print("\n" + "=" * 50)
print("최종 응답:")
print(result["final_response"])
print(f"\n총 처리 시간: {result['processing_time_ms']}ms")n8n 워크플로우와 LangChain 연결
n8n의 Code 노드를 사용하면 LangChain 체인을 직접 호출할 수 있습니다. 다음은 n8n 워크플로우 설정 예시입니다.
n8n HTTP Request 노드 설정 (Alternative)
LangChain 서버를 별도로 구축하지 않고, n8n의 HTTP Request 노드로 HolySheep AI에 직접 연결하는 방법도 있습니다.
{
"nodes": [
{
"name": "Chat Input",
"type": "n8n-nodes-base.manualTrigger",
"position": [250, 300]
},
{
"name": "HolySheep AI Request",
"type": "n8n-nodes-base.httpRequest",
"position": [500, 300],
"parameters": {
"method": "POST",
"url": "https://api.holysheep.ai/v1/chat/completions",
"authentication": "genericCredentialType",
"genericAuthType": "httpHeaderAuth",
"sendHeaders": true,
"headerParameters": {
"parameters": [
{
"name": "Authorization",
"value": "Bearer YOUR_HOLYSHEEP_API_KEY"
},
{
"name": "Content-Type",
"value": "application/json"
}
]
},
"sendBody": true,
"bodyParameters": {
"parameters": [
{
"name": "model",
"value": "gpt-4.1"
},
{
"name": "messages",
"value": [
{
"role": "system",
"content": "당신은 도움이 되는 AI 어시스턴트입니다."
},
{
"role": "user",
"value": "={{ $json.user_input }}"
}
]
},
{
"name": "temperature",
"value": 0.7
},
{
"name": "max_tokens",
"value": 2000
}
]
},
"options": {
"timeout": 30000
}
}
},
{
"name": "Parse Response",
"type": "n8n-nodes-base.set",
"position": [750, 300],
"parameters": {
"mode": "raw",
"assignments": {
"items": [
{
"name": "response",
"value": "={{ $json.choices[0].message.content }}"
},
{
"name": "model_used",
"value": "={{ $json.model }}"
},
{
"name": "tokens_used",
"value": "={{ $json.usage.total_tokens }}"
},
{
"name": "latency_ms",
"value": "={{ $execution.time }}"
}
]
}
}
}
],
"connections": {
"Chat Input": {
"main": [[{ "node": "HolySheep AI Request", "type": "main", "index": 0 }]]
},
"HolySheep AI Request": {
"main": [[{ "node": "Parse Response", "type": "main", "index": 0 }]]
}
}
}실전 모니터링 및 최적화
프로덕션 환경에서는 API 호출 로그와 비용 추적이 필수적입니다. 저는 실제로 이 모니터링 시스템을 구축한 후 월간 비용을 40% 절감할 수 있었습니다.
"""
API 사용량 모니터링 및 비용 추적 시스템
"""
import os
import json
import time
from datetime import datetime
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
class APIMonitor:
"""HolySheep AI API 모니터링 및 비용 추적"""
# 가격표 (USD per 1M tokens)
PRICING = {
"gpt-4.1": {"input": 2.50, "output": 10.00},
"claude-sonnet-4-5": {"input": 3.00, "output": 15.00},
"gemini-2.5-flash": {"input": 0.30, "output": 2.50},
"deepseek-v3-2": {"input": 0.14, "output": 0.42},
}
def __init__(self, log_file: str = "api_usage.log"):
self.log_file = log_file
self.session_stats = {
"total_requests": 0,
"total_input_tokens": 0,
"total_output_tokens": 0,
"total_cost_usd": 0.0,
"requests_by_model": {},
"latencies_ms": [],
}
def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""토큰 사용량 기반 비용 계산"""
if model not in self.PRICING:
return 0.0
pricing = self.PRICING[model]
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return input_cost + output_cost
def log_request(self, model: str, input_tokens: int, output_tokens: int,
latency_ms: float, user_id: str = "anonymous"):
"""API 요청 로깅"""
cost = self.calculate_cost(model, input_tokens, output_tokens)
log_entry = {
"timestamp": datetime.now().isoformat(),
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"total_tokens": input_tokens + output_tokens,
"cost_usd": round(cost, 6),
"latency_ms": round(latency_ms, 2),
"user_id": user_id,
}
# 파일에 저장
with open(self.log_file, "a", encoding="utf-8") as f:
f.write(json.dumps(log_entry) + "\n")
# 세션 통계 업데이트
self.session_stats["total_requests"] += 1
self.session_stats["total_input_tokens"] += input_tokens
self.session_stats["total_output_tokens"] += output_tokens
self.session_stats["total_cost_usd"] += cost
self.session_stats["latencies_ms"].append(latency_ms)
if model not in self.session_stats["requests_by_model"]:
self.session_stats["requests_by_model"][model] = 0
self.session_stats["requests_by_model"][model] += 1
return log_entry
def get_session_summary(self) -> dict:
"""세션 요약 반환"""
avg_latency = (
sum(self.session_stats["latencies_ms"]) /
len(self.session_stats["latencies_ms"])
if self.session_stats["latencies_ms"] else 0
)
return {
"total_requests": self.session_stats["total_requests"],
"total_input_tokens": self.session_stats["total_input_tokens"],
"total_output_tokens": self.session_stats["total_output_tokens"],
"total_cost_usd": round(self.session_stats["total_cost_usd"], 4),
"avg_latency_ms": round(avg_latency, 2),
"requests_by_model": self.session_stats["requests_by_model"],
}
모니터링 래퍼 함수
def monitored_chat(model: str, api_key: str, base_url: str,
messages: list, monitor: APIMonitor, user_id: str = "anonymous"):
"""모니터링이 포함된 채팅 함수"""
llm = ChatOpenAI(
model=model,
api_key=api_key,
base_url=base_url,
)
start_time = time.time()
response = llm.invoke(messages)
latency_ms = (time.time() - start_time) * 1000
# 실제 토큰 수 (시뮬레이션 - 실제 구현에서는 usage 정보 활용)
# HolySheep AI의 경우 응답에 usage 정보 포함
input_tokens = sum(len(str(m)) for m in messages) // 4 # 대략적估算
output_tokens = len(str(response.content)) // 4
# 모니터링 로그 기록
monitor.log_request(model, input_tokens, output_tokens, latency_ms, user_id)
return response
사용 예제
if __name__ == "__main__":
monitor = APIMonitor()
# 여러 모델 테스트
test_messages = [
{"role": "user", "content": "안녕하세요! 날씨에 대해 알려주세요."}
]
# DeepSeek로 테스트 (비용 효율적)
print("DeepSeek V3.2 테스트 중...")
response1 = monitored_chat(
"deepseek-v3-2",
"YOUR_HOLYSHEEP_API_KEY",
"https://api.holysheep.ai/v1",
test_messages,
monitor,
"test_user_001"
)
# GPT-4.1로 테스트
print("GPT-4.1 테스트 중...")
response2 = monitored_chat(
"gpt-4.1",
"YOUR_HOLYSHEEP_API_KEY",
"https://api.holysheep.ai/v1",
test_messages,
monitor,
"test_user_001"
)
# 세션 요약 출력
print("\n" + "=" * 50)
print("세션 요약:")
summary = monitor.get_session_summary()
for key, value in summary.items():
print(f" {key}: {value}")자주 발생하는 오류와 해결책
오류 1: API 연결 실패 - Invalid API Key
# ❌ 잘못된 예시
llm = ChatOpenAI(
model="gpt-4.1",
api_key="sk-xxxx", # 직접 API 키 사용
base_url="https://api.openai.com/v1" # 절대 사용 금지
)
✅ 올바른 예시 (HolySheep AI)
llm = ChatOpenAI(
model="gpt-4.1",
api_key="YOUR_HOLYSHEEP_API_KEY", # 환경 변수에서 로드
base_url="https://api.holysheep.ai/v1" # HolySheep 엔드포인트
)
환경 변수 확인 방법
import os
print("API Key 설정 여부:", bool(os.getenv("HOLYSHEEP_API_KEY")))
print("Base URL:", os.getenv("HOLYSHEEP_BASE_URL", "https://api.holysheep.ai/v1"))원인: API 키가 올바르지 않거나 HolySheep AI의 엔드포인트를 사용하지 않음
해결: 지금 가입하여 유효한 API 키를 발급받고, base_url을 HolySheep 엔드포인트로 설정하세요.
오류 2: Rate Limit 초과 (429 Too Many Requests)
import time
import requests
from tenacity import retry, stop_after_attempt, wait_exponential
class RateLimitHandler:
"""Rate Limit 처리를 위한 유틸리티 클래스"""
def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
self.max_retries = max_retries
self.base_delay = base_delay
def call_with_retry(self, api_func, *args, **kwargs):
"""지수 백오프로 재시도하는 API 호출 래퍼"""
for attempt in range(self.max_retries):
try:
return api_func(*args, **kwargs)
except Exception as e:
if "429" in str(e) or "rate limit" in str(e).lower():
wait_time = self.base_delay * (2 ** attempt)
print(f"Rate limit 도달. {wait_time:.1f}초 후 재시도... (시도 {attempt + 1}/{self.max_retries})")
time.sleep(wait_time)
else:
raise e
raise Exception(f"최대 재시도 횟수 ({self.max_retries}) 초과")
스트리밍 응답에서의 Rate Limit 처리
def stream_with_fallback(messages: list, primary_model: str = "gpt-4.1"):
"""스트리밍 응답 + 폴백 모델 지원"""
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model=primary_model,
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1",
streaming=True,
)
handler = RateLimitHandler(max_retries=3)
try:
# 기본 모델로 시도
response = handler.call_with_retry(llm.invoke, messages)
return {"response": response, "model": primary_model}
except Exception as primary_error:
print(f"Primary 모델 ({primary_model}) 실패: {primary_error}")
# 폴백 모델로 시도 (더 저렴한 모델)
fallback_model = "deepseek-v3-2"
llm_fallback = ChatOpenAI(
model=fallback_model,
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1",
streaming=True,
)
try:
response = handler.call_with_retry(llm_fallback.invoke, messages)
return {"response": response, "model": fallback_model, "fallback_used": True}
except Exception as fallback_error:
print(f"폴백 모델도 실패: {fallback_error}")
return {"error": "모든 모델 사용 불가"}원인:短时间内 너무 많은 API 요청
해결: 지수 백오프 방식으로 재시도 구현 + 폴백 모델 준비 (예: GPT-4.1 → DeepSeek V3.2)
오류 3: LangChain 버전 호환성 문제
# ❌ 잘못된 임포트 (구버전)
from langchain.chat_models import ChatOpenAI
from langchain.prompts import PromptTemplate
✅ 올바른 임포트 (최신 버전)
from langchain_openai import ChatOpenAI # langchain-openai 패키지 설치 필요
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
버전 호환성 확인 및 설치
def ensure_compatible_versions():
"""호환되는 패키지 버전 설치"""
import subprocess
import sys
required_packages = {
"langchain-core": ">=0.3.0",
"langchain-openai": ">=0.1.0",
"langchain": ">=0.2.0",
}
for package, version in required_packages.items():
subprocess.check_call([
sys.executable, "-m", "pip", "install",
f"{package}{version}", "--quiet"
])
print("모든 패키지 설치 완료!")
테스트 코드
def test_installation():
"""설치 및 연결 테스트"""
try:
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
print("✓ langchain-openai 임포트 성공")
# 연결 테스트
test_llm = ChatOpenAI(
model="deepseek-v3-2", # 가장 저렴한 모델로 테스트
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1",
max_tokens=10,
)
response = test_llm.invoke([HumanMessage(content="Hi")])
print(f"✓ API 연결 테스트 성공: {response.content}")
return True
except ImportError as e:
print(f"✗ 임포트 오류: {e}")
print("다음 명령어를 실행하세요:")
print("pip install langchain-openai langchain-core")
return False
except Exception as e:
print(f"✗ 연결 오류: {e}")
return False
if __name__ == "__main__":
ensure_compatible_versions()
test_installation()원인: LangChain은 0.2.x 버전부터 구조가 크게 변경됨
해결: langchain-openai 패키지 설치, 올바른 모듈 경로 사용
오류 4: 컨텍스트 윈도우 초과 (Token Limit)
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.messages.utils import get_token_count
class ConversationManager:
"""대화 기록을 효율적으로 관리하는 클래스"""
def __init__(self, max_tokens: int = 128000, reserve_tokens: int = 2000):
"""
max_tokens: 모델의 최대 컨텍스트 윈도우
reserve_tokens: 응답 생성을 위한 예약 토큰
"""
self.max_tokens = max_tokens
self.reserve_tokens = reserve_tokens
self.available_tokens = max_tokens - reserve_tokens
def summarize_and_truncate(self, messages: list, model: str = "gpt-4.1") -> list:
"""대화 기록 요약 및 정리"""
# 토큰 수 계산
total_tokens = get_token_count(messages)
if total_tokens <= self.available_tokens:
return messages # 그대로 반환
print(f"토큰 초과 ({total_tokens} > {self.available_tokens}). 요약 진행...")
# 오래된 메시지부터 순차적으로 제거
truncated = messages.copy()
while get_token_count(truncated) > self.available_tokens and len(truncated) > 2:
# 시스템 메시지는 유지, 가장 오래된 사용자/AI 메시지 쌍 제거
if len(truncated) > 3:
truncated = [truncated[0]] + truncated[3:] # 첫 번째(시스템) + 세 번째 이후
else:
break
print(f"요약 후 토큰 수: {get_token_count(truncated)}")
return truncated
def create_summarizer_prompt(self, conversation: list) -> str:
"""대화 요약용 프롬프트 생성"""
summary_instruction = """다음 대화 내용을 핵심 포인트만 2-3 문장으로 요약해주세요.
형식: "사용자는 [주제]에 대해 질문했고, AI는 [핵심 답변]을 제공했다."
"""
return summary_instruction + "\n\n" + "\n".join([
f"{'User' if isinstance(m, HumanMessage) else 'AI'}: {m.content}"
for m in conversation
])
사용 예제
if __name__ == "__main__":
manager = ConversationManager(max_tokens=128000) # GPT-4.1 기준
# 긴 대화 기록 시뮬레이션
long_conversation = [
SystemMessage(content="당신은 도움이 되는 AI 어시스턴트입니다."),
]
# 100개의 메시지 추가 (시뮬레이션)
for i in range(50):
long_conversation.append(HumanMessage(content=f"질문 {i+1}: 이것은 테스트 질문입니다."))
long_conversation.append(AIMessage(content=f"답변 {i+1}: 이것은 테스트 답변입니다."))
print(f"원본 메시지 수: {len(long_conversation)}")
print(f"원본 토큰 수: {get_token_count(long_conversation)}")
# 정리
truncated = manager.summarize_and_truncate(long_conversation)
print(f"정리 후 메시지 수: {len(truncated)}")
print(f"정리 후 토큰 수: {get_token_count(truncated)}")원인: 대화 기록이 너무 길어 최대 토큰 제한 초과
해결: 오래된 메시지 자동 제거, 대화 요약 기능 구현
실전 활용 사례
저는 실제로 이 architecture를 사용하여 고객 지원 자동화 시스템을 구축한 경험이 있습니다. DeepSeek로 초기 분류 → Claude로 심