1. sqlalchemy CRUD 예제 실습

2. callcenter 프로젝트 제작중
2026-06-17 18:27:04 +09:00
parent 06eb3c57ab
commit b0503baac5
36 changed files with 1286 additions and 23 deletions
@@ -0,0 +1,9 @@
+from langchain_ibm import WatsonxEmbeddings
+from backend.config.settings import settings
+
+watson_embedding = WatsonxEmbeddings(
+    model_id="ibm/granite-embedding-278m-multilingual",
+    url=f"{settings.watsonx_url}",
+    api_key=f"{settings.watsonx_api_key}",
+    project_id=f"{settings.watsonx_project_id}",
+)
@@ -0,0 +1,17 @@
+from langchain_ibm import ChatWatsonx
+from backend.config.settings import settings
+from langchain_openai import ChatOpenAI
+
+watson_llm = ChatWatsonx(
+    model_id="ibm/granite-4-h-small",
+    url=f"{settings.watsonx_url}",
+    api_key=f"{settings.watsonx_api_key}",
+    project_id=f"{settings.watsonx_project_id}",
+    max_tokens=2000,
+)
+
+hugging_llm = ChatOpenAI(
+    base_url="https://router.huggingface.co/v1",
+    api_key=f"{settings.hf_token}",
+    model_name="Qwen/Qwen3-8B:nscale",
+)
@@ -0,0 +1,12 @@
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file="backend/.env", extra="ignore")
+    # 사용할 모델
+    watsonx_api_key: str = Field(alias="WATSONX_API_KEY")
+    watsonx_project_id: str = Field(alias="WATSONX_PROJECT_ID")
+    watsonx_url: str = Field(alias="WATSONX_URL")
+    hf_token: str = Field(alias="HF_TOKEN")
+
+settings = Settings()
@@ -0,0 +1,34 @@
+from fastapi import FastAPI
+from starlette.staticfiles import StaticFiles
+from contextlib import asynccontextmanager
+
+from backend.repository.db_init import Base, SessionLocal, engine
+from backend.routers.call_router import router as call_router
+from backend.repository.seed import seed_customers
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    print("서버 시작")
+
+    # Base 에 등록된 모든 모델에 테이블 자동 생성
+    Base.metadata.create_all(bind=engine)
+    print("[DB] 테이블 생성 완료 (또는 이미 존재)")
+
+    db = SessionLocal()
+    try:
+        # 기본 user 데이터 삽입
+        seed_customers(db)
+    finally:
+        db.close()
+
+    yield
+    print("서버 종료")
+
+# app = FastAPI()
+app = FastAPI(title="상담 LLM", version="1.0", lifespan=lifespan)
+
+# static 폴더 지정
+app.mount("/static", StaticFiles(directory="backend/static"), name="static")
+
+# 라우터 등록
+app.include_router(call_router)
@@ -0,0 +1,28 @@
+# 요약 프롬프트 생성
+SUMMARY_SYSTEM_PROMPT = """
+당신은 콜센터 상담 기록 분석 전문가입니다.
+
+다음 항목을 추출하세요.
+
+1. summary
+    - 상담 내용을 한 문장으로 요약
+
+2. keywords
+    - 상담 내용에서 중요한 키워드 3~5개 추출
+
+3. category
+    - 상담 유형
+    - 예: 장애신고, 기술지원, 요금문의, 해지문의, 서비스변경
+    
+4. sentiment
+    - 상담 내용의 감정 분석 결과 = 예: 긍정(positive), 부정(negative), 중립(neutral) 중 하나
+
+5. action_items:
+    - 상담 후 필요한 후속 조치
+
+6. customer_issue
+    - 고객이 겪은 문제
+
+7. resolution
+    - 상담사가 제공한 해결 방법
+"""
@@ -0,0 +1,23 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import DeclarativeBase, sessionmaker
+from pathlib import Path
+
+Path("db").mkdir(parents=True, exist_ok=True)
+
+# echo sql 구문 출력시키는
+engine = create_engine('sqlite:///db/callcenter.db', echo=True)
+
+# 세션 팩토리 생성
+SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+
+# Base = declarative_Base()
+class Base(DeclarativeBase):
+    pass
+
+# session 을 다른 모듈에서 사용할 수 있도록 제공
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
@@ -0,0 +1,30 @@
+from backend.repository.db_init import Base
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy import DateTime, func, String, ForeignKey
+from datetime import datetime
+
+class Customer(Base):
+    __tablename__ = 'customers'
+
+    customer_id:Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    name:Mapped[str] = mapped_column(String(50), nullable=False)
+    phone:Mapped[str] = mapped_column(String(20), nullable=False)
+
+    def __str__(self):
+        return f"<Customer id={self.customer_id} name={self.name} phone = {self.phone}]"
+
+
+# 상담기록 저장
+class CallHistory(Base):
+    __tablename__ = 'call_history'
+
+    call_id:Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    customer_id:Mapped[int] = mapped_column(ForeignKey('customers.customer_id'), nullable=False)
+    transcript:Mapped[str]
+    summary:Mapped[str]
+    category:Mapped[str] = mapped_column(String(50))
+    sentiment:Mapped[str] = mapped_column(String(20))
+    customer_issue:Mapped[str]
+    resolution:Mapped[str]
+    # created_at:Mapped[datetime] = mapped_column(server_default=func.now(), nullable=False)
+    created_at:Mapped[datetime] = mapped_column(default=datetime.now, nullable=False)
@@ -0,0 +1,28 @@
+# 강제로 회원가입
+from sqlalchemy.orm import Session
+from backend.repository.db_init import SessionLocal
+from backend.repository.models import Customer
+
+db = SessionLocal()
+
+DEFAULT_CUSTOMERS = [
+    Customer(name="홍길동", phone="010-1234-5678"),
+    Customer(name="최철수", phone="010-4321-8765"),
+    Customer(name="박영희", phone="010-5678-1234"),
+]
+
+def seed_customers(db: Session):
+    """
+    customer 테이블에 기본(연습용) 회원 데이터 삽입
+    (중복 실행 방지 : 이미 데이터가 있으면 건너뜀)
+    """
+
+    existing = db.query(Customer).first()
+    if existing:
+        print("[Seed] customer 테이블에 이미 데이터가 있습니다.")
+        return
+
+    db.add_all(DEFAULT_CUSTOMERS)
+    db.commit()
+    db.close()
+    print(f"[Seed] 기본 회원 {len(DEFAULT_CUSTOMERS)}명 생성 완료")
@@ -0,0 +1,18 @@
+from fastapi import APIRouter, Depends
+from backend.schemas.summary_schema import SummaryRequest, CallSummary, CallRequest
+from backend.services.call_service import summary_call, create_call_history
+from sqlalchemy.orm import Session
+from backend.repository.db_init import get_db
+
+
+router = APIRouter(prefix="/api/calls", tags=["Summary"])
+
+# 요약 라우터
+@router.post("", response_model=CallSummary)
+def generate_summary(req:SummaryRequest):
+    return summary_call(req.transcript)
+
+@router.post("/save", response_model=CallSummary)
+# @router.post("/save", response_model=CallRequest)
+def create_call(req: CallRequest, db:Session = Depends(get_db)):
+    return create_call_history(req, db)
@@ -0,0 +1,28 @@
+from pydantic import BaseModel
+
+class SummaryRequest(BaseModel):
+    transcript: str
+
+class CallSummary(BaseModel):
+    summary: str
+    keywords: list[str]
+    category: str
+    sentiment: str
+    action_items: list[str]
+    customer_issue: str
+    resolution: str
+
+# 상담 요청 시 사용할 타입
+class CallRequest(BaseModel):
+    customer_id: int
+    transcript: str
+
+# 상담 요약 저장 시 사용할 타입
+class CallCreate(BaseModel):
+    customer_id: int
+    transcript: str
+    summary: str
+    category: str
+    sentiment: str
+    customer_issue: str
+    resolution: str
@@ -0,0 +1,26 @@
+from langchain_community.vectorstores import Chroma
+from backend.ai.embedding import watson_embedding
+from langchain_core.documents import Document
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+from pathlib import Path
+
+def main():
+    # data 폴더 안의 파일을 읽은 후
+    data_dir = Path("data")
+
+    documents = []
+
+    # Document 객체 생성
+    for file_path in data_dir.glob("*.txt"):
+        content = file_path.read_text(encoding="utf-8")
+        documents.append(Document(page_content=content, metadata={"source": str(file_path.name)}))
+
+    # 분할
+    splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=30)
+    splite_docs = splitter.split_documents(documents)
+
+    # 인덱스 설정(벡터 db) ./vectordb
+    Chroma.from_documents(documents = splite_docs, embedding=watson_embedding, persist_directory="./vectordb")
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,72 @@
+from backend.ai.llm import hugging_llm
+from langchain_core.prompts import ChatPromptTemplate
+from backend.prompts.all_prompt import SUMMARY_SYSTEM_PROMPT
+from backend.repository.models import CallHistory
+from backend.schemas.summary_schema import CallSummary, CallCreate
+from sqlalchemy.orm import Session
+
+from backend.schemas.summary_schema import CallRequest
+
+# LLM transcript 요약 시키기
+def summary_call(transcript: str):
+
+    summary_prompt = ChatPromptTemplate.from_messages(
+        [
+            ("system", SUMMARY_SYSTEM_PROMPT),
+            ("human", "상담내용\n{transcript}"),
+        ]
+    )
+
+    structured_llm = hugging_llm.with_structured_output(CallSummary)
+    summary_chain = summary_prompt | structured_llm
+
+    result = summary_chain.invoke({"transcript" : transcript})
+    return result
+
+def save_call_history(db, data):
+    """상담 저장"""
+    history = CallHistory(
+        customer_id = data.customer_id,
+        transcript = data.transcript,
+        summary = data.summary,
+        category = data.category,
+        sentiment = data.sentiment,
+        customer_issue = data.customer_issue,
+        resolution = data.resolution
+    )
+
+    db.add(history)
+    db.commit()
+    db.refresh(history)
+
+    return history
+
+def evaluate_call():
+    """상담  평가"""
+
+    pass
+
+def save_call_evaluation():
+    """상담 평가 내용 저장"""
+
+    pass
+
+def create_call_history(req: CallRequest, db:Session):
+    # 요약
+    # CallSummary
+    summary = summary_call(req.transcript)
+    print("summary",summary)
+    # 데이터베이스 저장용 객체
+    call_data = CallCreate(
+        customer_id = req.customer_id,
+        transcript = req.transcript,
+        summary = summary.summary,
+        category = summary.category,
+        sentiment = summary.sentiment,
+        customer_issue = summary.customer_issue,
+        resolution = summary.resolution
+    )
+
+    return save_call_history(db = db, data = call_data)
+
+