엔터프라이즈 AI 챗봇 구축 가이드 - RAG + Agent + Fine-Tuning 통합

def route_query(user_message: str, context: dict) -> str:
    """사용자 질의를 적절한 파이프라인으로 라우팅"""
    classification = classify_intent(user_message)
    
    if classification == "document_search":
        return rag_pipeline(user_message, context)
    
    elif classification == "system_action":
        return agent_pipeline(user_message, context)
    
    elif classification == "general_chat":
        return chat_pipeline(user_message, context)
    
    elif classification == "data_query":
        return text_to_sql_pipeline(user_message, context)
    
    else:
        return "죄송합니다. 질문을 이해하지 못했습니다. 다시 질문해주세요."

# 다양한 소스에서 문서 수집 및 인덱싱
sources = {
    "confluence": ConfluenceLoader(url="https://wiki.company.com"),
    "github": GitHubLoader(repos=["company/docs", "company/runbook"]),
    "notion": NotionLoader(token="..."),
    "slack_history": SlackLoader(channels=["#engineering", "#incidents"]),
}
 
all_docs = []
for source_name, loader in sources.items():
    docs = loader.load()
    for doc in docs:
        doc.metadata["source"] = source_name
        doc.metadata["access_level"] = get_access_level(doc)
    all_docs.extend(docs)
 
# 청킹 + 임베딩 + 벡터 DB 저장
chunks = text_splitter.split_documents(all_docs)
vectorstore = Chroma.from_documents(chunks, embeddings)

def secure_rag_search(query: str, user: dict) -> str:
    """사용자 권한에 따른 RAG 검색"""
    # 사용자 권한으로 필터 구성
    access_filter = {
        "access_level": {"$in": user["allowed_levels"]},
        "department": {"$in": user["departments"]}
    }
    
    # 필터링된 검색
    docs = vectorstore.similarity_search(
        query, k=5, filter=access_filter
    )
    
    # LLM으로 응답 생성
    context = format_docs_with_sources(docs)
    response = rag_chain.invoke({"context": context, "question": query})
    
    return response

from langchain_core.tools import tool
 
@tool
def search_jira(query: str) -> str:
    """Jira에서 이슈를 검색합니다."""
    issues = jira_client.search_issues(query, maxResults=5)
    return format_issues(issues)
 
@tool
def create_jira_ticket(title: str, description: str, priority: str) -> str:
    """Jira 티켓을 생성합니다. 사용자 확인 후 실행됩니다."""
    issue = jira_client.create_issue(
        project="ENG", summary=title, description=description, priority=priority
    )
    return f"티켓 생성 완료: {issue.key}"
 
@tool
def query_grafana(metric: str, time_range: str) -> str:
    """Grafana에서 메트릭을 조회합니다."""
    data = grafana_client.query(metric, time_range)
    return format_metrics(data)
 
@tool
def send_slack(channel: str, message: str) -> str:
    """Slack 채널에 메시지를 전송합니다."""
    slack_client.chat_postMessage(channel=channel, text=message)
    return f"메시지 전송 완료: {channel}"
 
# Agent 구성
agent = create_react_agent(
    llm=fine_tuned_llm,  # Fine-Tuned 모델 사용
    tools=[search_jira, create_jira_ticket, query_grafana, send_slack, search_docs],
    prompt="당신은 Data Dynamics의 AI 어시스턴트입니다..."
)

class ConversationManager:
    def __init__(self, max_history: int = 20):
        self.sessions = {}  # Redis 기반 세션 관리
        self.max_history = max_history
    
    def get_context(self, session_id: str) -> list:
        """대화 이력 조회 (요약 포함)"""
        history = self.sessions.get(session_id, [])
        
        if len(history) > self.max_history:
            # 오래된 대화 요약
            old = history[:-10]
            summary = llm.invoke(f"다음 대화를 3줄로 요약: {old}")
            history = [{"role": "system", "content": f"이전 대화 요약: {summary}"}] + history[-10:]
            self.sessions[session_id] = history
        
        return history
    
    def add_message(self, session_id: str, role: str, content: str):
        if session_id not in self.sessions:
            self.sessions[session_id] = []
        self.sessions[session_id].append({"role": role, "content": content})