이 글에서 배울 내용:
- 데이터베이스에 대화를 저장하는 생산 환경용 AI 에이전트 구축 방법
- 지능형 데이터 추출 및 엔티티 추적 구현 방법
- 자동 복구 기능을 통한 견고한 오류 처리 구현 방법
- Bright Data의 실시간 웹 데이터로 에이전트 성능 향상하는 방법
시작해 보겠습니다!
상태 비저장형 AI 대화의 한계
현재 AI 에이전트는 일반적으로 상태 비저장 시스템으로 작동합니다. 각 대화를 별개의 이벤트로 처리하죠. 이러한 역사적 맥락의 부재로 인해 사용자는 정보를 반복해야 합니다. 결과적으로 운영 비효율성과 사용자 불만을 초래합니다. 또한 기업은 개인화나 서비스 개선을 위한 장기적 데이터 활용 기회를 놓치게 됩니다.
데이터 지속형 AI는 모든 상호작용을 구조화된 데이터베이스에 기록함으로써 이 문제를 해결합니다. 지속적인 기록을 유지함으로써 이러한 시스템은 역사적 맥락을 기억하고, 시간에 따라 특정 개체를 추적하며, 과거 상호작용 패턴을 활용하여 일관되고 개인화된 사용자 경험을 제공할 수 있습니다.
구축 목표: 데이터베이스 연동 AI 에이전트 시스템
LangChain과 GPT-4를 활용해 메시지를 처리하는 실전형 AI 에이전트를 구축할 예정입니다. 각 대화 내용을 PostgreSQL에 저장하며, 실시간으로 엔티티와 인사이트를 추출합니다. 세션 간 완전한 대화 이력을 유지하고, 자동 재시도 시스템으로 오류를 관리합니다. 로깅을 통한 모니터링 기능을 제공합니다.
이 시스템은 다음을 처리합니다:
- 적절한 관계와 인덱스를 갖춘 데이터베이스 스키마
- 커스텀 데이터베이스 도구를 갖춘 LangChain 에이전트
- 자동 대화 지속성 및 엔티티 추출
- 데이터 수집을 위한 백그라운드 처리 파이프라인
- 트랜잭션 관리를 통한 오류 처리
- 과거 데이터 검색을 위한 쿼리 인터페이스
- 웹 인텔리전스를 위한 Bright Data와의 RAG 통합
필수 조건
개발 환경 설정:
- Python 3.10 이상. 최신 비동기 기능 및 타입 힌트에 필수
- PostgreSQL 14+ 또는 SQLite 3.35+. 데이터 지속성을 위한 데이터베이스
- OpenAI API 키. GPT-4 접근을 위해 필요합니다. OpenAI 플랫폼에서 획득하세요
- LangChain. AI 에이전트 구축용 프레임워크. 문서를 참조하세요
- Python 가상 환경. 종속성을 격리합니다.
venv문서를 참조하세요
환경 설정
프로젝트 디렉터리를 생성하고 종속성을 설치하세요:
mkdir database-agent
cd database-agent
python -m venv venv
# macOS/Linux: source venv/bin/activate
# Windows: venv\Scripts\activate
pip install langchain langchain-openai sqlalchemy psycopg2-binary python-dotenv pydanticagent. py라는 새 파일을 생성하고 다음 임포트를 추가하세요:
import os
import json
import logging
import time
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional
from queue import Queue
from threading import Thread
# SQLAlchemy 임포트
from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, Float, JSON, ForeignKey, text
from sqlalchemy.orm import sessionmaker, relationship, Session, declarative_base
from sqlalchemy.pool import QueuePool
from sqlalchemy.exc import SQLAlchemyError
# LangChain 임포트
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.tools import Tool
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain.schema import HumanMessage, AIMessage, SystemMessage
# RAG 임포트
from langchain_community.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
import requests
# 환경 설정
from dotenv import load_dotenv
load_dotenv()
# 로깅 구성
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)인증 정보가 포함된 .env 파일을 생성하세요:
# 데이터베이스 구성
DATABASE_URL="postgresql://username:password@localhost:5432/agent_db"
# 또는 SQLite의 경우: DATABASE_URL="sqlite:///./agent_data.db"
# API 키
OPENAI_API_KEY="your-openai-api-key"
# 선택 사항: Bright Data (7단계용)
BRIGHT_DATA_API_KEY="your-bright-data-api-key"
# 애플리케이션 설정
AGENT_MODEL="gpt-4-turbo-preview"
CONNECTION_POOL_SIZE=5
MAX_RETRIES=3필요한 사항:
- 데이터베이스 URL: PostgreSQL 또는 SQLite 연결 문자열
- OpenAI API 키: GPT-4를 통한 에이전트 인텔리전스용
- Bright Data API 키: 선택 사항, 단계 7의 실시간 웹 데이터용
데이터베이스 연결 AI 에이전트 구축
1단계: 데이터베이스 스키마 설계
사용자, 대화, 메시지 및 추출된 엔티티를 위한 테이블을 설계합니다. 스키마는 데이터 무결성을 유지하기 위해 외래 키와 관계를 사용합니다.
Base = declarative_base()
class User(Base):
"""사용자 프로필 테이블 - 사용자 정보 및 선호도를 저장합니다."""
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
user_id = Column(String(255), unique=True, nullable=False, index=True)
name = Column(String(255))
email = Column(String(255))
preferences = Column(JSON, default={})
created_at = Column(DateTime, default=datetime.utcnow)
last_active = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# 관계
conversations = relationship("Conversation", back_populates="user", cascade="all, delete-orphan")
def __repr__(self):
return f"<User(user_id='{self.user_id}', name='{self.name}')>"
class Conversation(Base):
"""대화 세션 테이블 - 개별 대화 세션을 추적합니다."""
__tablename__ = 'conversations'
id = Column(Integer, primary_key=True)
conversation_id = Column(String(255), unique=True, nullable=False, index=True)
user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
title = Column(String(500))
summary = Column(Text)
status = Column(String(50), default='active') # 활성, 보관됨, 삭제됨
meta_data = Column(JSON, default={})
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# 관계
user = relationship("User", back_populates="conversations")
messages = relationship("Message", back_populates="conversation", cascade="all, delete-orphan")
entities = relationship("Entity", back_populates="conversation", cascade="all, delete-orphan")
def __repr__(self):
return f"<Conversation(id='{self.conversation_id}', user='{self.user_id}')>"
class Message(Base):
"""개별 메시지 테이블 - 대화 내 각 메시지를 저장합니다."""
__tablename__ = 'messages'
id = Column(Integer, primary_key=True)
conversation_id = Column(Integer, ForeignKey('conversations.id'), nullable=False, index=True)
role = Column(String(50), nullable=False) # user, assistant, system
content = Column(Text, nullable=False)
tokens = Column(Integer)
model = Column(String(100))
meta_data = Column(JSON, default={})
created_at = Column(DateTime, default=datetime.utcnow)
# 관계
conversation = relationship("Conversation", back_populates="messages")
def __repr__(self):
return f"<Message(role='{self.role}', conversation='{self.conversation_id}')>"
class Entity(Base):
"""추출된 엔티티 테이블 - 대화에서 추출된 명명된 엔티티를 저장합니다."""
__tablename__ = 'entities'
id = Column(Integer, primary_key=True)
conversation_id = Column(Integer, ForeignKey('conversations.id'), nullable=False, index=True)
entity_type = Column(String(100), nullable=False, index=True) # person, organization, location, etc.
entity_value = Column(String(500), nullable=False)
context = Column(Text)
confidence = Column(Float, default=0.0)
meta_data = Column(JSON, default={})
extracted_at = Column(DateTime, default=datetime.utcnow)
# 관계
conversation = relationship("Conversation", back_populates="entities")
def __repr__(self):
return f"<Entity(type='{self.entity_type}', value='{self.entity_value}')>"
class AgentLog(Base):
"""에이전트 운영 로그 테이블 - 모니터링을 위한 운영 로그 저장."""
__tablename__ = 'agent_logs'
id = Column(Integer, primary_key=True)
conversation_id = Column(String(255), index=True)
level = Column(String(50), nullable=False) # INFO, WARNING, ERROR
operation = Column(String(255), nullable=False)
message = Column(Text, nullable=False)
error_details = Column(JSON)
execution_time = Column(Float) # 초 단위
created_at = Column(DateTime, default=datetime.utcnow)
def __repr__(self):
return f"<AgentLog(level='{self.level}', operation='{self.operation}')>"스키마는 다섯 개의 핵심 테이블을 정의합니다. User 테이블은 유연한 데이터를 위한 JSON 환경설정이 포함된 프로필을 저장합니다. Conversation 테이블은 상태 추적과 함께 세션을 기록합니다. Message 테이블은 사용자 대 어시스턴트 메시지를 구분하는 역할 표시자와 함께 개별 교류를 보관합니다. Entity 테이블은 신뢰도 점수와 함께 추출된 정보를 캡처합니다. AgentLog 테이블은 모니터링을 위한 작업 추적을 제공합니다. 외래 키는 참조 무결성을 유지합니다. 자주 쿼리되는 필드에 대한 인덱스는 성능을 최적화합니다. cascade="all, delete-orphan" 설정은 부모 레코드가 삭제될 때 관련 레코드를 정리합니다.
2단계: 데이터베이스 연결 계층 설정
SQLAlchemy를 사용하여 데이터베이스 연결 관리자를 구성합니다. 이 관리자는 연결 풀링, 상태 점검 및 자동 재시도 로직을 처리하여 안정성을 보장합니다.
class DatabaseManager:
"""
데이터베이스 연결 및 작업을 관리합니다.
기능:
- 효율적인 리소스 사용을 위한 연결 풀링
- 데이터베이스 연결성 확인을 위한 상태 점검
- 자동 테이블 생성
"""
def __init__(self, database_url: str, pool_size: int = 5, max_retries: int = 3):
"""
데이터베이스 관리자 초기화.
인수:
database_url: 데이터베이스 연결 문자열 (예: 'sqlite:///./agent_data.db')
pool_size: 풀에 유지할 연결 수
max_retries: 실패한 작업에 대한 최대 재시도 횟수
"""
self.database_url = database_url
self.max_retries = max_retries
# 연결 풀링을 지원하는 엔진 생성
self.engine = create_engine(
database_url,
poolclass=QueuePool,
pool_size=pool_size,
max_overflow=10,
pool_pre_ping=True, # 사용 전 연결 확인
echo=False # SQL 디버깅 시 True로 설정
)
# 세션 팩토리 생성
self.SessionLocal = sessionmaker(
bind=self.engine,
autocommit=False,
autoflush=False
)
logger.info(f"✓ {pool_size} 연결 풀로 데이터베이스 엔진 생성 완료")
def initialize_database(self):
"""데이터베이스 내 모든 테이블 생성"""
try:
Base.metadata.create_all(bind=self.engine)
logger.info("✓ 데이터베이스 테이블 생성 성공")
except Exception as e:
logger.error(f"❌ 데이터베이스 테이블 생성 실패: {e}")
raise
def get_session(self) -> Session:
"""작업 수행을 위한 새 데이터베이스 세션 가져오기."""
return self.SessionLocal()
def health_check(self) -> bool:
"""
데이터베이스 연결 상태 확인.
반환값:
bool: 데이터베이스 상태가 양호하면 True, 그렇지 않으면 False
"""
try:
with self.engine.connect() as conn:
conn.execute(text("SELECT 1"))
logger.info("✓ 데이터베이스 상태 점검 통과")
return True
except Exception as e:
logger.error(f"❌ 데이터베이스 상태 점검 실패: {e}")
return FalseDatabaseManager는 SQLAlchemy의 연결 풀링을 사용하여 연결을 설정합니다. pool_size=5 설정은 효율성을 위해 다섯 개의 지속적 연결을 유지합니다. pool_pre_ping 옵션은 사용 전 연결을 검증합니다. 이는 오래된 연결 오류를 방지합니다. 재시도 메커니즘은 지수적 백오프를 적용하여 실패한 작업을 최대 세 번까지 시도합니다. 이는 일시적인 네트워크 문제를 처리합니다.
3단계: LangChain 에이전트 핵심 구축
데이터베이스와 상호작용하는 맞춤형 도구를 사용하여 LangChain으로 AI 에이전트를 생성합니다. 이 에이전트는 함수 호출을 통해 정보를 저장하고 대화 기록을 검색합니다.
class DataPersistentAgent:
"""
데이터베이스 지속성 기능을 갖춘 AI 에이전트.
이 에이전트는:
- 세션 간 대화를 기억합니다
- 사용자 정보를 저장하고 검색합니다
- 중요한 엔티티를 추출하여 저장합니다
- 기록을 기반으로 개인화된 응답을 제공합니다
"""
def __init__(
self,
db_manager: DatabaseManager,
model_name: str = "gpt-4-turbo-preview",
temperature: float = 0.7
):
"""
데이터 지속성 에이전트를 초기화합니다.
인수:
db_manager: 데이터베이스 관리자 인스턴스
model_name: 사용할 LLM 모델 (기본값: gpt-4-turbo-preview)
temperature: 응답 생성을 위한 모델 온도
"""
self.db_manager = db_manager
self.model_name = model_name
# LLM 초기화
self.llm = ChatOpenAI(
model=model_name,
temperature=temperature,
openai_api_key=os.getenv("OPENAI_API_KEY")
)
# 에이전트 도구 생성
self.tools = self._create_agent_tools()
# 에이전트 프롬프트 생성
self.prompt = self._create_agent_prompt()
# 메모리 초기화
self.memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
# 에이전트 생성
self.agent = create_openai_functions_agent(
llm=self.llm,
tools=self.tools,
prompt=self.prompt
)
# 에이전트 실행기 생성
self.agent_executor = AgentExecutor(
agent=self.agent,
tools=self.tools,
memory=self.memory,
verbose=True,
handle_parsing_errors=True,
max_iterations=5
)
logger.info(f"✓ {model_name}으로 데이터 지속성 에이전트 초기화 완료")
def _create_agent_tools(self) -> List[Tool]:
"""데이터베이스 작업을 위한 커스텀 도구 생성."""
def save_user_info(user_data: str) -> str:
"""사용자 정보를 데이터베이스에 저장."""
try:
data = json.loads(user_data)
session = self.db_manager.get_session()
user = session.query(User).filter_by(user_id=data['user_id']).first()
if not user:
user = User(**data)
session.add(user)
else:
for key, value in data.items():
setattr(user, key, value)
session.commit()
session.close()
return f"✓ 사용자 정보 저장 성공"
except Exception as e:
logger.error(f"사용자 정보 저장 실패: {e}")
return f"❌ 사용자 정보 저장 오류: {str(e)}"
def retrieve_user_history(user_id: str) -> str:
"""사용자의 대화 기록을 가져옵니다."""
try:
session = self.db_manager.get_session()
user = session.query(User).filter_by(user_id=user_id).first()
if not user:
return "사용자 없음"
conversations = session.query(Conversation).filter_by(user_id=user.id).order_by(Conversation.created_at.desc()).limit(5).all()
history = []
for conv in conversations:
messages = session.query(Message).filter_by(conversation_id=conv.id).all()
history.append({
'conversation_id': conv.conversation_id,
'created_at': conv.created_at.isoformat(),
'message_count': len(messages),
'summary': conv.summary
})
session.close()
return json.dumps(history, indent=2)
except Exception as e:
logger.error(f"기록 조회 실패: {e}")
return f"❌ 기록 조회 오류: {str(e)}"
def extract_entities(text: str) -> str:
"""텍스트에서 엔티티를 추출하여 데이터베이스에 저장합니다."""
try:
entities = []
# 간단한 키워드 추출 (적절한 NER로 대체)
keywords = ['important', 'key', 'critical']
for keyword in keywords:
if keyword in text.lower():
entities.append({
'entity_type': 'keyword',
'entity_value': keyword,
'confidence': 0.8
})
return json.dumps(entities, indent=2)
except Exception as e:
logger.error(f"엔티티 추출 실패: {e}")
return f"❌ 엔티티 추출 오류: {str(e)}"
tools = [
Tool(
name="SaveUserInfo",
func=save_user_info,
description="사용자 정보를 데이터베이스에 저장합니다. 입력은 사용자 정보가 포함된 JSON 문자열이어야 합니다."
),
Tool(
name="RetrieveUserHistory",
func=retrieve_user_history,
description="데이터베이스에서 사용자의 대화 기록을 검색합니다. 입력은 user_id이어야 합니다."
),
Tool(
name="ExtractEntities",
func=extract_entities,
description="텍스트에서 중요한 엔티티를 추출하여 데이터베이스에 저장합니다. 입력값은 분석할 텍스트여야 합니다."
)
]
return tools
def _create_agent_prompt(self) -> ChatPromptTemplate:
"""에이전트 프롬프트 템플릿 생성"""
system_message = """당신은 대화를 기억하고 학습할 수 있는 유용한 AI 어시스턴트입니다.
다음 도구들을 사용할 수 있습니다:
- SaveUserInfo: 향후 대화를 위해 기억할 사용자 정보 저장
- RetrieveUserHistory: 사용자와의 과거 대화 조회
- ExtractEntities: 대화에서 중요한 정보 추출 및 저장
이 도구들을 활용하여 개인화되고 맥락을 고려한 응답을 제공하세요. 응답하기 전에 항상 사용자와의 이전 대화 기록이 있는지 확인하세요.
향후 대화를 위해 중요한 정보를 적극적으로 저장하세요."""
prompt = ChatPromptTemplate.from_messages([
("system", system_message),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
return prompt
def chat(self, user_id: str, message: str, conversation_id: Optional[str] = None) -> Dict[str, Any]:
"""
채팅 메시지를 처리하고 데이터베이스에 저장합니다.
이 메서드는 다음을 처리합니다:
1. 대화 생성 또는 검색
2. 사용자 메시지를 데이터베이스에 저장
3. 에이전트 응답 생성
4. 에이전트 응답을 데이터베이스에 저장
5. 모니터링을 위한 작업 기록
인수:
user_id: 사용자의 고유 식별자
message: 사용자의 메시지 텍스트
conversation_id: 기존 대화를 계속하기 위한 선택적 대화 ID
반환값:
dict: conversation_id, response, execution_time 포함
"""
start_time = datetime.utcnow()
try:
# 대화 가져오기 또는 생성
session = self.db_manager.get_session()
if conversation_id:
conversation = session.query(Conversation).filter_by(conversation_id=conversation_id).first()
else:
# 새 대화 생성
user = session.query(User).filter_by(user_id=user_id).first()
if not user:
user = User(user_id=user_id, name=user_id)
session.add(user)
session.commit()
conversation = Conversation(
conversation_id=f"conv_{user_id}_{datetime.utcnow().timestamp()}",
user_id=user.id,
title=message[:100]
)
session.add(conversation)
session.commit()
# 사용자 메시지 저장
user_message = Message(
conversation_id=conversation.id,
role="user",
content=message,
model=self.model_name
)
session.add(user_message)
session.commit()
# 에이전트 응답 가져오기
response = self.agent_executor.invoke({
"input": f"[User ID: {user_id}] {message}"
})
# 어시스턴트 메시지 저장
assistant_message = Message(
conversation_id=conversation.id,
role="assistant",
content=response['output'],
model=self.model_name
)
session.add(assistant_message)
session.commit()
# 작업 기록
execution_time = (datetime.utcnow() - start_time).total_seconds()
log_entry = AgentLog(
conversation_id=conversation.conversation_id,
level="INFO",
operation="chat",
message="채팅 처리 성공",
execution_time=execution_time
)
session.add(log_entry)
session.commit()
# 세션 종료 전 conversation_id 추출
conversation_id_result = conversation.conversation_id
session.close()
logger.info(f"✓ 사용자 {user_id} 채팅 처리 완료, 소요 시간 {execution_time:.2f}초")
return {
'conversation_id': conversation_id_result,
'response': response['output'],
'execution_time': execution_time
}
except Exception as e:
logger.error(f"❌ 채팅 처리 중 오류: {e}")
# 오류 기록
session = self.db_manager.get_session()
error_log = AgentLog(
conversation_id=conversation_id or "unknown",
level="ERROR",
operation="chat",
message=str(e),
error_details={'exception_type': type(e).__name__}
)
session.add(error_log)
session.commit()
session.close()
raiseDataPersistentAgent는 LangChain의 함수 호출 에이전트를 데이터베이스 도구로 감쌉니다. SaveUserInfo 도구는 User 레코드를 생성하거나 업데이트하여 사용자 데이터를 영구 저장합니다. RetrieveHistory 도구는 과거 대화를 쿼리하여 컨텍스트를 제공합니다. 시스템 프롬프트는 에이전트에게 정보 저장 및 기록 확인을 능동적으로 수행하도록 지시합니다. ConversationBufferMemory는 세션 내에서 단기 컨텍스트를 유지합니다. 데이터베이스 저장소는 세션 간 장기적인 영구 저장을 제공합니다.
3.5단계: 데이터 수집 모듈 생성
대화에서 데이터를 추출하고 구조화하는 도구를 구축합니다. 수집기는 요약 생성, 선호도 추출, LLM을 활용한 엔티티 식별을 수행합니다.
class DataCollector:
"""
에이전트 대화에서 데이터를 수집하고 구조화합니다.
이 모듈은:
- 대화 요약 생성
- 대화 기록에서 사용자 선호도 추출
- 명명된 엔티티 식별 및 저장
"""
def __init__(self, db_manager: DatabaseManager, llm: ChatOpenAI):
"""
데이터 수집기 초기화.
인수:
db_manager: 데이터베이스 관리자 인스턴스
llm: 텍스트 분석용 언어 모델
"""
self.db_manager = db_manager
self.llm = llm
logger.info("✓ 데이터 수집기 초기화 완료")
def extract_conversation_summary(self, conversation_id: str) -> str:
"""
LLM을 사용하여 대화 요약 생성 및 저장.
인수:
conversation_id: 요약할 대화의 ID
반환값:
str: 생성된 요약 텍스트
"""
try:
session = self.db_manager.get_session()
conversation = session.query(Conversation).filter_by(conversation_id=conversation_id).first()
if not conversation:
return "대화 찾지 못함"
messages = session.query(Message).filter_by(conversation_id=conversation.id).all()
# 대화 텍스트 생성
conv_text = "n".join([
f"{msg.role}: {msg.content}" for msg in messages
])
# LLM을 이용한 요약 생성
summary_prompt = f"""다음 대화를 2~3문장으로 요약하여 주요 주제와 결과를 담으세요:
{conv_text}
요약:"""
summary_response = self.llm.invoke([HumanMessage(content=summary_prompt)])
summary = summary_response.content
# 요약으로 대화 업데이트
conversation.summary = summary
session.commit()
session.close()
logger.info(f"✓ 대화 {conversation_id} 요약 생성 완료")
return summary
except Exception as e:
logger.error(f"요약 생성 실패: {e}")
return ""
def extract_user_preferences(self, user_id: str) -> Dict[str, Any]:
"""
대화 기록에서 사용자 선호도를 추출하고 저장합니다.
매개변수:
user_id: 분석할 사용자의 ID
반환값:
dict: 추출된 선호도
"""
try:
session = self.db_manager.get_session()
user = session.query(User).filter_by(user_id=user_id).first()
if not user:
return {}
# 최근 대화 가져오기
conversations = session.query(Conversation).filter_by(user_id=user.id).order_by(Conversation.created_at.desc()).limit(10).all()
all_messages = []
for conv in conversations:
messages = session.query(Message).filter_by(conversation_id=conv.id).all()
all_messages.extend([msg.content for msg in messages if msg.role == "user"])
if not all_messages:
return {}
# LLM을 활용한 선호도 분석
분석 프롬프트 = f"""사용자의 다음 메시지를 분석하여 선호도, 관심사 및 커뮤니케이션 스타일을 추출하세요.
메시지:
{chr(10).join(all_messages[:20])}
다음 구조의 JSON 객체를 반환하세요:
{{
"interests": ["interest1", "interest2"],
"communication_style": "description",
"preferred_topics": ["topic1", "topic2"],
"language_preference": "language"
}}"""
response = self.llm.invoke([HumanMessage(content=analysis_prompt)])
try:
# 응답에서 JSON 추출
content = response.content
if '```json' in content:
content = content.split('```json')[1].split('```')[0].strip()
elif '```' in content:
content = content.split('```')[1].split('```')[0].strip()
preferences = json.loads(content)
# 사용자 선호도 업데이트
user.preferences = preferences
session.commit()
logger.info(f"✓ 사용자 {user_id}의 선호도 추출 완료")
return preferences
except json.JSONDecodeError:
logger.warning("선호도 JSON 파싱 실패")
return {}
finally:
session.close()
except Exception as e:
logger.error(f"선호도 추출 실패: {e}")
return {}
def extract_entities_with_llm(self, conversation_id: str) -> List[Dict[str, Any]]:
"""
LLM을 사용하여 명명된 엔티티를 추출합니다.
인수:
conversation_id: 분석할 대화의 ID
반환값:
list: 추출된 엔티티의 리스트
"""
try:
session = self.db_manager.get_session()
conversation = session.query(Conversation).filter_by(conversation_id=conversation_id).first()
if not conversation:
return []
messages = session.query(Message).filter_by(conversation_id=conversation.id).all()
conv_text = "n".join([msg.content for msg in messages])
# LLM을 사용하여 엔티티 추출
entity_prompt = f"""다음 대화에서 명명된 엔티티를 추출하세요. 식별 대상:
- 인물(PERSON) (PERSON)
- 조직 (ORG)
- 장소 (LOC)
- 날짜 (DATE)
- 제품 (PRODUCT)
- 기술 (TECH)
대화 내용:
{conv_text}
다음 형식의 엔티티 JSON 배열을 반환하세요:
[
{{"type": "PERSON", "value": "John Doe", "context": "팀 리더로 언급됨"}},
{{"type": "ORG", "value": "Acme Corp", "context": "고객사"}}
]"""
response = self.llm.invoke([HumanMessage(content=entity_prompt)])
try:
content = response.content
if '```json' in content:
content = content.split('```json')[1].split('```')[0].strip()
elif '```' in content:
content = content.split('```')[1].split('```')[0].strip()
entities_data = json.loads(content)
# 엔티티를 데이터베이스에 저장
saved_entities = []
for entity_data in entities_data:
entity = Entity(
conversation_id=conversation.id,
entity_type=entity_data['type'],
entity_value=entity_data['value'],
context=entity_data.get('context', ''),
confidence=0.9 # LLM 추출은 높은 신뢰도를 가짐
)
session.add(entity)
saved_entities.append(entity_data)
session.commit()
session.close()
logger.info(f"✓ 대화 {conversation_id}에서 {len(saved_entities)}개의 엔티티 추출 완료")
return saved_entities
except json.JSONDecodeError:
logger.warning("엔티티 JSON 파싱 실패")
return []
except Exception as e:
logger.error(f"엔티티 추출 실패: {e}")
return []데이터 수집기는 LLM을 활용해 대화를 분석합니다. ` extract_conversation_summary ` 메서드는 대화의 간결한 요약을 생성합니다. ` extract_user_preferences ` 메서드는 메시지 패턴을 분석해 사용자 관심사와 소통 스타일을 식별합니다. ` extract_entities_with_llm ` 메서드는 구조화된 프롬프트를 사용해 사람, 조직, 기술 등 명명된 개체를 추출합니다. 추출된 모든 데이터는 향후 참조를 위해 데이터베이스에 저장됩니다.
4단계: 스마트 데이터 처리 파이프라인 구축
에이전트 작업을 차단하지 않고 데이터 수집을 처리하기 위해 백그라운드 처리를 구현합니다. 이 파이프라인은 요약 및 명명된 개체 처리를 위해 작업자 스레드와 큐를 사용합니다.
class DataProcessingPipeline:
"""
비동기 데이터 처리 파이프라인.
이 파이프라인은:
- 백그라운드에서 대화 처리
- 요약 생성
- 메인 흐름 차단 없이 엔티티 추출
- 사용자 선호도 주기적 업데이트
"""
def __init__(self, db_manager: DatabaseManager, collector: DataCollector, batch_size: int = 10):
"""
처리 파이프라인 초기화.
인수:
db_manager: 데이터베이스 관리자 인스턴스
collector: 처리 작업을 위한 데이터 수집기
batch_size: 각 배치에서 처리할 항목 수
"""
self.db_manager = db_manager
self.collector = collector
self.batch_size = batch_size
# 처리 큐
self.summary_queue = Queue()
self.entity_queue = Queue()
self.preference_queue = Queue()
# 작업자 스레드
self.workers = []
self.running = False
logger.info("✓ 데이터 처리 파이프라인 초기화 완료")
def start(self):
"""백그라운드 처리 작업자 시작."""
self.running = True
# 작업자 스레드 생성
summary_worker = Thread(target=self._process_summaries, daemon=True)
entity_worker = Thread(target=self._process_entities, daemon=True)
preference_worker = Thread(target=self._process_preferences, daemon=True)
summary_worker.start()
entity_worker.start()
preference_worker.start()
self.workers = [summary_worker, entity_worker, preference_worker]
logger.info("✓ 백그라운드 처리 작업자 3개 시작됨")
def stop(self):
"""백그라운드 처리 작업자 중지."""
self.running = False
for worker in self.workers:
worker.join(timeout=5)
logger.info("✓ 백그라운드 처리 작업자 중지됨")
def queue_conversation_for_processing(self, conversation_id: str, user_id: str):
"""
대화 처리 대기열에 추가합니다.
매개변수:
conversation_id: 처리할 대화 ID
user_id: 선호도 추출을 위한 사용자 ID
"""
self.summary_queue.put(conversation_id)
self.entity_queue.put(conversation_id)
self.preference_queue.put(user_id)
logger.info(f"✓ 처리 대기열에 대화 {conversation_id} 추가됨")
def _process_summaries(self):
"""대화 요약 처리 작업자."""
while self.running:
try:
if not self.summary_queue.empty():
conversation_id = self.summary_queue.get()
self.collector.extract_conversation_summary(conversation_id)
self.summary_queue.task_done()
else:
time.sleep(1)
except Exception as e:
logger.error(f"요약 작업자 오류: {e}")
def _process_entities(self):
"""엔티티 추출 처리 작업자."""
while self.running:
try:
if not self.entity_queue.empty():
conversation_id = self.entity_queue.get()
self.collector.extract_entities_with_llm(conversation_id)
self.entity_queue.task_done()
else:
time.sleep(1)
except Exception as e:
logger.error(f"엔티티 작업자 오류: {e}")
def _process_preferences(self):
"""사용자 선호도 처리 작업자."""
while self.running:
try:
if not self.preference_queue.empty():
user_id = self.preference_queue.get()
self.collector.extract_user_preferences(user_id)
self.preference_queue.task_done()
else:
time.sleep(1)
except Exception as e:
logger.error(f"선호도 작업자 오류: {e}")
def get_queue_status(self) -> Dict[str, int]:
"""
현재 큐 크기를 가져옵니다.
반환값:
dict: 각 처리 유형별 큐 크기
"""
return {
'summary_queue': self.summary_queue.qsize(),
'entity_queue': self.entity_queue.qsize(),
'preference_queue': self.preference_queue.qsize()
}ProcessingPipeline은 데이터 수집과 메시지 처리를 분리합니다. 대화 완료 시 즉시 처리되지 않고 큐에 추가됩니다. 별도의 작업자 스레드가 이 큐에서 항목을 가져와 백그라운드에서 처리합니다. 이는 데이터 수집이 에이전트 응답을 차단하는 것을 방지합니다. daemon=True 설정은 메인 프로그램 종료 시 작업자가 종료되도록 보장합니다. 큐 상태 모니터링은 처리 백로그 추적에 도움이 됩니다.
5단계: 실시간 모니터링 및 로깅 추가
에이전트 성능 추적, 오류 감지 및 보고서 생성을 위한 모니터링 시스템을 구축합니다. 모니터는 로그를 분석하여 운영 인사이트를 제공합니다.
class AgentMonitor:
"""
실시간 모니터링 및 메트릭 수집.
이 모듈은:
- 성능 메트릭 추적
- 시스템 상태 모니터링
- 분석 보고서 생성
"""
def __init__(self, db_manager: DatabaseManager):
"""
에이전트 모니터 초기화.
인수:
db_manager: 데이터베이스 관리자 인스턴스
"""
self.db_manager = db_manager
logger.info("✓ 에이전트 모니터 초기화 완료")
def get_performance_metrics(self, hours: int = 24) -> Dict[str, Any]:
"""
지정된 시간 범위의 성능 메트릭을 가져옵니다.
인수:
hours: 조회할 시간(시간 단위)
반환값:
dict: 작업 횟수 및 오류율 등 성능 메트릭
"""
try:
session = self.db_manager.get_session()
cutoff_time = datetime.utcnow() - timedelta(hours=hours)
# 로그 쿼리
logs = session.query(AgentLog).filter(
AgentLog.created_at >= cutoff_time
).all()
# 메트릭 계산
total_operations = len(logs)
error_count = len([log for log in logs if log.level == "ERROR"])
avg_execution_time = sum([log.execution_time or 0 for log in logs]) / max(total_operations, 1)
# 대화 수 조회
conversations = session.query(Conversation).filter(
Conversation.created_at >= cutoff_time
).count()
messages = session.query(Message).join(Conversation).filter(
Message.created_at >= cutoff_time
).count()
session.close()
metrics = {
'시간대(시간)': hours,
'총 작업 수': total_operations,
'오류 횟수': error_count,
'오류율': error_count / max(total_operations, 1),
'avg_execution_time': avg_execution_time,
'conversations_created': conversations,
'messages_processed': messages
}
logger.info(f"✓ 지난 {hours}시간 동안의 성능 메트릭 생성 완료")
return metrics
except Exception as e:
logger.error(f"성능 메트릭스 획득 실패: {e}")
return {}
def health_check(self) -> Dict[str, Any]:
"""
헬스 체크 수행.
반환값:
dict: 데이터베이스 연결 상태 및 오류율을 포함한 헬스 상태
"""
try:
# 데이터베이스 연결 상태 확인
db_healthy = self.db_manager.health_check()
# 최근 오류율 확인
metrics = self.get_performance_metrics(hours=1)
recent_errors = metrics.get('error_count', 0)
# 전체 상태 판단
is_healthy = db_healthy and recent_errors < 10
health_status = {
'status': 'healthy' if is_healthy else 'degraded',
'database_connected': db_healthy,
'recent_errors': recent_errors,
'timestamp': datetime.utcnow().isoformat()
}
logger.info(f"✓ Health check: {health_status['status']}")
return health_status
except Exception as e:
logger.error(f"건강 상태 확인 실패: {e}")
return {
'status': 'unhealthy',
'error': str(e),
'timestamp': datetime.utcnow().isoformat()
}AgentMonitor는 시스템 운영에 대한 가시성을 제공합니다. AgentLog 테이블을 쿼리하여 총 작업 수, 오류율, 평균 실행 시간 등의 메트릭을 추적합니다. get_metrics 메서드는 구성 가능한 시간 창에 대한 통계를 계산합니다. get_error_report 메서드는 디버깅을 위한 상세한 오류 정보를 가져옵니다. 이 모니터링은 사전적 문제 감지를 가능하게 합니다. 높은 오류율은 사용자에게 영향이 미치기 전에 조사를 촉발합니다.
6단계: 쿼리 인터페이스 구축
저장된 데이터를 검색하고 분석할 수 있는 쿼리 기능을 생성합니다. 이 인터페이스는 대화 검색, 엔터티 추적, 분석 생성 메서드를 제공합니다.
class DataQueryInterface:
"""
저장된 에이전트 데이터 쿼리용 인터페이스.
이 모듈은 다음 메서드를 제공합니다:
- 사용자 분석 쿼리
- 대화 기록 검색
- 특정 정보 검색
"""
def __init__(self, db_manager: DatabaseManager):
"""
쿼리 인터페이스 초기화.
인수:
db_manager: 데이터베이스 관리자 인스턴스
"""
self.db_manager = db_manager
logger.info("✓ 쿼리 인터페이스 초기화 완료")
def get_user_analytics(self, user_id: str) -> Dict[str, Any]:
"""
특정 사용자에 대한 분석 데이터를 가져옵니다.
인수:
user_id: 분석할 사용자의 ID
반환값:
dict: 대화 수 및 선호도 등을 포함한 사용자 분석 데이터
"""
try:
session = self.db_manager.get_session()
user = session.query(User).filter_by(user_id=user_id).first()
if not user:
return {}
# 대화 수 조회
conversation_count = session.query(Conversation).filter_by(user_id=user.id).count()
# 메시지 수 조회
message_count = session.query(Message).join(Conversation).filter(
Conversation.user_id == user.id
).count()
# 엔티티 개수 가져오기
entity_count = session.query(Entity).join(Conversation).filter(
Conversation.user_id == user.id
).count()
# 시간 범위 가져오기
first_conversation = session.query(Conversation).filter_by(
user_id=user.id
).order_by(Conversation.created_at).first()
last_conversation = session.query(Conversation).filter_by(
user_id=user.id
).order_by(Conversation.created_at.desc()).first()
session.close()
분석 = {
'user_id': user_id,
'name': user.name,
'conversation_count': conversation_count,
'message_count': message_count,
'entity_count': entity_count,
'preferences': user.preferences,
'first_interaction': first_conversation.created_at.isoformat() if first_conversation else None,
'last_interaction': last_conversation.created_at.isoformat() if last_conversation else None,
'avg_messages_per_conversation': message_count / max(conversation_count, 1)
}
logger.info(f"✓ 사용자 {user_id}에 대한 분석 데이터 생성 완료")
return analytics
except Exception as e:
logger.error(f"사용자 분석 데이터 가져오기 실패: {e}")
return {}QueryInterface는 저장된 데이터에 접근하는 메서드를 제공합니다. get_user_conversations 메서드는 선택적으로 메시지를 포함하여 전체 대화 기록을 가져옵니다. search_conversations 메서드는 SQL의 ILIKE 연산자를 사용하여 메시지 콘텐츠 전체에 대한 풀텍스트 검색을 수행합니다. get_entity_mentions 메서드는 특정 엔티티가 언급된 모든 대화를 찾습니다. get_user_analytics 메서드는 사용자 활동에 대한 통계를 생성합니다. 이러한 쿼리를 통해 대시보드 구축, 보고서 생성 및 개인화된 경험 창출이 가능합니다.
7단계: Bright Data의 실시간 웹 데이터로 RAG 구축
Bright Data의 실시간 웹 인텔리전스로 데이터베이스 연결 에이전트에 RAG 기능을 강화하세요. 이 통합은 대화 기록과 최신 웹 데이터를 결합하여 더 나은 응답을 제공합니다.
class BrightDataRAGEnhancer:
"""
Bright Data 웹 인텔리전스로 데이터 지속형 에이전트 강화.
이 모듈은:
- Bright Data에서 실시간 웹 데이터 가져오기
- RAG용 벡터 저장소에 웹 데이터 통합
- 웹 강화 지식으로 에이전트 성능 향상
"""
def __init__(self, api_key: str, db_manager: DatabaseManager):
"""
Bright Data로 RAG 강화기 초기화.
인수:
api_key: Bright Data API 키
db_manager: 데이터베이스 관리자 인스턴스
"""
self.api_key = api_key
self.db_manager = db_manager
self.base_url = "https://api.brightdata.com"
# RAG용 벡터 저장소 초기화
self.embeddings = OpenAIEmbeddings()
self.vector_store = Chroma(
embedding_function=self.embeddings,
persist_directory="./chroma_db"
)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
logger.info("✓ Bright Data RAG enhancer 초기화 완료")
def fetch_dataset_data(
self,
dataset_id: str,
filters: Optional[Dict[str, Any]] = None,
limit: int = 1000
) -> List[Dict[str, Any]]:
"""
Bright Data Dataset Marketplace에서 데이터를 가져옵니다.
매개변수:
dataset_id: 가져올 데이터셋의 ID
filters: 데이터에 대한 선택적 필터
limit: 가져올 최대 레코드 수
반환값:
list: 가져온 데이터셋 레코드
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
endpoint = f"{self.base_url}/datasets/v3/snapshot/{dataset_id}"
params = {
"format": "json",
"limit": limit
}
if filters:
params["filter"] = json.dumps(filters)
try:
response = requests.get(endpoint, headers=headers, params=params)
response.raise_for_status()
data = response.json()
logger.info(f"✓ Bright Data 데이터셋 {dataset_id}에서 {len(data)}개의 레코드 가져옴")
return data
except Exception as e:
logger.error(f"Bright Data 데이터셋 가져오기 실패: {e}")
return []
def ingest_web_data_to_rag(
self,
dataset_records: List[Dict[str, Any]],
text_fields: List[str],
metadata_fields: Optional[List[str]] = None
) -> int:
"""
웹 데이터를 RAG 벡터 저장소에 입력합니다.
매개변수:
dataset_records: Bright Data의 레코드
text_fields: 텍스트 콘텐츠로 사용할 필드
metadata_fields: 메타데이터에 포함할 필드
반환값:
int: 수집된 문서 청크 수
"""
try:
documents = []
for record in dataset_records:
# 텍스트 필드 결합
text_content = " ".join([
str(record.get(field, ""))
for field in text_fields
if record.get(field)
])
if not text_content.strip():
continue
# 메타데이터 생성
metadata = {
"source": "bright_data",
"record_id": record.get("id", "unknown"),
"timestamp": datetime.utcnow().isoformat()
}
if metadata_fields:
for field in metadata_fields:
if field in record:
metadata[field] = record[field]
# 텍스트를 청크로 분할
chunks = self.text_splitter.split_text(text_content)
for chunk in chunks:
documents.append({
"content": chunk,
"metadata": metadata
})
# 벡터 저장소에 추가
if documents:
texts = [doc["content"] for doc in documents]
metadatas = [doc["metadata"] for doc in documents]
self.vector_store.add_texts(
texts=texts,
metadatas=metadatas
)
logger.info(f"✓ {len(documents)}개의 문서 조각을 RAG에 수집했습니다")
return len(documents)
except Exception as e:
logger.error(f"RAG에 웹 데이터 가져오기 실패: {e}")
return 0
def create_rag_enhanced_agent(
self,
base_agent: DataPersistentAgent
) -> DataPersistentAgent:
"""
기존 에이전트를 RAG 기능으로 강화합니다.
인수:
base_agent: 강화할 기본 에이전트
반환값:
DataPersistentAgent: RAG 도구가 추가된 강화된 에이전트
"""
def rag_search(query: str) -> str:
"""대화 기록과 웹 데이터를 모두 검색합니다."""
try:
# 대화 기록에서 검색
session = self.db_manager.get_session()
messages = session.query(Message).filter(
Message.content.ilike(f'%{query}%')
).order_by(Message.created_at.desc()).limit(5).all()
results = []
for msg in messages:
results.append({
'content': msg.content,
'source': 'conversation_history',
'relevance': 0.8
})
session.close()
# 벡터 저장소(웹 데이터)에서 검색
try:
vector_results = self.vector_store.similarity_search_with_score(query, k=5)
for doc, score in vector_results:
results.append({
'content': doc.page_content,
'source': 'web_data',
'relevance': 1 - score
})
except Exception as e:
logger.error(f"벡터 저장소에서 검색 실패: {e}")
if not results:
return "관련 정보가 없습니다."
# 컨텍스트 포맷팅
context_text = "nn".join([
f"[{item['source']}] {item['content'][:200]}..."
for item in results[:5]
])
return f"검색된 컨텍스트:n{context_text}"
except Exception as e:
logger.error(f"RAG 검색 실패: {e}")
return f"검색 수행 중 오류: {str(e)}"
# 에이전트에 RAG 도구 추가
rag_tool = Tool(
name="SearchKnowledgeBase",
func=rag_search,
description="관련 정보를 찾기 위해 대화 기록과 실시간 웹 데이터를 모두 검색합니다. 입력은 검색 쿼리여야 합니다."
)
base_agent.tools.append(rag_tool)
# 새로운 도구로 에이전트 재생성
base_agent.agent = create_openai_functions_agent(
llm=base_agent.llm,
tools=base_agent.tools,
prompt=base_agent.prompt
)
base_agent.agent_executor = AgentExecutor(
agent=base_agent.agent,
tools=base_agent.tools,
memory=base_agent.memory,
verbose=True,
handle_parsing_errors=True,
max_iterations=5
)
logger.info("✓ RAG 기능을 갖춘 향상된 에이전트")
return base_agentBrightDataEnhancer는 실시간 웹 데이터를 에이전트에 통합합니다. fetch_dataset 메서드는 Bright Data 마켓플레이스에서 구조화된 데이터를 가져옵니다. ingest_to_rag 메서드는 이 데이터를 처리하고 분할합니다. 의미적 검색을 위해 Chroma 벡터 데이터베이스에 저장합니다. retrieve_context 메서드는 하이브리드 검색을 수행합니다. 데이터베이스 기록과 벡터 유사도 검색을 결합합니다. create_rag_tool 메서드는 이 기능을 에이전트가 사용하는 LangChain 도구로 패키징합니다. enhance_agent 메서드는 기존 에이전트에 이 RAG 기능을 추가합니다. 이를 통해 에이전트는 내부 대화 기록과 최신 외부 데이터를 모두 활용하여 질문에 답변할 수 있습니다.
데이터 지속형 에이전트 시스템 전체 실행하기
모든 구성 요소를 결합하여 작동하는 시스템을 생성합니다.
def main():
"""모든 구성 요소가 함께 작동하는 메인 실행 흐름."""
print("=" * 60)
print("데이터 지속형 AI 에이전트 시스템 - 초기화")
print("=" * 60)
# 1단계: 데이터베이스 초기화
print("n[Step 1] 데이터베이스 연결 설정 중...")
db_manager = DatabaseManager(
database_url=os.getenv("DATABASE_URL"),
pool_size=5,
max_retries=3
)
db_manager.initialize_database()
# 2단계: 핵심 에이전트 초기화
print("n[Step 2] AI 에이전트 핵심 구축 중...")
agent = DataPersistentAgent(
db_manager=db_manager,
model_name=os.getenv("AGENT_MODEL", "gpt-4-turbo-preview")
)
# Step 3: 데이터 수집기 초기화
print("n[Step 3] 데이터 수집 모듈 생성 중...")
collector = DataCollector(db_manager, agent.llm)
# Step 4: 처리 파이프라인 초기화
print("n[Step 4] 데이터 처리 파이프라인 구현 중...")
pipeline = DataProcessingPipeline(db_manager, collector)
pipeline.start()
# 5단계: 모니터링 초기화
print("n[Step 5] 모니터링 및 로깅 추가 중...")
monitor = AgentMonitor(db_manager)
# 6단계: 쿼리 인터페이스 초기화
print("n[Step 6] 쿼리 인터페이스 구축 중...")
query_interface = DataQueryInterface(db_manager)
# 7단계: 선택적 Bright Data RAG 강화
print("n[Step 7] RAG 강화 (선택 사항)...")
bright_data_key = os.getenv("BRIGHT_DATA_API_KEY")
if bright_data_key and bright_data_key != "your-bright-data-api-key":
print("Bright Data에서 실시간 웹 데이터 가져오기 중...")
enhancer = BrightDataRAGEnhancer(bright_data_key, db_manager)
# 예시: 웹 데이터 가져오기 및 통합
web_data = enhancer.fetch_dataset_data(
dataset_id="example_dataset_id",
limit=100
)
if web_data:
enhancer.ingest_web_data_to_rag(
dataset_records=web_data,
text_fields=["title", "content", "description"],
metadata_fields=["url", "published_date"]
)
# RAG로 에이전트 강화
agent = enhancer.create_rag_enhanced_agent(agent)
print("✓ Bright Data RAG 기능으로 에이전트 강화 완료")
else:
print("⚠️ Bright Data API 키 미검출 - 웹 데이터 통합 생략")
print("n" + "=" * 60)
print("데모 대화")
print("=" * 60)
# 데모 사용자 상호작용
test_user = "demo_user_001"
# 첫 번째 대화
print("n📝 대화 1:")
response1 = agent.chat(
user_id=test_user,
message="안녕하세요! 머신 러닝에 대해 배우고 싶어요."
)
print(f"에이전트: {response1['response']}n")
# 처리 대기열에 추가
pipeline.queue_conversation_for_processing(
response1['conversation_id'],
test_user
)
# 두 번째 대화
print("📝 대화 2:")
response2 = agent.chat(
user_id=test_user,
message="신경망에 대해 이해하도록 도와주세요?",
conversation_id=response1['conversation_id']
)
print(f"에이전트: {response2['response']}n")
# 백그라운드 처리 대기
print("⏳ 백그라운드에서 데이터 처리 중...")
time.sleep(5)
print("n" + "=" * 60)
print("분석 및 모니터링")
print("=" * 60)
# 성능 지표 가져오기
metrics = monitor.get_performance_metrics(hours=1)
print(f"n📊 성능 지표:")
print(f" - 총 작업 수: {metrics.get('total_operations', 0)}")
print(f" - 오류율: {metrics.get('error_rate', 0):.2%}")
print(f" - 평균 실행 시간: {metrics.get('avg_execution_time', 0):.2f}초")
print(f" - 생성된 대화: {metrics.get('conversations_created', 0)}개")
print(f" - 처리된 메시지: {metrics.get('messages_processed', 0)}")
# 사용자 분석 데이터 가져오기
analytics = query_interface.get_user_analytics(test_user)
print(f"n👤 사용자 분석:")
print(f" - 대화 수: {analytics.get('conversation_count', 0)}")
print(f" - 메시지 수: {analytics.get('message_count', 0)}")
print(f" - 엔티티 수: {analytics.get('entity_count', 0)}")
print(f" - 대화당 평균 메시지 수: {analytics.get('avg_messages_per_conversation', 0):.1f}")
# 상태 점검
health = monitor.health_check()
print(f"n🏥 시스템 상태: {health['status']}")
# 대기열 상태
queue_status = pipeline.get_queue_status()
print(f"n📋 처리 중인 대기열:")
print(f" - 요약 대기열: {queue_status['summary_queue']}")
print(f" - 엔티티 대기열: {queue_status['entity_queue']}")
print(f" - 선호도 큐: {queue_status['preference_queue']}")
# 파이프라인 중지
pipeline.stop()
print("n" + "=" * 60)
print("데이터 지속형 에이전트 시스템 - 완료")
print("=" * 60)
print("n✓ 모든 데이터 데이터베이스에 저장됨")
print("✓ 백그라운드 처리 완료됨")
print("✓ 시스템 프로덕션 사용 준비 완료됨")
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("nn⚠️ 정상 종료 중...")
except Exception as e:
logger.error(f"시스템 오류: {e}")
import traceback
traceback.print_exc()데이터베이스 연결 에이전트 시스템 실행:
python agent.py시스템은 전체 워크플로를 실행합니다. 데이터베이스를 초기화하고 모든 테이블을 생성합니다. 데이터베이스 도구로 LangChain 에이전트를 설정합니다. 처리를 위한 백그라운드 워커를 시작합니다. 데모 대화를 처리하고 데이터베이스에 저장합니다. 백그라운드에서 엔티티를 추출하고 요약문을 생성합니다. 실시간 분석 및 메트릭을 표시합니다.
각 구성 요소가 초기화되고 데이터를 처리할 때 상세한 로깅을 확인할 수 있습니다. 에이전트는 모든 메시지를 저장합니다. 인사이트를 추출합니다. 전체 대화 컨텍스트를 유지합니다.
실용적인 사용 사례
1. 전체 기록을 활용한 고객 지원
# 에이전트가 과거 상호작용을 검색
support_agent = DataPersistentAgent(db_manager)
response = support_agent.chat(
user_id="customer_123",
message="아직도 연결 문제가 발생합니다")
# 에이전트가 연결 문제에 대한 이전 대화 확인2. 학습 기능을 갖춘 개인 AI 어시스턴트
# 에이전트가 시간에 따라 선호도를 학습
query_interface = QueryInterface(db_manager)
analytics = query_interface.get_user_analytics("user_456")
# 상호작용 패턴, 선호도, 자주 다루는 주제 표시3. 지식 기반을 활용한 연구 보조
# 대화 기록과 웹 데이터를 결합
enhancer = BrightDataEnhancer(api_key, db_manager)
enhancer.ingest_to_rag(research_data, ["title", "abstract", "content"])
agent = enhancer.enhance_agent(agent)
# 에이전트가 과거 논의와 최신 연구를 모두 참조혜택 요약
| 기능 | 데이터베이스 미사용 | 데이터베이스 지속성 적용 시 |
|---|---|---|
| 메모리 | 재시작 시 손실 | 영구 저장 |
| 개인화 | 없음 | 전체 기록 기반 |
| 분석 | 불가능 | 완전한 상호작용 데이터 |
| 오류 복구 | 수동 개입 | 자동 재시도 및 로깅 |
| 확장성 | 단일 인스턴스 | 공유 상태를 가진 다중 인스턴스 |
| 인사이트 | 세션 종료 후 손실 | 추출 및 추적됨 |
마무리
이제 데이터베이스에 대화를 저장하는 프로덕션 환경에 바로 적용 가능한 AI 에이전트 시스템을 구축했습니다. 이 시스템은 모든 상호작용을 저장하고, 엔티티와 인사이트를 추출하며, 전체 대화 기록을 유지하며, 자동 오류 복구 기능을 포함한 모니터링을 제공합니다.
보안 액세스를 위한 사용자 인증 추가, 분석 시각화를 위한 대시보드 구축, 의미론적 검색을 위한 임베딩 구현, 통합을 위한 API 엔드포인트 생성, 확장성을 위한 Docker 배포 등을 통해 시스템을 강화하세요. 모듈식 설계로 특정 요구사항에 맞게 쉽게 맞춤화할 수 있습니다.
더 많은 기능을 위해 고급 AI 에이전트 패턴과 Bright Data의 웹 인텔리전스 플랫폼을 살펴보세요.
무료 계정을 생성하여 기억하고 학습하는 지능형 시스템 구축을 시작하세요.
