1. 기존 지식 관리 시스템의 한계와 개선 방안
대부분의 기업은 내부 정보 공유 과정에서 다음과 같은 비효율성을 겪고 있습니다:
- 검색 효율성 부재: 단순 키워드 매칭 방식에 의존하여 유사 어구나 자연어 질문을 인식하지 못합니다.
- 정보 통합 부족: 복수의 문서 목록만 제공되며, 사용자는 직접 내용을 요약해야 합니다.
- 유지보수 복잡성: 정보가 다양한 툴 (Confluence, Notion 등) 에 분산되어 있어 실시간 동기화가 어렵습니다.
이에 따라 OpenAI 에서 제공하는 Assistants API 를 도입하면, 별도의 RAG(검색 증강 생성) 아키텍처를 구축하는 부담 없이 수 주 내에 정확도 높은 대화형 시스템을 배포할 수 있습니다.
2. Assistants API 핵심 구성 요소 이해
이 프레임워크는 크게 5 가지 주요 구성 단위로 이루어져 있으며, 각 역할은 다음과 같습니다:
| 구성 요소 | 주요 기능 |
|---|---|
| Assistant | 모델 선택, 프롬프트 설정, 도구 연결 등의 기본 에이전트 속성 관리 |
| Thread | 사용자와 AI 간의 대화 이력을 보관하며 컨텍스트 윈도우 자동 제어 |
| Message | 텍스트, 파일, 이미지 등을 포함하는 상호작용 데이터 단위 |
| Run | 단일 요청 실행 인스턴스로, 백그라운드에서 도구를 호출 및 응답 생성 |
| Tool | 파일 검색 (File Search), 코드 인터프리터, 외부 함수 호출 지원 |
이 구조를 활용하여 기업 전용 데이터를 학습된 에이전트에 안전하게 부여하고, 권한 관리 하에 접근을 제어할 수 있습니다.
3. 시스템 구현 단계별 프로세스
3.1 개발 환경 초기화
구현을 위해 Python 3.10 이상의 버전과 관련된 라이브러리들이 필요합니다. 다음 명령어로 필수 패키지를 설치합니다.
pip install openai fastapi uvicorn python-dotenv streamlit requests pydantic프로젝트 디렉토리 구조는 다음과 같이 설계됩니다.
enterprise_rag_system/
├── .env # 보안 키 관리
├── data_storage # 로컬 문서 파일 저장소
├── assistant_setup.py # 보조 프로그램 초기화 스크립트
├── server_api.py # 백엔드 FastAPI 서버
└── web_app.py # 사용자 인터페이스 (Streamlit).env 파일에는 API 인증키와 모델 지정값을 기록합니다.
OPENAI_API_KEY=sk-proj-xxxxx
TARGET_MODEL=gpt-4o
ASSISTANT_ID=asst_xxx
PUBLIC_STORE_ID=vs_xxx3.2 지식 데이터베이스 및 에이전트 생성
우선 조직별로 분류된 문서를 벡터 저장소에 업로드하고, 이를 참조할 AI 에이전트를 등록해야 합니다. 아래 스크립트는 특정 부서마다 다른 저장을 생성하는 예시입니다.
import os
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def setup_corporate_knowledge_base():
departments = ["hr", "finance", "engineering"]
store_refs = {}
# 부서별 벡터 저장소 생성
for dept in departments:
new_store = client.beta.vector_stores.create(name=f"KB_{dept}")
store_refs[dept] = new_store.id
doc_path = f"./data_storage/{dept}"
if os.path.exists(doc_path):
files = [open(f, "rb") for f in os.listdir(doc_path)]
batch = client.beta.vector_stores.file_batches.upload_and_poll(
vector_store_id=new_store.id,
files=files
)
print(f""{dept}:" {batch.file_counts.completed}개 파일 처리 완료")
# Assistant 정의 및 규칙 설정
agent = client.beta.assistants.create(
name="사내 지식 상담원",
instructions="""
당신은 회사의 공식 지식 파트너입니다.
제공된 문서 밖의 정보는 추측하지 마시고 반드시 '확인 불가'라고 답변하세요.
모든 답변 뒤에 출처 파일명을 명시하세요.
""",
tools=[{"type": "file_search"}, {"type": "code_interpreter"}],
model=os.getenv("TARGET_MODEL")
)
print(f"에이전트 ID 생성됨: {agent.id}")
return agent.id, store_refs
if __name__ == "__main__":
setup_corporate_knowledge_base()3.3 백엔드 서비스 구현 (FastAPI)
백엔드는 OAuth2 기반 인증을 통해 사용자 역할을 식별하고, 해당 역할에 허용된 지식베이스로 제한하여 조회하도록 설계되었습니다.
import json
from fastapi import FastAPI, Depends, HTTPException, status
from pydantic import BaseModel
from typing import List, Optional
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
app = FastAPI(title="Enterprise Knowledge Service")
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
class MessageInput(BaseModel):
user_query: str
current_thread: Optional[str] = None
role_access_map = {
"admin": ["public", "hr", "finance"],
"engineer": ["public", "engineering"],
"staff": ["public"]
}
@app.post("/query")
async def get_answer(req: MessageInput):
try:
thread_id = req.current_thread
# 세션 유지 또는 새로 생성
if not thread_id:
thread = client.beta.threads.create()
thread_id = thread.id
client.beta.threads.messages.create(
thread_id=thread_id,
role="user",
content=req.user_query
)
run = client.beta.threads.runs.create_and_poll(
thread_id=thread_id,
assistant_id=os.getenv("ASSISTANT_ID"),
tool_resources={
"file_search": {
"vector_store_ids": [os.getenv('PUBLIC_STORE_ID')]
}
}
)
if run.status == "completed":
msgs = client.beta.threads.messages.list(thread_id=thread_id, limit=1)
response_text = msgs.data[0].content[0].text.value
return {"status": "success", "id": thread_id, "reply": response_text}
else:
raise HTTPException(status_code=500, detail="Processing failed")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))3.4 프론트엔드 애플리케이션 (Streamlit)
실제 사용자 상호작용을 위한 간단한 대시보드입니다.
import streamlit as st
import requests
API_URL = "http://localhost:8000/query"
st.title("사내 지식 검색기")
if "session_thread" not in st.session_state:
st.session_state.session_thread = None
if "chat_history" not in st.session_state:
st.session_state.chat_history = []
prompt = st.text_input("질문을 입력해주세요.")
if st.button("조회"):
payload = {
"user_query": prompt,
"current_thread": st.session_state.session_thread
}
res = requests.post(API_URL, json=payload)
if res.status_code == 200:
data = res.json()
st.markdown(data["reply"])
st.session_state.session_thread = data["id"]
st.session_state.chat_history.append({"q": prompt, "a": data["reply"]})4. 성능 및 비용 최적화 전략
시스템 운영 단계에서는 다음과 같은 기술적 조정이 필요합니다.
- 문서 정제: 불필요한 헤더, 푸터, 이미지 캡션을 제거하여 토큰 사용량을 줄이고 관련성 점수를 높입니다.
- 모델 선택: 복잡한 추론 작업에는 GPT-4o 를 사용하고, 단순 조회에는 GPT-3.5-turbo 를 사용하여 비용을 절감합니다.
- 캐싱 전략: 빈번히 발생하는 질문 패턴을 Redis 등으로 메모리에 저장하여 API 호출 횟수를 최소화합니다.
5. 보안 및 규정 준수 고려사항
중요한 개인정보가 포함된 문서는 공개 클라우드에 직접 업로드하기 위험할 수 있습니다. 이를 방지하기 위해 로컬 임베딩 서버를 구동하여 검색을 수행하고, 찾은 스니펫만 AI 모델로 전달하는 하이브리드 방식을 사용할 수 있습니다.
6. 자주 묻는 질문 (FAQ)
Q. 지원되는 파일 크기와 포맷은 무엇인가요?
Vector Store 는 최대 100GB 까지 저장 가능하며, PDF, TXT, Markdown, JSON 등 다양한 포맷을 지원합니다.
Q. 데이터 훈련 사용 여부는 어떻게 확인하나요?
OpenAI 의 API 정책상 사용자 데이터를 모델 학습에 사용하지 않으며, 특정 계약 조건에 따라 데이터 삭제 옵션을 적용할 수 있습니다.
Q. 동시 접속 가능 사용자가 어느 정도인가요?
기본 속도 제한은 분당 1,000 건이며, 필요 시 증가 신청을 통해 더 많은 트래픽을 처리할 수 있습니다.