[Backend] 충돌 방지

동시 접속 충돌 방지

API 키

기존 코드에서는 전역 변수를 활용하여 사용자에게 API 키를 제공했습니다.
이러한 방식은 여러 사용자의 동시 접속 상황에서 문제가 발생할 수 있습니다.

API 키 호출

def get_api_key(db: Session, user_email: str, provider: str):
    user = db.query(User).filter(User.email == user_email).first()

    user_id = user.id
    if provider == 'openai':
        api = db.query(ApiKey).filter(ApiKey.user_id == user_id, ApiKey.provider_id==4).first()
        #config.GPT_API = api.api_key
        get_embedding_key(db=db)
        return api.api_key
    elif provider == "anthropic":
        api = db.query(ApiKey).filter(ApiKey.user_id == user_id, ApiKey.provider_id==2).first()
        #config.CLAUDE_API = api.api_key
        get_embedding_key(db=db)
        return api.api_key

우선 API 키를 호출하는 코드를 수정합니다.
전역 변수를 직접 변경하는 방식이 아니라, API 키를 DB에서 검색하고 반환만 합니다.

LLM 호출 코드 변경

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
import core.config as config

def get_llm(provider="openai", model=None, api_key : str = None):
    if provider == "openai":
        model_name = model or config.DEFAULT_CHAT_MODEL
        return ChatOpenAI(
            openai_api_key = api_key,
            model_name=model_name,
            temperature=0.7
        )
    elif provider == "anthropic":
        model_name = model or "claude-3-sonnet-20240229"
        return ChatAnthropic(
            anthropic_api_key = api_key,
            model=model_name,
            temperature=0.7
        )
    else:
        raise ValueError(f"지원되지 않는 제공자: {provider}")

LLM을 호출할 때에도 전역 변수가 아닌, 함수에서 입력받는 방식으로 바꿉니다.

LangChian 코드 변경

def qa_chain(db: Session, session_id, project_id, user_email, conversation, provider="openai", model=None, api_key : str = None):
    llm = get_llm(provider, model, api_key = api_key)
    vector = text_to_vector(conversation)
    relevant_messages = get_relevant_messages(db, session_id, vector, top_n=5)

    formatted_history = "\n".join(
        [f"{msg['message_role'].capitalize()}: {msg['conversation']}" for msg in relevant_messages]
    )

    prompt = PromptTemplate(
        input_variables=["history", "input"],
        template="{history}\nHuman: {input}\nAI:"
    )
    chain = prompt | llm | StrOutputParser()

    if provider == "openai":
        with get_openai_callback() as cb:
            response_text = chain.invoke({"history": formatted_history, "input": conversation})
            update_usage(db = db, user_email = user_email, provider = provider, usage = cb.total_tokens)
            print(f"[LLM 사용량 - OpenAI] prompt: {cb.prompt_tokens} / completion: {cb.completion_tokens} / total: {cb.total_tokens} / cost: ${cb.total_cost:.6f}")
    elif provider == "anthropic":
        prompt_tokens = count_tokens(formatted_history)
        response_text = chain.invoke({"history": formatted_history, "input": conversation})
        completion_tokens = count_tokens(response_text)
        cost_data = estimate_claude_cost(model, prompt_tokens, completion_tokens)
        update_usage(db=db, user_email=user_email, provider=provider, usage=cost_data['total'])
        print(f"[LLM 사용량 - Claude] prompt: {cost_data['prompt']} / completion: {cost_data['completion']} / total: {cost_data['total']} / cost: ${cost_data['cost']:.6f}")
    else:
        response_text = chain.invoke({"history": formatted_history, "input": conversation})
        print("[LLM 사용량] 추적 불가: 미지원 provider")

    vector2 = text_to_vector(response_text)
    add_message(db=db, session_id=session_id, project_id=project_id, user_email=user_email,
                message_role='user', conversation=conversation, vector_memory=vector)

    add_message(db=db, session_id=session_id, project_id=project_id, user_email=user_email,
                message_role='assistant', conversation=response_text, vector_memory=vector2)

    return response_text

LangChain 기반 질문과 답변 관련 Chain에서도 API Key를 직접 호출될 때 입력받아서,
get_llm 함수에 넘겨주는 방식으로 변경합니다.

엔드포인트

@langchain_router.post('/RequestMessage')
async def request_message(request: RequestMessageRequest, db: Session = Depends(get_db)):
    email = request.user_email
    project_id = request.project_id
    message = request.messageInput
    session = request.session
    model = request.selected_model
    first = is_this_first(db=db, session_id = session)
    if model in config.OPENAI_MODELS:
        provider = "openai"
    elif model in config.ANTHROPIC_MODELS:
        provider = "anthropic"
    else:
        return JSONResponse(content={"message": "해당 모델은 META LLM MSP에서 제공하지 않는 모델입니다."})
    api_key = get_api_key(db=db, user_email=email, provider=provider)

    if first:
        agent_executor = get_session_agent(session)
        response = agent_executor(message)
        change_session_title(db=db, session_id=session, content=response.content)
    a = qa_chain(db = db, session_id=session, project_id=project_id, user_email=email, conversation=message, provider=provider, model=model, api_key=api_key)
    return a

마지막으로 메세지를 입력받을 때,
Get API KEY 함수로 API 키를 검색해서 모든 답변 생성을 통합적으로 관리하는 QA Chain 코드에 전달합니다.

API 관련 오류

기존 코드

원래 보유한 API 키가 없는데 해당 LLM을 사용하려고 하면 오류가 발생하는 문제가 있었습니다.

엔드포인트 수정

@langchain_router.post('/RequestMessage')
async def request_message(request: RequestMessageRequest, db: Session = Depends(get_db)):
    email = request.user_email
    project_id = request.project_id
    message = request.messageInput
    session = request.session
    model = request.selected_model
    first = is_this_first(db=db, session_id = session)
    if model in config.OPENAI_MODELS:
        provider = "openai"
    elif model in config.ANTHROPIC_MODELS:
        provider = "anthropic"
    else:
        return "해당 모델은 아직 지원되지 않는 모델입니다.\n다른 모델을 선택해주세요."
    api_key = get_api_key(db=db, user_email=email, provider=provider)
    if not api_key:
        return "보유 중인 API키가 없습니다.\n우선 API키를 등록해주세요."

    if first:
        agent_executor = get_session_agent(session)
        response = agent_executor(message)
        change_session_title(db=db, session_id=session, content=response.content)
    try :
        a = qa_chain(db = db, session_id=session, project_id=project_id, user_email=email, conversation=message, provider=provider, model=model, api_key=api_key)
        return a
    except Exception as e:
        print(f"Error Occured f{e}")
        return "현재 등록하신 API 키는 유효하지 않습니다.\n유효하는 API키를 등록해주세요."