지난번에 채팅화면을 화면에 띄우고 SSE로 통신하는거까지만 구현을 했었다.
설계부분도 바뀌었다. 처음에 AI Analytics를 썼었는데 이건 너무 구시대적인 방식이다...
왜냐고?
애초에 답을 정해놓는 구조기 때문이다.
처음에 질문이 들어오면 AI가 ㅗㅜㅑ... 이건 RAG네 이건 JIRA네 하고 딱 이분법적으로 나눴었는데 이문제는
복합적인 질문이 들어오면 멍청해지고 도구가 추가될때마다 작업을 해줘야 하므로 굉장히 비효율적이다.
그래서 알아서 AI가 연장을 챙기는 구조(Tool Calling) 방식으로 재설계 하였다.
즉 질문을 분석하는 지능과 도구를 고르는 지능을 모두 합쳐버린거다.
일단 내가 설계한 아키텍처다.
보라색 네모부분 (LLM#1 ~ LLM#2 부분까지) 모두 오케스트레이터가 처리한다.
위와같이 Tool Caliing 기반으로 설계했다.
Tool Calling과 RAG에 대한 내용은
아래 게시글 참조
https://ssmyefrin.tistory.com/125
[AI] RAG 와 VectorDB
✔️ RAG (Retrieval-Augmented Generation) 한국말로 번역하면 검색 증강 생성으로 LLM의 답변 능력을 외부의 신뢰할 수 있는 데이터로 보강하여 답변을 생성하는 기술 및 아키텍처 이다. 즉, LLM이 접근할
ssmyefrin.tistory.com
https://ssmyefrin.tistory.com/126
[AI] Tool Calling
챗봇을 개발하려 하다보니 툴 콜링 툴 콜링 자꾸 등장하는 단어가 있길래 아니... 뭔말이여.. 하다가 정리해보는 글 ✔️ Tool Calling AI 업계에서 LLM이 외부세계와 연결될때 쓰이는 가장 보편적인
ssmyefrin.tistory.com
그리고 코딩하기 전에 SSE 개념도 알면 좋을듯하다.
https://ssmyefrin.tistory.com/121
[AI] SSE(Feat.Streaming HTTP)
나는 채팅을 개발한다면 흔히 WebSocket을 써서 개발해야 되겠구나 하고 생각했다. 근데 보다보니 SSE?라는게 보이길래 아니 이건 또 무슨 기술이여? 하고 찾다가 정리해본다. 일단 내가 개발하는건
ssmyefrin.tistory.com
사용자 질문창이 조금 수정되었다.
1. 채팅창 Frontend 만들기
<!-- 채팅 메시지 영역 -->
<div class="chat-messages" id="chatMessages">
<div class="welcome-message">
<p>무엇을 도와드릴까요?</p>
</div>
</div>
<!-- 입력 영역 -->
<div class="chat-input">
<input
type="text"
id="messageInput"
placeholder="메시지를 입력하세요..."
autocomplete="off"
/>
<button id="sendButton">
<svg width="20" height="20" viewBox="0 0 24 24" fill="none">
<path d="M22 2L11 13M22 2l-7 20-4-9-9-4 20-7z" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"/>
</svg>
</button>
</div>
</div>
async sendMessage() {
const message = this.messageInput.value.trim();
if (!message || this.isProcessing) return;
// UI 업데이트
this.addMessage('user', message);
this.messageInput.value = '';
this.messageInput.focus();
// 처리 중 표시
this.isProcessing = true;
this.sendButton.disabled = true;
this.showTypingIndicator();
try {
// SSE 연결
const url = `/api/chat/stream?message=${encodeURIComponent(message)}&conversation_id=${this.conversationId}`;
const eventSource = new EventSource(url);
let assistantMessage = '';
let messageElement = null;
// 데이터 수신 (답변 청크)
eventSource.addEventListener('message', (e) => {
const data = JSON.parse(e.data);
if (data.chunk) {
// 타이핑 인디케이터 제거 (첫 청크에서만)
if (!messageElement) {
this.removeTypingIndicator();
messageElement = this.createAssistantMessage();
}
// 청크 추가
assistantMessage += data.chunk;
messageElement.querySelector('.message-bubble').textContent = assistantMessage;
this.scrollToBottom();
}
});
// 상태 업데이트
eventSource.addEventListener('status', (e) => {
const data = JSON.parse(e.data);
console.log('Status:', data.message);
});
// 완료
eventSource.addEventListener('done', (e) => {
const data = JSON.parse(e.data);
console.log('Done:', data);
// 시간 추가
if (messageElement) {
const timeDiv = document.createElement('div');
timeDiv.className = 'message-time';
timeDiv.textContent = new Date().toLocaleTimeString('ko-KR', {
hour: '2-digit',
minute: '2-digit'
});
messageElement.appendChild(timeDiv);
}
eventSource.close();
this.isProcessing = false;
this.sendButton.disabled = false;
});
// 에러 처리
eventSource.addEventListener('error', (e) => {
console.error('SSE Error:', e);
this.removeTypingIndicator();
if (!messageElement) {
this.addMessage('assistant', '죄송합니다. 오류가 발생했습니다. 다시 시도해주세요.');
}
eventSource.close();
this.isProcessing = false;
this.sendButton.disabled = false;
});
} catch (error) {
console.error('Error:', error);
this.removeTypingIndicator();
this.addMessage('assistant', '죄송합니다. 오류가 발생했습니다. 다시 시도해주세요.');
this.isProcessing = false;
this.sendButton.disabled = false;
}
}
SSE를 통해 연결하는 부분이다.
실행해보면 아래와 같은 화면이보인다.
2. 채팅창 Backend 만들기.
@router.get("/stream")
async def chat_stream(
message: str = Query(..., description="사용자 질문"),
conversation_id: str = Query(None, description="대화 ID")
):
"""
Tool Calling Flow:
1. LLM에게 질문 + Tool 정의 전달
2. LLM이 판단:
- Tool 불필요 → 바로 답변
- Tool 필요 → tool_calls 반환
3. Tool 실행 (JIRA/RAG)
4. 결과를 LLM에게 재전달 → 최종 답변
5. SSE로 스트리밍
"""
app_logger.info(f"📨 메시지 받음: {message}")
if not message or not message.strip():
raise HTTPException(status_code=400, detail="메시지가 비어있습니다")
async def event_generator():
"""SSE 이벤트 생성기"""
try:
# === 1. Orchestrator로 처리 (Tool Calling) ===
yield f"event: status\ndata: {json.dumps({'message': '처리 중...'}, ensure_ascii=False)}\n\n"
# ChatRequest 생성
request = ChatRequest(
message=message,
conversation_id=conversation_id
)
# Orchestrator가 Tool Calling 처리
response = await orchestrator.process_query(request)
app_logger.info(f"✅ 답변 생성 완료: {len(response.answer)} chars")
# === 2. SSE로 스트리밍 ===
yield f"event: status\ndata: {json.dumps({'message': '답변 생성 중...'}, ensure_ascii=False)}\n\n"
# 답변을 한 글자씩 스트리밍
for char in response.answer:
chunk_data = json.dumps({'chunk': char}, ensure_ascii=False)
yield f"data: {chunk_data}\n\n"
await asyncio.sleep(0.03) # 타이핑 효과
# === 3. 완료 ===
done_data = json.dumps({
'query_type': response.query_type.value if response.query_type else 'GENERAL',
'sources': [
{
'type': src.type,
'name': src.name,
'key': src.key,
'url': src.url
} for src in (response.sources or [])
],
'tool_used': response.raw_data.get('tool_used', None) if response.raw_data else None
}, ensure_ascii=False)
yield f"event: done\ndata: {done_data}\n\n"
app_logger.info("✅ 스트리밍 완료!")
except Exception as e:
app_logger.error(f"❌ 에러 발생: {e}")
import traceback
traceback.print_exc()
error_data = json.dumps({
'error': str(e),
'message': '죄송합니다. 오류가 발생했습니다.'
}, ensure_ascii=False)
yield f"event: error\ndata: {error_data}\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no"
}
)
일단 입력하면 "죄송함다 아무것도 모릅니다" 가 나올것이다.
숨을 불어넣어주자.
3. Orchestrator 만들기
def __init__(self):
self.settings = get_settings()
# Gemini 설정
genai.configure(api_key=self.settings.gemini_api_key)
# 인프라 클라이언트 초기화
self.retriever = VectorRetriever() # Qdrant 연결용
self.jira_client = JIRAClient()
# 1. Tool 정의 활성화
self.tools = self._define_tools()
# 2. Gemini 모델 초기화 (Tool 설정 포함)
self.model = genai.GenerativeModel(
model_name=self.settings.llm_model or "gemini-2.0-flash",
generation_config={
"temperature": 0.2, # RAG의 정확도를 위해 낮게 설정
"max_output_tokens": 2048,
},
tools=self.tools
)
나는 Gemini-2.0-flash 모델을 이용하였다.
그리고 핵심부분이다.
✔️ LLM에게 어떤 Tool이 있는지 알려주는 부분
def _define_tools(self) -> List[Dict[str, Any]]:
"""Gemini Function Calling을 위한 Tool 정의"""
return [
{
"function_declarations": [
{
"name": "search_jira",
"description": "JIRA에서 이슈를 검색합니다. 티켓 조회, 진행률 확인, 담당자별 이슈를 찾을 때 사용합니다.",
"parameters": {
"type": "object",
"properties": {
"jql": {
"type": "string",
"description": "JIRA Query Language. 예: 'project = ORP AND status = \"In Progress\"'"
}
},
"required": ["jql"]
}
},
{
"name": "search_documents",
"description": "구축된 Qdrant 문서 DB(RAG)에서 프로젝트 설계서, 가이드, SQL 분석 등의 정보를 검색합니다.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "문서에서 찾고자 하는 구체적인 질문 또는 키워드"
},
"module": {
"type": "string",
"description": "필터링할 모듈명"
}
},
"required": ["query"]
}
}
]
}
]
이부분이 있어야 LLM이 어떤 Tool 이 있는지 안다.
✔️ LLM 이 질문을 받아 판단하는 부분
async def process_query(self, request: ChatRequest) -> ChatResponse:
"""사용자 질문 처리 메인 로직"""
try:
app_logger.info(f"🚀 처리 시작: {request.message[:50]}...")
# 캐시 확인
app_logger.info(f" 캐시확인 ... ")
cache_key = query_cache._generate_key(request.message)
cached_response = query_cache.get(cache_key)
if cached_response:
app_logger.info(f" 캐시가 있음!! ")
return cached_response
# 1. 시스템 프롬프트 구성 및 채팅 세션 시작
system_prompt = self._build_system_prompt()
chat = self.model.start_chat(history=[])
# 2. 1차 LLM 호출 (어떤 Tool을 쓸지 판단)
app_logger.info(f"--- 1차 LLM 호출")
response = await asyncio.to_thread(
chat.send_message,
f"{system_prompt}\n\n질문: {request.message}"
)
app_logger.info(f"Full Response Object: {repr(response)}")
app_logger.info("🤔 Gemini가 질문을 분석하고 도구 사용 여부를 결정했습니다.")
# 3. Tool Calling 초기화
tool_name = None
tool_result = None
# 4. [판단 지점] Gemini가 도구를 쓸지 말지 결정함
part = response.candidates[0].content.parts[0]
app_logger.info(f"🔍 Gemini Part Data: {part}")
if part.function_call:
# 4-1. JIRA나 RAG가 필요하다고 판단하면 도구를 실행
tool_name = part.function_call.name
tool_args = dict(part.function_call.args)
app_logger.info(f"🔧 Tool 호출 결정: {tool_name}({tool_args})")
# 실제 Tool 실행 (JIRA 또는 Qdrant)
tool_result = await self._execute_tool(tool_name, tool_args)
# 5. 2차 LLM 호출 (검색 결과를 Gemini에게 전달하여 최종 답변 생성)
app_logger.info(f"--- 2차 LLM 호출")
# 객체 생성 대신 딕셔너리 구조로 전달
response = await asyncio.to_thread(
chat.send_message,
{
"role": "function",
"parts": [
{
"function_response": {
"name": tool_name,
"response": {"result": tool_result}
}
}
]
}
)
else:
app_logger.info("💬 도구 사용 없이 직접 답변을 생성합니다.")
# 6. 최종 응답 객체 조립
final_answer = response.text # 1차 응답 혹은 업데이트된 2차 응답
sources = self._extract_sources(tool_name, tool_result) if tool_name else []
chat_response = ChatResponse(
answer=final_answer,
query_type=self._map_tool_to_query_type(tool_name) if tool_name else QueryType.GENERAL,
sources=sources,
raw_data={'tool_used': tool_name, 'result_count': len(tool_result) if isinstance(tool_result, list) else 0},
processing_time=0,
conversation_id=request.conversation_id,
timestamp=datetime.now()
)
query_cache.set(cache_key, chat_response)
return chat_response
except Exception as e:
app_logger.error(f"❌ Orchestrator Error: {e}")
raise
LLM에게 숨을 불어넣었다. 그럼 요놈이 쓸수있는 Tool 들을 연결해보자.
4. RAG
호출부분
# orchestrator/core.py
self.retriever = VectorRetriever() # ← 여기
chunks = await loop.run_in_executor(
None,
self.retriever.search, # ← 이 메서드
query,
10,
module,
None,
None
)
✔️ VectorRetriever
def __init__(self):
self.client = QdrantClient(host="localhost", port=6333)
self.collection_name = "onerp_documents"
self.embedder = Embedder() # ← 임베딩 모델 로드
self._init_collection() # 컬렉션 생성/확인
def search(
self,
query: str,
limit: int = 10,
module_filter: Optional[ModuleType] = None,
artifact_type_filter: Optional[ArtifactType] = None,
screen_id_filter: Optional[str] = None
) -> List[Dict[str, Any]]: # DocumentChunk 대신 Dict
"""벡터 검색"""
try:
# 쿼리 임베딩
query_embedding = self.embedder.embed_text(query)
# 검색 실행 (필터 없이 간단하게)
results = self.client.query_points(
collection_name=self.collection_name,
query=query_embedding,
limit=limit
).points
# 간단한 dict로 변환
chunks = []
for result in results:
payload = result.payload
chunks.append({
"file_name": payload.get("file_name", "unknown"),
"chunk_id": payload.get("chunk_id", 0),
"content": payload.get("content", ""),
"file_path": payload.get("file_path", "")
})
app_logger.info(f"Found {len(chunks)} chunks for query: {query[:50]}...")
return chunks
except Exception as e:
app_logger.error(f"Error searching vector DB: {e}")
raise
✔️ Embedder
self.model_name = "paraphrase-multilingual-MiniLM-L12-v2"
self.model = SentenceTransformer(self.model_name)
self.dimension = 384 # 벡터 차원
def embed_text(self, text: str) -> List[float]:
"""
단일 텍스트를 임베딩
Args:
text: 임베딩할 텍스트
Returns:
임베딩 벡터
"""
# 캐시 확인
cache_key = self._get_cache_key(text)
cached = embedding_cache.get(cache_key)
if cached:
app_logger.debug(f"Cache hit for embedding: {text[:50]}...")
return cached
try:
# Sentence Transformers로 임베딩 생성
embedding = self.model.encode(text).tolist()
# 캐시 저장
embedding_cache.set(cache_key, embedding)
return embedding
except Exception as e:
app_logger.error(f"Error creating embedding: {e}")
raise
def embed_batch(self, texts: List[str], batch_size: int = 100) -> List[List[float]]:
"""
배치로 임베딩 생성
Args:
texts: 임베딩할 텍스트 리스트
batch_size: 배치 크기 (Sentence Transformers는 한번에 처리)
Returns:
임베딩 벡터 리스트
"""
embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
# 캐시되지 않은 텍스트만 필터링
uncached_texts = []
uncached_indices = []
batch_embeddings = [None] * len(batch)
for idx, text in enumerate(batch):
cache_key = self._get_cache_key(text)
cached = embedding_cache.get(cache_key)
if cached:
batch_embeddings[idx] = cached
else:
uncached_texts.append(text)
uncached_indices.append(idx)
# 캐시되지 않은 것만 임베딩 생성
if uncached_texts:
try:
# Sentence Transformers 배치 임베딩
new_embeddings = self.model.encode(
uncached_texts,
show_progress_bar=False
)
for idx, embedding in enumerate(new_embeddings):
original_idx = uncached_indices[idx]
batch_embeddings[original_idx] = embedding.tolist()
# 캐시 저장
cache_key = self._get_cache_key(uncached_texts[idx])
embedding_cache.set(cache_key, embedding.tolist())
app_logger.info(f"Created {len(uncached_texts)} new embeddings")
except Exception as e:
app_logger.error(f"Error creating batch embeddings: {e}")
raise
embeddings.extend(batch_embeddings)
return embeddings
✔️ 캐시 시스템
class SimpleCache:
"""간단한 TTL 캐시"""
def __init__(self, maxsize: int = 1000, ttl: int = 3600):
"""
Args:
maxsize: 최대 캐시 크기
ttl: Time To Live (초)
"""
self.cache = TTLCache(maxsize=maxsize, ttl=ttl)
def _generate_key(self, *args, **kwargs) -> str:
"""캐시 키 생성"""
key_data = {
"args": args,
"kwargs": kwargs
}
key_string = json.dumps(key_data, sort_keys=True)
return hashlib.md5(key_string.encode()).hexdigest()
def get(self, key: str) -> Optional[Any]:
"""캐시에서 값 가져오기"""
return self.cache.get(key)
def set(self, key: str, value: Any) -> None:
"""캐시에 값 저장"""
self.cache[key] = value
def delete(self, key: str) -> None:
"""캐시에서 값 삭제"""
if key in self.cache:
del self.cache[key]
def clear(self) -> None:
"""모든 캐시 삭제"""
self.cache.clear()
def exists(self, key: str) -> bool:
"""캐시 키 존재 여부"""
return key in self.cache
# 전역 캐시 인스턴스
query_cache = SimpleCache(maxsize=500, ttl=3600)
embedding_cache = SimpleCache(maxsize=10000, ttl=86400) # 24시간
같은 질문은 캐시에서 바로 반환하고(API 호출안함) 같은 내용 10번 물어봐도 임베딩은 1번만 된다.
5. API(JIRA)
def _execute_jql(self, jql: str, params: JIRAQueryParams) -> JIRAResult:
"""
JQL 실행
Args:
jql: JIRA Query Language
params: 쿼리 파라미터
Returns:
JIRA 결과
"""
try:
app_logger.info(f"Executing JQL: {jql}")
issues = self.client.search_issues(
jql,
maxResults=100,
fields='summary,status,assignee,priority,created,updated,labels,description'
)
jira_issues = []
for issue in issues:
jira_issue = JIRAIssue(
key=issue.key,
summary=issue.fields.summary,
status=issue.fields.status.name,
assignee=issue.fields.assignee.displayName if issue.fields.assignee else None,
priority=issue.fields.priority.name if issue.fields.priority else "None",
created=datetime.fromisoformat(issue.fields.created.replace('Z', '+00:00')),
updated=datetime.fromisoformat(issue.fields.updated.replace('Z', '+00:00')),
labels=issue.fields.labels or [],
description=issue.fields.description or ""
)
jira_issues.append(jira_issue)
result = JIRAResult(
issues=jira_issues,
total_count=len(jira_issues),
query_params=params
)
app_logger.info(f"Found {len(jira_issues)} issues")
return result
except Exception as e:
app_logger.error(f"Error executing JQL: {e}")
raise
def search_issues(self, jql: str, max_results: int = 20) -> List[Any]:
"""검색 인터페이스"""
try:
app_logger.info(f"Gemini requested JQL: {jql}")
issues = self.client.search_issues(
jql,
maxResults=max_results,
fields='summary,status,assignee,priority,updated'
)
return issues
except Exception as e:
app_logger.error(f"Gemini JQL execution failed: {e}")
return []
각 세부 Tool들도 다 구현해주었으니, 이제 실행해서 호출해보자.
✓ case1. JIRA(API) 호출하는 경우
Q. 지금 진행중인 이슈가 몇개지?
Response를 까보면
GenerateContentResponse(
done=True,
iterator=None,
result=protos.GenerateContentResponse({
"candidates": [
{
"content": {
"parts": [
{
"function_call": {
"name": "search_jira",
"args": {
"jql": "status = \"In Progress\""
}
}
}
],
"role": "model"
},
"finish_reason": "STOP",
"index": 0
}
],
"usage_metadata": {
"prompt_token_count": 318,
"candidates_token_count": 21,
"total_token_count": 391
},
"model_version": "gemini-2.5-flash"
}),
)
여기서 보면 중요한 부분은 바로 이부분이다.
{
name: "search_jira"
args {
fields {
key: "jql"
value {
string_value: "status = \"In Progress\""
}
}
}
}
뭘 호출해야하는지 LLM이 분석해서 Tool을 결정한다.
- Client 화면
✓ case2. RAG에서 검색하기
1) 일반질문 : Q. FCM모듈은 어떻게 되어있지?
{
name: "search_documents"
args {
fields {
key: "query"
value {
string_value: "FCM모듈"
}
}
fields {
key: "module"
value {
string_value: "FCM"
}
}
}
}
여기보면 LLM 이 search_documents Tool을 선택해야 한다고 판단한다.
2) 심화질문
ㅋㅋ 이건 RAG 데이터를 가지고 LLM이 얼마나 추론할 수 있는지 궁금해서 물어봤다.
Q. 전표 등록 관련 테이블 정의서들 내용을 종합해서, 어떤 순서로 데이터 가 인서트(Insert) 되는지 로직을 추론해서 설명해줘
- Client 화면
와.... 소름이다.ㅋㅋㅋㅋㅋㅋ
지식 파편 중에서 "전표 로직이 없네?" 하고 포기하는 대신, 내가 입력한 힌트를 듣고 테이블 정의서(SQL/Excel) 조각들을 모아 머릿속에서 시뮬레이션을 돌린 답변이다.
✓ case3. 일반질문
Q. 너 뭐할수 있는데?
라고 물어보면 Reponse를 까보면 이렇게 나온다.
GenerateContentResponse(
done=True,
iterator=None,
result=protos.GenerateContentResponse({
"candidates": [
{
"content": {
"parts": [
{
"text": "....."
}
],
"role": "model"
},
"finish_reason": "STOP",
"index": 0
}
],
"usage_metadata": {
"prompt_token_count": 314,
"candidates_token_count": 155,
"total_token_count": 469
},
"model_version": "gemini-2.5-flash"
}),
)
잘보면 여기에 아까 위랑 다르게 Function_call이 없다.
즉, 도구를 호출하지 않아도 되는 질문이라 LLM이 판단한 것.
- Client 화면
-끝-
'web > AI' 카테고리의 다른 글
| [AI] 로컬에 LLM 설치하기 (1) | 2025.12.23 |
|---|---|
| [AI] RAG 똑똑하게 만들기(1) (1) | 2025.12.22 |
| [AI] Qdrant 로 RAG 구축하기(2) (0) | 2025.12.18 |
| [AI] Tool Calling (0) | 2025.12.17 |
| [AI] RAG 와 VectorDB (1) | 2025.12.17 |