에이전트가 사용자의 여행 유형을 어떻게 대화에 반영하고, 사용자의 여행 선호도를 어떻게 분석하는지 코드를 통해 살펴보자.

(직접 코드로 보는게 이해가 더 빠를수도 있으나 짧게 설명을 진행합니다)

여행 가이드 에이전트 전체 코드

LangGraph 에이전트 작동 흐름

이전 게시물에서 이미 작동 흐름에 대해 설명하였다.

코드의 이해를 위해 LangGraph에 대해 조금 설명하자면, LangGraph는 그래프 기반의 워크 플로우로 노드(Node)와 엣지(Edge), 상태(State)를 활용하여 위와 같은 흐름으로 구조화 할 수 있다.

보라색 네모칸은 LangGraph에서 노드를 의미하며 각 노드는 하나의 작업을 수행하는 함수로 상태(State)를 매개변수로 받아 업데이트된 상태(State)를 반환한다. 에이전트의 상태(State)는 여러 노드가 작업을 수행하며 값을 공유하고 업데이트하는 저장소이다. 이렇게 구현된 노드를 엣지(Edge)를 통해 연결하는데 여러 조건 분기도 가능하다.

자세한 LangGraph의 설명은 공식 문서를 살펴보자🙂 LangGraph 공식 문서

1. 메인 코드

# 사용자별 에이전트 생성 함수
def create_user_agent(userId: Optional[str], model: LanguageModelLike, tools: Union[ToolExecutor, Sequence[BaseTool], ToolNode]) -> CompiledGraph:
    # 사용자별 mcheckpointer 가져오기
    user_checkpointer = get_user_checkpointer(userId)

    # 에이전트 생성
    return create_my_agent(
        model=model,
        tools=tools,
        checkpointer=user_checkpointer
    )

class QuestionRequest(BaseModel):
    userMessage: str
    userId: Optional[str] = None
    tbtiType: Optional[str] = None


class AiResponse(BaseModel):
    typeNum: Optional[int] = None
    answer: str
    place: Optional[List[Dict]] = None


app = FastAPI()

# 호출할 함수 리스트 가져오기
tools = tools_of_travel["list_of_func"] + tools_of_type["list_of_func"]


# 이전 state 값 저장
previous_state = {
    "messages" : None,
    "previous_result" : None,
    "final_response" : None,
    "tbti_of_user" : None,
    "filtering" : {}
}

db = database

@app.post("/ask-ai/", response_model=AiResponse)
async def ask_ai(request: QuestionRequest):
    global previous_state
    userId = request.userId or "false"
    userMessage = request.userMessage
    tbtiType = request.tbtiType

    try:
        db.reconnect()

        # 사용자 ID를 기반으로 에이전트 생성
        user_agent = create_user_agent(userId, llm, tools)

         # TBTI 유형 저장
        previous_state["tbti_of_user"] = tbtiType

        # 사용자 ID를 포함한 설정 구성
        config = {"configurable": {"thread_id": userId, "user_id": userId}}

        system_prompt = """
        - You are a tour guide called 'TBTI'. Ask the user a short and clear question.
        - Only up to five locations will be notified.
        - Don't ask a question what type of trip the user wants.
        - Don't ask specifically what kind of trip users want.
        """

        messages_list = [("system", f"{system_prompt}")] 
        messages_list.append(("human", f"{userMessage}"))
        previous_state["messages"] = messages_list

        # 에이전트 실행
        response = user_agent.invoke(previous_state, config)
        previous_state = response

        # JSON 직렬화 시 SecretStr 값 처리
        return response['final_response']

    except Exception as e:
        print("에러 발생: ", e)
        raise HTTPException(status_code=500, detail="AI 처리 중 오류 발생")
    finally:
        db.unconnect()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8001)

FastAPI를 통해 사용자의 질문을 받고 프론트단으로 답변을 전달한다.

• 전달받는값: 사용자 질문, 사용자 ID, 사용자의 여행 유형(TBTI)
• 응답값: typeNum, AI 챗봇 응답, 여행지 추천 장소

주요한 부분은 create_user_agent에서 사용자만의 에이전트를 생성하고, 사용자의 질문 및 시스템 프롬프트를 에이전트 상태(State)에 저장하여 에이전트를 실행시킨다.

previous_state 딕셔너리에 에이전트의 이전 상태값을 저장하는 이유는 사용자의 여행 유형이나 이전 대화 기록을 통해 사용자와의 대화를 자연스럽게 연결하기 위함이다.

이제 에이전트의 구현 코드를 살펴보자

2. 에이전트 구현 코드

def create_my_agent(
    model: LanguageModelLike,
    tools: Union[ToolExecutor, Sequence[BaseTool], ToolNode],
    checkpointer: Optional[BaseCheckpointSaver] = None
) -> CompiledGraph:

    # LLM 시스템 프롬프트 리스트 로드 - 작동되는 함수에 따라 시스템 프롬프트 내용 다름
    configuration_for_answers = system_informations_of_functions

    # 도구 호출 가능한 모델 로드
    model_with_tools = model.bind_tools(tools)

    def save_user_info(state: AgentState):
        return {'tbti_of_user': state["tbti_of_user"]}

                                ...... 더보기

create_my_agent 는 에이전트 생성 함수로 LLM, LLM 외부 도구(Tools), 채팅 메모리를 매개변수로 받아 CompiledGraph를 리턴한다. 이 함수에는 노드(Node)로 작동되는 작업들이 구현되어 있으며 노드와 엣지로 작업 흐름 또한 구현한다.

에이전트의 상태 및 작업 함수들을 하나씩 살펴보자.

2-1. 에이전트 상태(State)

# 노드에 전달되는 state
class AgentState(TypedDict):
    messages : Annotated[list, add_messages] 
    previous_result : Optional[str]                   
    final_response : Optional[dict]                  
    tbti_of_user : Optional[str]             
    filtering : Optional[dict]              

LangGraph에서 에이전트의 상태(State)를 TypedDict을 사용하여 다음과 같이 정의할 수 있고, 모든 노드에서 이 값들을 공유하고 수정할 수 있다. 여행 가이드의 경우로 다섯가지의 요소만 저장을 했다.

• messages : 대화 기록 저장
• previous_result : 이전 단계에서의 생성된 AI 답변
• final_response : 사용자에게 전달될 최종 답변
• tbti_of_user : 사용자의 여행 유형 TBTI
• filtering : 사용자의 여행 특성 (데이터베이스 검색 필터 조건) 이 값은 사용자의 여행 유형에 따라 추가되는 필터로 사용자에게 맞는 여행지를 검색할 때 사용되는 정보이다.

2-2. 노드(Node) 작업 구현

노드의 흐름대로 노드에 어떤 작업이 이뤄지는지 설명하고자 한다.

1. 사용자의 여행 유형 저장

    # 첫 노드
    def save_user_info(state: AgentState):
        return {'tbti_of_user': state["tbti_of_user"]}

    # 검색 필터를 생성해야 하는지 파악
    def should_create_filter(state: AgentState):
        filtering = state['filtering']
        tbti = state['tbti_of_user']

        if len(filtering) > 0:
            return "start talking"
        elif len(filtering) == 0 and tbti != None:
            return "create filter"
        else:
            return "start talking"

사용자의 여행 유형을 서버에서 받게 되면 유형을 에이전트 상태(State)에 저장하고, 여행 유형을 에이전트가 처음 받게 되면 검색 필터가 존재하지 않는다.

따라서 should_create_filter를 통해 검색 필터가 없는지 파악하고, 없다면 create filter 문자열을 리턴한다. 리턴된 값을 통해 검색 필터를 생성하는 작업 함수로 이동할 수 있다. 이 함수는 조건부 엣지(Conditional Edge)를 만들기 위해 사용되는데 에이전트 작동 흐름을 다시 보면 첫 노드 start-node에서 두 갈래로 나누어진 조건부 엣지를 확인할 수 있다.

2. 검색 필터 생성 및 시스템 프롬프트 추가

# 새로운 검색 필터 생성 노드
    def generate_new_filter(state: AgentState):
        messages = state["messages"]
        filtering = {}
        system_prompt = ['Ask a question in Korean one by one.']

        # 사용자 여행 유형 가져오기
        tbti = state['tbti_of_user']
        try:
            if tbti not in ['AIEU', 'AIEP', 'AIFU', 'AIFP', 'ASEU', 'ASEP', 'ASFU', 'ASFP', 'CIEU', 'CIEP', 'CIFU', 'CIFP', 'CSEU', 'CSEP', 'CSFU', 'CSFP'] and tbti == None:
                return {"filtering": filtering, "messages": messages}

            else:
                # 여행 유형에 따른 검색 필터 및 추가 질문 준비
                tbti = list(tbti)
                for one_type in tbti:
                    match one_type:
                        case 'A':
                            filtering["mood"] = "(mood == 0)"
                        case 'C':
                            filtering["mood"] = "(mood == 1)"
                        case 'I':
                            system_prompt.append(tools_of_type['check_companion_animal']["added_system_message"])
                        case 'P':
                            pass
                        case 'S':
                            system_prompt.append(tools_of_type['check_child']["added_system_message"])
                            system_prompt.append(tools_of_type['check_companion_animal']["added_system_message"])
                        case 'E':
                            system_prompt.append(tools_of_type['check_distance']["added_system_message"])
                        case 'F':
                            filtering["parking"] = "(parking == true)"
                        case 'U':
                            filtering["reservation"] = "(reservation == true)"

        except Exception as e:
            print(e, " 올바른 TBTI 유형을 전달하세요.")

        added_system_msg = ' '.join(system_prompt)
        messages = [
            ("system", f"{added_system_msg}")
        ]

        return {"filtering": filtering, "messages": messages}

위의 작업은 사용자의 여행 유형에 따른 검색 필터를 에이전트 상태(State)에 저장한다. 특정 유형에 경우, 추가적인 여행 특성을 구체적으로 파악하기 위해 시스템 프롬프트를 추가하여 LLM이 사용자의 여행에 대해 추가적인 질문을 하도록 유도한다.
만약 사용자의 여행 유형이 APSU 인 경우 👆

• 검색 필터 filtering: mood == 0, reservation == True 저장
• S 유형에 속하기 때문에 LLM의 시스템 프롬프트에는

  "Ask first if users travel with their children."
  "Ask first if users are going to travel with their pets."

위의 두 문장이 추가되어 사용자 여행에 대해 구체적으로 알기 위한 LLM의 추가 질문이 이어질 수 있다.

3. AI 모델 로드 및 답변 생성

 # 사용할 AI 모델 로드 및 AI 답변 처리
    def talk_to_model(state: AgentState):
        response = model_with_tools.invoke(state['messages'])
        last_response = response.content.replace('\"', '\'')
        last_response = f'{{\"answer\": \"{last_response}\", \"place\": null}}'

        # AI 답변을 json 형식의 문자열로 만들어 previous_result에 저장 / 답변 history에 저장 
        return {"previous_result" : last_response , "messages" : [response]}

     # 도구를 작동 시킬 지 파악
    def should_continue(state: AgentState):
        messages = state["messages"]
        last_message = messages[-1]

        # 함수 호출이 없으면 바로 사용자에게 리턴
        if not last_message.tool_calls:
            return "pass"
        # 있으면 워크플로우 지속
        else:
            return "work"

상태(State)에 사용자의 여행 유형과 검색 필터가 모두 저장되고 나면 사용자의 질문에 대한 LLM 답변을 생성하는 작업으로 진행된다.

LLM 답변을 생성하여 JSON 형식의 문자열로 만들고 리턴하며 상태 속 previous_result 요소에 답변을 저장한다.

사용자에 질문을 LLM이 받게 되면 단순한 답변을 바로 생성해도 되는지, 추가 작업을 위해 외부 도구 호출이 필요한지 파악한다. 외부 도구 호출이 필요할 경우, should_continue 함수에서 work 문자열을 리턴하여 도구(Tools) 노드로 이동시킨다. 호출 가능한 외부 도구는 아래 경로로 확인할 수 있다.

여행 가이드 챗봇 외부 도구

4. 필터링 추가 기능

# 검색 필터를 추가할 지, 검색 결과를 통한 답변을 생성할 지 파악
    def should_make_answer(state: AgentState):
        name_of_tools = tools_of_type.keys()
        tool_messages = state["messages"][-1]

        if tool_messages.name in name_of_tools:
            return "add type"
        else:
            return "make answer"

    # 여행 취향을 파악하기 위한 추가 질문 답변 처리
    def process_type_result(state: AgentState):
        filtering = state["filtering"]     
        messages = state["messages"]

        ai_message = filter_messages(messages, include_types=[AIMessage])[-1]
        tool_call_ids = [item['id'] for item in ai_message.tool_calls]

        # 이전 실행된 tool 결과 가져오기
        tool_messages =  filter_messages(messages, include_types=[ToolMessage], include_ids=tool_call_ids)
        for tool in tool_messages:
            content = tool.content
            split_result = content.split(',')
            if split_result[1] != 'null':
                filtering[split_result[0]] = split_result[1]
            else:
                if split_result[0] in filtering:
                    del filtering[split_result[0]]

        print('filteirng: ', filtering)
        return {"filtering": filtering}

여행 가이드 챗봇이 가진 외부 도구에는 여행지 추천, 여행 계획 수립 도구가 있지만, 사용자의 여행 특성을 추가하는 도구도 존재한다.

만약 👆

• 사용자 질문: 나 강아지랑 여행가고 싶어

  # 반려동물과 함께 여행하는지 아닌지에 대한 답변에 작동되어 답변 저장
  @tool
  def check_companion_animal(response: bool) -> str:
  """check whether the users are going to travel with their pets or not and save their answers to the parameters.
  
    Args:
      response: if the user is going to travel with a pet, put 'true'. if not, put 'false'.
  """
  
  if response == True:
    filter_string = 'animal,(animal == true)'
  else:
    filter_string = 'animal,null'
  return filter_string

  위의 도구가 실행되고, 위의 도구는 animal,(animal == true) 문자열을 리턴한다. 위 값은 process_type_result 함수에서 후처리 되어 에이전트 상태(State)의_ filtering_에 (animal == true) 값이 추가되도록 한다.

결과적으로, 사용자가 어떠한 여행을 원하는지 구체적인 정보를 이러한 방식으로 알아낼 수 있고, 사용자의 여행 특성이 변화한다면 즉각적으로 수정하여 LLM이 반영할 수 있도록 한다.

❓필터링(filtering) 값은 언제 쓰이는가?

@tool
def recommand_travel_destination(question : str, location : str, area : str, filtering: Annotated[dict, InjectedState('filtering')]) -> str:
    """
    recommand the various places that user wants to know or to travel
    It only works when user wants to know the various places.
    It doesn't work when user told to plan the trip and when user told to reserve the place.

    Args:
        question: input the user's questions.
        location: input the area of Korea to travel, e.g. 서울 or 부산 or 대구 or 강원도
        area: Enter only the following words to indicate where the place in the user's question belongs to the following Korean administrative districts. e.g. 강원특별자치도 
              - 한국 행정 구역 : 서울특별시, 부산광역시, 인천광역시, 대구광역시, 대전광역시, 광주광역시, 울산광역시, 세종특별자치시, 경기도, 충청북도, 충청남도, 전라남도, 경상북도, 경상남도, 강원특별자치도, 전북특별자치도, 제주특별자치도
    """
    milvus = database

    # 사용자 질문 벡터화
    vector  =  embedding(question)

    # 필터링 생성 후 테이블 검색 진행
    milvus_filter = None
    area_filter = f"area_name == '{area}'"

    if bool(filtering):
        area_filter = area_filter + ' && '
        another = ' && '.join(filtering.values())
        milvus_filter = area_filter + another
    else:
        milvus_filter = area_filter

    results_localCreator, results_nowLocal = milvus.search_all_tables(embedding=vector, filtering=milvus_filter, top_k=5)

    # 쿼리 결과 합치기
    total_results = milvus.get_formatted_results(results_localCreator, results_nowLocal)

    return f"user question: {question} \n\nreference: \n{total_results}"

여행지 추천 함수가 작동 시 위의 함수의 파라미터로 가져와,** milvus 데이터 베이스** 필터 조건으로 쓰여 사용자에게 맞는 장소만 가져오는 역할을 하게됨

5. 도구 호출 후, AI 답변 생성 및 전달

# 도구 작동 후 함수 결과 LLM에게 최종 전달 후 답변 생성
    def respond_after_calling_tools(state: AgentState):
        # 작동된 마지막 도구 메시지 가져오기
        messages = state["messages"]
        last_tool_message = messages[-1]

        # 작동된 함수 이름 가져오기
        name_of_functions_called = last_tool_message.name

        # 함수 리턴값 가져오기 = 검색 결과 가져오기
        reference = last_tool_message.content

        # 작동된 도구에 맞는 시스템 프롬프트 가져오기
        system_prompt = configuration_for_answers[name_of_functions_called]['system_prompt']
        response_format = configuration_for_answers[name_of_functions_called]['response_format']

        # 결과 참고해서 LLM 답변 생성
        messages = [
            {"role":"system", "content": f"{system_prompt}"},
            {"role": "user", "content":f"{reference}"}
        ]

        llm_response = chat_completion_request(
            messages=messages,
            response_format=response_format
        ).choices[0].message.content   

        return {"previous_result": llm_response}

     def post_processing_of_answer(state: AgentState):
        ai_answer = state["previous_result"]
        escaped_response = escape_json_strings(ai_answer)
        return {"final_response" : escaped_response}

일반적인 여행지 추천과 같은 외부 도구가 실행된 후, 실행 결과를 LLM이 참고하여 답변을 생성할 수 있도록 한다. 여러가지 외부 도구들이 존재할 것이고, 작동된 도구에 맞는 시스템 프롬프트, LLM의 답변 형식을 불러와 LLM을 실행한다. LLM의 답변은 에이전트 상태의 _previous_result_에 저장한다.

post_processing_of_answer 노드에서는 json 문자열을 딕셔너리 자료형으로 변환하는 작업을 수행하고, 최종 답변으로 에이전트 상태(State)에 저장한다.

6. 에이전트 노드 생성 및 흐름 구현

  # 새로운 그래프 정의
    workflow = StateGraph(AgentState)

    # 각 노드 생성
    workflow.add_node("start-node", save_user_info)
    workflow.add_node("generate-filter", generate_new_filter)
    workflow.add_node("talk-to-human", talk_to_model)
    workflow.add_node("tools", ToolNode(tools))
    workflow.add_node("add-filter", process_type_result)
    workflow.add_node("respond", respond_after_calling_tools)
    workflow.add_node("json-processing", post_processing_of_answer)

    # 그래프 진입 포인트 설정
    workflow.set_entry_point("start-node")

    workflow.add_conditional_edges(
        "start-node",
        should_create_filter,
        {
            "start talking": "talk-to-human",
            "create filter": "generate-filter",
        },
    )

    workflow.add_edge("generate-filter", "talk-to-human")

    workflow.add_conditional_edges(
        "talk-to-human",
        should_continue,
        { 
            "work": "tools",
            "pass": "json-processing"
        }
    )

    workflow.add_conditional_edges(
        "tools",
        should_make_answer,
        {
            "add type": "add-filter",
            "make answer": "respond"
        }
    )

    workflow.add_edge("add-filter", "talk-to-human")
    workflow.add_edge("respond", "json-processing")
    workflow.add_edge("json-processing", END)

    graph = workflow.compile(
        checkpointer=checkpointer
    )

위의 모든 작업 함수들로 노드를 생성하고, 그래프의 진입 포인트, 엣지를 통하여 그래프의 작업 흐름을 생성한다.

compile 메서드로 생성된 graph가 바로 에이전트가 된다. Main 코드에서 생성된 에이전트는 사용자의 상호작용하며 기록된 대화 내용을 통해 사용자의 여행 유형을 답변에 반영하며, 사용자의 여행 계획이나 취향이 변화였을 때 즉각적으로 수정할 수 있었다.

7. 느낀 점 😶

조금 지난 코드인데 이 프로그램을 개발할 당시, 어떤 구조로 어떻게 개발해야 할지 까마득했던 기억이 난다. 현재, AI 에이전트의 라이브러리나 다양한 도구가 많이 개발된 것을 보아 새로운 문서와 기능들을 공부하여 더욱 보완된 에이전트를 개발하고 싶다.

새로운 기능

이전 게시물에서 보완해야 하는 부분에 사용자의 여행 취향을 반영할 수 있는 효율적인 방법이 필요하다고 얘기했었다. 대화를 시작하기 전 사용자의 여행 유형을 파악하고 그 유형을 기반으로 대화를 진행하면 사용자에게 더 잘 맞는 답변을 만들수 있지 않을까?

그래서 일명 TBTI (Travel MBTI)를 만들고 이 설문 조사를 바탕으로 사용자의 유형을 정한 후 가이드 봇은 이에 따라 달라지는 추가적인 질문을 통해 사용자의 여행 선호도를 분석한다.

여행 유형 분석 퀴즈

팀원들과 TBTI 테스트를 만들어 사용자의 여행 부캐를 찾을 수 있도록 하였다.

질문은 총 4가지로 구성된다.

1. 어떤 분위기의 여행 선호하는지?

2. 혼자 여행을 선호하는지 ?

3. 여행 시, 대중교통 vs 자동차 ?

4. 즉흥적인 여행 vs 계획적인 여행 ?

TBTI 결과

설문 조사가 끝나면 유형에 따른 귀여운 동물로 개인의 TBTI 캐릭터를 설명과 함께 얻게 된다.

사용자 유형 기반 채팅 시작

그림과 같이 채팅방 상단에 사용자의 유형을 볼 수 있으며 사용자와 채팅을 할 때, 가이드 봇은 유형에 따라 다른 질문들로 사용자의 선호 여행을 분석할 수 있도록 한다.

달라진 가이드 봇 작동 흐름

위의 흐름은 LangGraph의 기능을 통해 자동 생성된 그림이다. 전보다 살짝은 더 복잡해졌다. 위의 보라색 네모칸은 LangGraph에서 노드를 의미하고, 프로세스 진행 시 모든 노드에 공유하는 정보를 저장할 수 있다. 노드의 주요 정보는 사용자의 여행 유형(TBTI), 필터(filter)

필터(filter)는 쉽게 말하면 사용자의 여행 특성으로 여행 장소 데이터베이스 검색 시 검색 조건으로 이용된다.

노드 별로 진행 흐름을 설명해본다.

1. start-node : 사용자의 TBTI(여행 유형 특성)을 전달 받아 이 정보를 프로세스에 저장한다. 사용자의 필터 정보가 없을 경우, 필터를 생성하기 위해 generate-filter 노드로 이동한다. 그 외의 경우는 talk-to-human 노드로 이동하여 AI와 대화가 바로 진행 된다.
2. talk-to-human : 사용자의 질문에 대한 답변을 생성한다. 사용자의 대화 중 상세한 답변 생성을 데이터베이스 검색 등과 같은 추가 작업이 필요한 경우, 함수를 추가 호출한다. 함수가 호출될 경우, tools 노드로 이동한다.
3. add-filter : 대화 중 사용자의 여행 특성을 추가하는 역할을 한다. 예를 들어, 사용자가 대화 중 갑자기 아이와 여행하기로 여행 계획이 변경되었다고 말한다면, 아이와 함께 여행하기에 좋은 장소를 검색하기 위해 검색 조건을 필터 변수에 추가한다.
4. respond : 함수 호출 결과들을 바탕으로 답변을 생성한다.
5. json-processing : 사용자에게 답변을 보내기 위해 올바른 json 형태의 문자열을 딕셔너리 자료형으로 변환하는 단순 처리 과정이다. AI 답변을 정해진 형식으로 후처리 후 사용자에게 전달 된다.

메인 코드는 다음과 같고 다음 게시물에서 프로그램에 대한 자세할 설명이 이어진다.
LangGraph Agent 메인 코드

JSON 개념 정리와 자주 일어나는 JSONDecodeError에 대해

구조화된 출력(Structrued Outputs)?

우리가 사용하는 OpenAI 모델은 웹 서버와의 통신을 용이하게 하기 위해 구조화된 출력(Structured Outputs)을 사용해야 했다.

구조화된 출력(Structured Outputs)은 LLM의 답변을 지정된 JSON 형식에 맞게 생성시켜주는 기능이다. 아래의 형태처럼 일반적인 긴 줄글의 답변이 아닌 유효한 JSON 형식의 답변으로 모델이 생성할 수 있다.

{
  "steps": [
    {
      "explanation": "Subtract 31 from both sides to isolate the term with x.",
      "output": "8x + 31 - 31 = 2 - 31"
    },
    {
      "explanation": "This simplifies to 8x = -29.",
      "output": "8x = -29"
    },
    {
      "explanation": "Divide both sides by 8 to solve for x.",
      "output": "x = -29 / 8"
    }
  ],
  "final_answer": "x = -29 / 8"
}

OpenAI API에서 구조화된 출력은 gpt-4o로 시작되는 모델에서만 사용이 가능하다.

• o1-2024-12-17 이후 모델
• gpt-4o-mini-2024-07-18 이후 모델
• gpt-4o-2024-08-06 이후 모델

원래는 JSON 모드를 통해 답변을 출력하다가 더 정확하고 유효한 JSON 답변을 출력하기 위해서 구조화된 출력을 사용하기 시작했다.구조화된 출력 기능은 개발자가 제공한 스키마와 일치하도록 제한하고, 복잡한 스키마를 더 잘 이해하도록 모델을 훈련시킨 결과이다.

OpenAI가 구조화된 출력을 평가한 결과 gpt-4o-2024-08-06 모델에서 100%의 신뢰성을 가진 출력을 얻을 수 있었다고 한다.

OpenAI의 Structured Outputs에 대해서는 아래의 문서를 참고하면 더 구체적인 개념과 사용법을 알 수 있다.

• Introducing Structured Outputs in the API

• Structured Outputs

구조화된 출력 사용 예시

구조화된 출력을 사용하는 경우는 두가지 인데 하나는 함수 호출을 사용할 때와 JSON 스키마 응답 형식을 사용할 때이다. 나는 함수 호출은 다른 방식을 이용했고, 모델이 함수의 작동 결과를 참고하여 답변을 생성할 때 JSON 형식으로 생성하도록 사용하였다.

1. 출력 구조 정의

먼저 모델이 출력할 때 따라야 하는 JSON 스키마로서 데이터 구조를 정의해줘야 한다. 정의하는 방법은 SDK를 사용하여 Pydantic을 사용할 수 있지만 나는 사용자 질문에 따라 답변하는 JSON 값이 미묘하게 다르기 때문에 더 정확한 출력 유도를 위해서 JSON 스키마 여러개를 직접 정의하였다.

myFormat = {
    "type": "json_schema",
    "json_schema" : {
        "name" : "A_unique_answer",
        "schema" : {
            "type": "object",
            "properties": {
                "response" : {
                    "type": "string",
                    "description" : "put a short sentence that gives information about the places the user like"
                },
                "additionalInfo": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties" : {
                            "name" : {
                                "type": "string",
                                "description" : "The name of the place"
                            },
                            "explanation" : {
                                "type": "string",
                                "description" : "A brief description of the place."
                            }
                        },
                        "additionalProperties": False,
                        "required" : ["place_name", "explanation"]
                    }
                }
            },
            "additionalProperties": False,
            "required" : ["response", "additionalInfo"] 
        },
        "strict" : True
    }

구조화된 출력을 사용하기 위해서는 모든 필드를 required로 지정해야 하고 strict는 True로 설정하여 제공된 스키마에 출력을 일치하도록 맞춘다. JSON 스키마에 정의되지 않은 추가적인 키/값 생성을 막기 위해서 additionalProperties를 False로 설정하였다.

이렇게 스키마를 정의하고 myFormat 변수에 저장하였다.

2. 호출 API에 출력 구조 제공

messages = [
    {"role":"system", "content": "당신은 사용자의 취향을 파악하여 사용자가 좋아할 만한 장소를 추천해주는 봇입니다."},
    {"role": "user", "content":f"{reference}"}
]

response = client.chat.completions.create(
        model=model,
        messages=messages,
        response_format=myFormat
)

위에서 정의한 스키마를 OpenAPI를 호출할 때 response_format 파라미터에 넣어준다. 이렇게 넣어주면 내가 지정한 JSON 형식으로 답변을 생성한다. Pydantic 객체로 출력 구조를 정의했으면 client.chat.completions.parse() 메소드를 이용해 JSON 응답을 내가 만든 객체로 변환해 가져올 수 있다.

3. 생성된 출력 확인

answer = response.choices[0].message.content
print(answer)

출력)
'{
    "response" : "사용자가 좋아할 만한 장소를 알려드립니다.",
    "additionalInfo" :   
        [
            {
                 "name" : "하늘하늘 카페",
                "explanation" : "따뜻한 무드의 카페로 가족과 함께 방문하기 좋은 곳입니다."
            },
            {
                 "name" : "바다의 울림",
                "explanation" : "사용자는 귀여운 물품을 좋아하기 때문에 바다와 관련된 물품을 파는 곳에서 다양한 굿즈를 얻을 수 있습니다."
            },
            ...
        ]
 }'

위의 방법대로 답변을 확인해 봤을 때, 지정한 JSON 형식의 문자열로 답변이 잘 생성된 것을 알 수 있었다.

def change_to_dict(answer):
    try:
        response_dict = json.loads(answer, strict=False)
        return response_dict
    except Exception as e:
        print(f"Error escaping JSON strings: {e}")
        return response

생성한 JSON 문자열을 json.loads() 함수를 이용해 파이썬 딕셔너리로 변환할 수 있다. 이때, 유효하지 않은 JSON 형식의 데이터일 경우 오류가 발생하는데 Structured Outputs로 생성한 답변은 오류가 발생하지 않았다.

JSON 개념 정리

JSON은?

• 데이터를 교환하기 위해 경량화된 형식
• 인간이 읽고 쓰기에 용이하고 기계가 파싱하고 생성하기도 간편한 형식
• 웹 어플리케이션 데이터 교환에 많이 사용

기본 JSON 문법

{
    "first_name" : "Katie",
    "last_name" : "Rodgers"
}

• 키-값의 쌍으로 데이터가 쓰인다.
• 여러 쌍의 키-값의 모음은 객체라고 한다. 객체는 중괄호 안에 있다.

직렬화(serialization)

import json

data = {"response" : "hi", "info": None}

json_object = json.dumps(data)

print(type(json_object))
print(json_object)

# 출력
# <class 'str'>
# {"response": "hi", "info": null}

파이썬에서json.dumps() 함수를 이용해 파이썬 객체를 JSON 형식의 문자열로 변환할 수 있고 이를 직렬화(serialization)라고 한다. 변환된 데이터는 파일에 저장하거나 네트워크를 통해 전송할 수 있다.

파이썬의 None 값은 JSON으로 변환되면 null로 바뀌고 JSON 모듈은 항상 바이트 객체가 아닌 문자열 객체를 생성한다는 것을 참고하길 바란다.

json.dump() 함수는 파이썬 객체를 JSON 형식 스트림(파일류 객체)로 직렬화 한다는 점에서 차이가 있다.

역직렬화(deserialization)

import json

string = '{"response" : "hi", "info": null}'

json_object = json.loads(string)

print(type(json_object))
print(json_object)

#출력
#<class 'dict'>
#{'response': 'hi', 'info': None}

json.loads() 함수를 통해 JSON 형식의 문자열이나 바이트 배열 객체를 파이썬 객체로 변환할 수 있다. 이를 역직렬화(deserialization)라고 한다.

위의 경우는 JSON 형식의 문자열이 파이썬 객체인 딕셔너리로 변환된 모습이다. 역직렬화 되는 데이터가 유효한 JSON 문자열이 아닌 경우 JSONDecodeError를 발생시키는 데 몇가지 오류는 아래에서 살펴 본다.

json.load()는 JSON 형식을 포함하는 텍스트나 바이너리 파일을 파이썬 객체로 역직렬화 하는 함수이다.

JSONDecodeError: 변환할 때 조심해야 하는 것

1. JSON의 키-값은 모두 이중 따옴표(”)로 둘러 쌓여 있어야 함

import json

# 옳게된 경우: '{"response" : "hi", "info": null}'
string = "{'response' : 'hi', 'info': null}"

# json.loads() 부분 오류 발생!!
json_object = json.loads(string)

• 발생한 오류: json.decoder.JSONDecodeError: Expecting property name enclosed in double quotes: line 1 column 2 (char 1)

*2. JSON 값 안에서 이중 따옴표를 사용할 수 없음 - 그냥 따옴표는 사용 *가능

import json

# 옳게된 경우: '{"response" : "안녕 너 \'김치\' 먹어 봤어? ", "info": null}'
string = '{"response" : "안녕 너 "김치" 먹어 봤어? ", "info": null}'

# json.loads() 부분 오류 발생!!
json_object = json.loads(string)

• 발생한 오류: json.decoder.JSONDecodeError: Expecting ',' delimiter: line 1 column 22 (char 21)

3. JSON 값 내에 제어 문자 포함될 수 없음

import json

string = '{"response" : "안녕 너 \'김치\' 먹어 봤어? \n", "info": null}'

# json.loads() 부분 오류 발생!!
json_object = json.loads(string)
print(json_object)

0-31 범위의 문자 코드인 \t \n \0 \r 같은 제어 문자가 문자열 내부에 허용될 수 없다.

• 발생한 오류: json.decoder.JSONDecodeError: Invalid control character at: line 1 column 33 (char 32)

• 해결 방법 json.loads() 함수에 strict 파라미터를 False로 설정하면 JSON 문자열 내에 제어 문자가 허용된다.

    json_object = json.loads(string, strict=False)
    print(json_object)
  
    # 출력
    # {'response': "안녕 너 '김치' 먹어 봤어? \n", 'info': None}

파이썬 공식 문서

( langGraph 처음 사용했기 때문에 잘못된 부분이 있을 수 있습니다 🥲)

여행 가이드봇 전체 코드

여행 가이드봇 작동 흐름

흐름을 아래에서 다시 확인해보자 현재 흐름은 단순하다.

1. 언어 모델이 로드되어 있는 agent 노드에 사용자 메시지 전달

2. 사용자 메시지가 호출할 수 있는 함수와 관련이 있는지 판단 후 다음 단계 진입, 관련이 있는 경우 AI가 tool_calls 변수를 가진 메시지 생성

3. 호출 가능한 함수가 존재한다면, tools 노드로 진입 후 작동, 이때 질문과 관련된 정보 검색도 호출된 함수 안에서 실행 호출 가능한 함수가 없다면, AI의 일반적인 답변을 생성한 후 다음 노드로 전달

4. 호출 결과는 respond 노드로 전달되어 호출된 함수에 따라 다른 지시사항과 함께 적용되어 답변 생성 (답변은 JSON 형태로 생성)

5. 생성된 답변이 지정된 JSON 형식으로 잘 생성되고, 올바른 JSON인지 확인하는 단계를 json-processing 노드에서 진행 후 답변을 사용자에게 전달한다.

1. Main 진행 코드

import os
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, List, Dict
from langgraph.checkpoint.memory import MemorySaver

from openAI_api import llm
from access_milvusDB import database
from available_functions import callable_tools
from agent_executor import create_my_agent

# langsmith, langchain 환경 설정
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "api key"
os.environ["LANGCHAIN_PROJECT"] = "test name"


class QuestionRequest(BaseModel):
    question: str


class AiResponse(BaseModel):
    answer: str
    place: Optional[List[Dict]] = None


app = FastAPI()

# 대화 기록 메모리 생성
memory = MemorySaver()

# 호출할 함수 리스트 가져오기
tools = callable_tools

# 에이전트 생성 
agent = create_my_agent(
    model=llm,
    tools=tools,
    checkpointer=memory
)

config = {"configurable": {"thread_id": "test-thread1"}}    

db = database

@app.post("/ask-ai/", response_model=AiResponse)
async def ask_ai(request: QuestionRequest):
    question = request.question  # JSON에서 question 필드 추출

    try:
        db.reconnect()

        system_prompt = """
        - You are a tour guide called 'TBTI'. Ask the user a short and clear question.
        - Just ask once what kind of trip the user wants.
          ex. Is there anything you want when you travel?
        - Only up to five locations will be notified.
        """

        messages_list = [("system", f"{system_prompt}")] 
        messages_list.append(("human", f"{question}"))

        # 에이전트 실행
        response = agent.invoke({"messages": messages_list}, config)['final_response']
        #print(response)

        return response

    except Exception as e:
        print("에러 발생: ", e)
        raise HTTPException(status_code=500, detail="AI 처리 중 오류 발생")
    finally:
        db.unconnect()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8991)

사용자의 질문은 fastAPI 이용해서 받아오며 나중에 답변도 이를 통해 전달한다. 위의 코드에서 중요한 부분은 에이전트 생성과 실행 부분이다.

에이전트는 create_my_agent 함수를 통해 생성하는데 이는 LangGraph의 create_react_agent 코드와 아래의 문서를 참고해 구현하였다.

에이전트를 생성하면 invoke()에 사용자 질문과 메모리 설정을 담아 실행시킨다.

• LangGraph Prebuit Components

• How to return structured output with a ReAct style agent

2. agent 구현 코드

전체 코드는 다음과 같다. create_my_agent 안에 흐름의 각 단계를 의미하는 함수들이 정의되어 있고 마지막 부분엔 이 함수들을 이용한 노드와 노드를 연결하는 edges를 구성하여 흐름을 만들었다.

import json
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.tools import BaseTool
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.graph.graph import CompiledGraph
from langgraph.prebuilt.tool_executor import ToolExecutor
from langgraph.graph.message import add_messages
from langchain_core.language_models import LanguageModelLike

from criteria_of_answers import system_informations_of_functions
from openAI_api import chat_completion_request

from typing import (
    Optional, 
    TypedDict,
    Annotated, 
    Union, 
    Sequence
) 


def escape_json_strings(response):
    try:
        # JSON 문자열을 딕셔너리 자료형으로 변환
        response_dict = json.loads(response)
        return response_dict
    except Exception as e:
        print(f"Error escaping JSON strings: {e}")
        return response


# 노드에 전달되는 state
class AgentState(TypedDict):
    previous_result : str # 이전 단계에서의 결과값 저장
    final_response : dict # 사용자에게 전달되는 최종 메시지
    messages : Annotated[list, add_messages]  # 대화 history 전달


def create_my_agent(
    model: LanguageModelLike,
    tools: Union[ToolExecutor, Sequence[BaseTool], ToolNode],
    checkpointer: Optional[BaseCheckpointSaver] = None
) -> CompiledGraph:

    # LLM 시스템 프롬프트 리스트 로드 - 작동되는 함수에 따라 시스템 프롬프트 내용 다름
    configuration_for_answers = system_informations_of_functions

    # 함수 호출 도구 사용할 수 있는 모델 생성
    model_with_tools = model.bind_tools(tools)

    # 사용할 AI 모델 로드 및 AI 답변 처리
    def call_model(state: AgentState):
        response = model_with_tools.invoke(state['messages'])
        last_response = response.content.strip("<>() ").replace('\"', '\'')
        last_response = f'{{\"answer\": \"{last_response}\", \"place\": null}}'

        # AI 답변을 json 형식의 문자열로 만들어 previous_result에 저장 / 답변 history에 저장 
        return {"previous_result" : last_response , "messages" : [response]}

    # 도구 작동 후 함수 결과 LLM에게 최종 전달 후 답변 생성
    def respond_after_calling_tools(state: AgentState):
        # 작동된 마지막 도구 메시지 가져오기
        messages = state["messages"]
        last_tool_message = messages[-1]

        # 작동된 함수 이름 가져오기
        name_of_functions_called = last_tool_message.name
        print(name_of_functions_called)

        # 함수 리턴값 가져오기 = 검색 결과 가져오기
        reference = last_tool_message.content

        # 작동된 도구에 맞는 시스템 프롬프트 가져오기
        system_prompt = configuration_for_answers[name_of_functions_called]['system_prompt']
        response_format = configuration_for_answers[name_of_functions_called]['response_format']

        # 결과 참고해서 LLM 답변 생성
        messages = [
            {"role":"system", "content": f"{system_prompt}"},
            {"role": "user", "content":f"{reference}"}
        ]

        llm_response = chat_completion_request(
            messages=messages,
            response_format=response_format
        ).choices[0].message.content   

        return {"previous_result": llm_response}

    # Define the function that determines whether to continue or not
    def should_continue(state: AgentState):
        messages = state["messages"]
        last_message = messages[-1]

        # 함수 호출이 없으면 바로 사용자에게 리턴
        if not last_message.tool_calls:
            return "pass"
        # 있으면 워크플로우 지속
        else:
            return "work"

    def post_processing_of_answer(state: AgentState):
        ai_answer = state["previous_result"]
        escaped_response = escape_json_strings(ai_answer)
        return {"final_response" : escaped_response}


    # 새로운 그래프 정의
    workflow = StateGraph(AgentState)

    # agent, tools 노드 생성
    workflow.add_node("agent", call_model)
    workflow.add_node("tools", ToolNode(tools))
    workflow.add_node("respond", respond_after_calling_tools)
    workflow.add_node("json-processing", post_processing_of_answer)

    # Set the entrypoint as `agent`
    # This means that this node is the first one called
    workflow.set_entry_point("agent")

    # We now add a conditional edge
    workflow.add_conditional_edges(
        "agent",
        should_continue,
        {
            "work": "tools",
            "pass": "json-processing",
        },
    )

    workflow.add_edge("tools", "respond")
    workflow.add_edge("respond", "json-processing")
    workflow.add_edge("json-processing", END)

    graph = workflow.compile(
        checkpointer=checkpointer
    )

    return graph

하나씩 뽑아서 구체적으로 내용을 확인해 보겠다.

2-1. State 설정

# 노드에 전달되는 state
class AgentState(TypedDict):
    previous_result : str # 이전 단계에서의 결과값 저장
    final_response : dict # 사용자에게 전달되는 최종 메시지
    messages : Annotated[list, add_messages]  # 대화 history 전달

각 노드들이 공유할 수 있는 데이터 구조라고 볼 수 있다. 노드에서 위에 설정된 변수들에 접근하여 이를 사용하거나 저장할 수 있다.

나는 주석에 달린 것처럼 이전 단계에서의 결과값, 마지막 전달되는 최종 메시지, 대화 기록을 저장하여 사용하였다.

2-2. 실행할 함수(작업) 정의

-------------------------------- 작업 1 ---------------------------------

   # 함수 호출 도구 사용할 수 있는 모델 생성
    model_with_tools = model.bind_tools(tools)

    # 사용할 AI 모델 로드 및 AI 답변 처리
    def call_model(state: AgentState):
        response = model_with_tools.invoke(state['messages'])
        last_response = response.content.strip("<>() ").replace('\"', '\'')
        last_response = f'{{\"answer\": \"{last_response}\", \"place\": null}}'

        return {"previous_result" : last_response , "messages" : [response]}

bind_tools 함수를 이용해 호출 가능한 도구 정보들을 모델이 담고 이를 사용할 것이다. 작업 내용을 짧게 정리해보겠다.

• call_model : 도구를 호출할 수 있는 모델을 가져와 사용자의 메시지를 모델에 전달해 답변을 생성한다. 답변의 내용만을 추출해 previous_results 변수에 딕셔너리 형태의 문자열로 저장하고 messages 변수에는 대화 기록을 저장한다.

-------------------------------- 작업 2 ---------------------------------

    # 도구 작동 후 함수 결과 LLM에게 최종 전달 후 답변 생성
    def respond_after_calling_tools(state: AgentState):
        # 작동된 마지막 도구 메시지 가져오기
        messages = state["messages"]
        last_tool_message = messages[-1]

        # 작동된 함수 이름 가져오기
        name_of_functions_called = last_tool_message.name
        print(name_of_functions_called)

        # 함수 리턴값 가져오기 = 검색 결과 가져오기
        reference = last_tool_message.content

        # 작동된 도구에 맞는 시스템 프롬프트 가져오기
        system_prompt = configuration_for_answers[name_of_functions_called]['system_prompt']
        response_format = configuration_for_answers[name_of_functions_called]['response_format']

        # 결과 참고해서 LLM 답변 생성
        messages = [
            {"role":"system", "content": f"{system_prompt}"},
            {"role": "user", "content":f"{reference}"}
        ]

        llm_response = chat_completion_request(
            messages=messages,
            response_format=response_format
        ).choices[0].message.content   

        return {"previous_result": llm_response}

• respond_after_calling_tools : 호출된 도구의 작동 결과와 지정 프롬프트를 이용해 답변을 생성하는 작업이다. 작동 결과를 대화 기록에서 추출하고 작동된 도구 이름에 따라 각각 다른 시스템 프롬프트 내용을 가져온다.

이 함수에선 내부에 도구 호출 가능한 모델이 아닌 일반 LLM을 사용하여 답변을 생성하는데 아래의 json 스키마 같이 지정된 json 형태로 답변을 생성하도록 했다. 답변 json 형태도 호출되는 함수에 따라 다르다.

생성한 답변은 마지막에 previous_result 변수에 저장해 이전 단계 결과값으로 저장한다.

  response_format_1 = {
      "type": "json_schema",
      "json_schema" : {
          "name" : "A_general_answer",
          "schema" : {
              "type": "object",
              "properties" : {
                  "answer": {
                      "type": "string",
                      "description" : "put your answer in the value."
                  },
                  "place": {
                      "type": ["null", "object"],
                      "description": "This will be null."
                  }
              },
              "required" : ["answer", "place"],
              "additionalProperties": False
          },
          "strict" : True
      }
  }

-------------------------------- 작업 3,4 -------------------------------

    def should_continue(state: AgentState):
        messages = state["messages"]
        last_message = messages[-1]

        # 함수 호출이 없으면 바로 사용자에게 리턴
        if not last_message.tool_calls:
            return "pass"
        # 있으면 워크플로우 지속
        else:
            return "work"

    def post_processing_of_answer(state: AgentState):
        ai_answer = state["previous_result"]
        escaped_response = escape_json_strings(ai_answer)
        return {"final_response" : escaped_response}

• should_continue : 사용자의 질문이 도구 호출이 가능한 지 판단하는 작업 함수 호출이 필요하면 work 과정으로 진행, 없으면 pass 과정으로 진행

• post_processing_of_answer : 이전 단계에서 저장된 LLM의 답변을 다시 가져와 JSON 형식이 아닌 문자열이 있는지 검토하고 이를 파이썬 딕셔너리의 형태로 변환해 저장한다.

2-3. 그래프 생성

    # 새로운 그래프 정의
    workflow = StateGraph(AgentState)

    # agent, tools 노드 생성
    workflow.add_node("agent", call_model)
    workflow.add_node("tools", ToolNode(tools))
    workflow.add_node("respond", respond_after_calling_tools)
    workflow.add_node("json-processing", post_processing_of_answer)

    # Set the entrypoint as `agent`
    # This means that this node is the first one called
    workflow.set_entry_point("agent")

    # We now add a conditional edge
    workflow.add_conditional_edges(
        "agent",
        should_continue,
        {
            "work": "tools",
            "pass": "json-processing",
        },
    )

    workflow.add_edge("tools", "respond")
    workflow.add_edge("respond", "json-processing")
    workflow.add_edge("json-processing", END)

    graph = workflow.compile(
        checkpointer=checkpointer
    )

    return graph

이제 그래프를 정의해 에이전트 흐름을 구현해 본다.

StateGraph 객체를 통해 '상태'를 가지는 챗봇의 구조를 정의했다고 할 수 있다. 그 다음으로 위에서 정의한 작업 내용들을 노드로 구성하고, 호출 가능한 도구들도 ToolNode로 정의하여 하나의 노드로 만든다.

노드 생성 후에는 시작 노드를 설정하고 Edge를 통해 노드들을 연결한다. 이는 노드들이 통신하는 지점을 만드는 것이며 Edge는 진행 로직을 만들고 라우팅 되거나 중지되는 부분도 만들 수 있다.

마지막에 구현한 흐름을 compile()을 통해 실행 가능한 에이전트를 생성한다.

3. 호출 도구 정의 코드

import json
from langchain_core.tools import tool
from access_milvusDB import database
from openAI_api import embedding

@tool
def recommand_travel_destination(question : str, location : str, area : str) -> str:
    """
    recommand the various places that user wants to know or to travel
    It only works when user wants to know the various places.
    It doesn't work when user told to plan the trip and when user told to reserve the place.

    Args:
        question: Identify the travel the user want and input the questions.
        location: input the area of Korea to travel, e.g. 서울 or 부산 or 대구 or 강원도
        area: Enter only the following words to indicate where the place in the user's question belongs to the following Korean administrative districts. e.g. 강원특별자치도 
              - 한국 행정 구역 : 서울특별시, 부산광역시, 인천광역시, 대구광역시, 대전광역시, 광주광역시, 울산광역시, 세종특별자치시, 경기도, 충청북도, 충청남도, 전라남도, 경상북도, 경상남도, 강원특별자치도, 전북특별자치도, 제주특별자치도
    """

    milvus = database

    # 사용자 질문 벡터화
    vector  =  embedding(question)

    # 필터링 생성 후 테이블 검색 진행
    filtering = f"area_name == '{area}'" 
    results_localCreator, results_nowLocal = milvus.search_all_tables(embedding=vector, filtering=filtering)

    # 쿼리 결과 합치기
    total_results = milvus.get_formatted_results(results_localCreator, results_nowLocal)

    return f"user question: {question} \n\nreference: \n{total_results}"


@tool
def search_specific_place(question : str, place_name : str = None) -> json:
    """
    give the information of the specific places mentioned by the user.
    It works when a user question contains a name of specific place.
    It doesn't work when you recommand a place.

    Args:
        question: input the user's question as it is
        place_name: The name of the particular place that user wants to know or to reserve
    """

    # 특정 장소 검색 과정 진행....

    return f"user question: {question} \n\nreference: \n{total_results}"



callable_tools = [recommand_travel_destination, create_travel_plan, search_specific_place]

여행지 추천, 여행 계획 생성, 특정 장소 검색 기능을 도구들로 정의하여 모델이 사용자 질문의 의도를 파악해 그에 맞는 기능을 사용할 수 있도록 한다. 위의 코드는 중간 생략해서 전체 코드는 아니다.

호출할 도구는 위와 같이 정의하면 되는데 @tool 어노테이션을 달아 호출할 도구라는 것을 표시하고, 함수 내부의 주석으로 어떤 기능이며 언제 작동하고, 입력될 파라미터 값의 설명을 포함하여 LLM이 도구를 이해하고 사용할 수 있도록 만든다.

호출할 도구들은 정의 후 callable_tools안에 넣어 리스트로 저장한다.

코드 진행 예시

사용자 질문에서 LLM의 답변까지 어떻게 진행되는 건지 예시를 들어 정리해보도록 하겠다.

🙎 사용자: 강릉으로 여행 갈 건데 갈만한 곳 알려줘.

1. Main 작동 - http://0.0.0.0:8991/ask-ai 경로로 사용자 질문 post 요청 - 전달 받은 JSON 에서 사용자 질문을 추출 후 생성한 에이전트에 전달

2. 에이전트 작동

• agent node : 에이전트는 전달 받은 질문이 여행지 추천 함수인 recommand_travel_destination와 관련있다고 판단

• tools node : 여행지 추천 함수 호출하여 작동, 강릉의 여행지를 검색하여 사용자에게 맞는 여행지 리스트를 질문과 함께 리턴

• respond node : 함수 리턴 결과와 여행지 추천 시 답변 생성 기준을 가져와 LLM 답변 생성

• json-processing : LLM이 생성한 JSON 형태의 문자열이 올바른 JSON 형식인지 확인 후 딕셔너리로 변환 딕셔너리 객체를 최종 전달 메시지로 저장 후 사용자에게 전달

LangGraph 공식 문서

저번 게시물에서 프로그램에 대해 간단하게 소개하며 개발하면서 까다로웠던 점들을 이야기했다. 다시 한번 그 부분을 살펴보며 LangChain과 LangGraph 라이브러리로 어떻게 구현할 예정인지 살펴보겠다.

( 아직 개발 능력이 부족해 잘못 판단한 부분이 있을 수 있습니다 🥲)

개발하면서 까다로웠던 부분

*프로그램에 사용되는 LLM은.. *

1. 이전 대화 내용을 기억할 수 있어야 한다. 자연어 처리 모델를 이용해 대화를 하다 보면 대화 내용이 쌓이게 되는데, 앞선 대화 속에 사용자의 요구사항을 모델이 기억하고 알맞은 답변을 하기 위해서는 대화 내용을 기억할 수 있어야 한다.

👉 LangChain과 LangGraph의 memory를 사용할 수 있다.

2. 도구를 호출할 수 있어야 한다. 사용자의 다양한 요구에 맞는 능동적인 답변을 생성하기 위해서 사용자의 질문과 관련된 함수를 작동시킬 수 있도록 구성하였다. 따라서 위에서 설명한 여행지 추천, 여행 계획 생성, 특정 장소 검색 기능은 각각 다른 함수들로 구성되어 있고, 사용자 질문의 의도를 파악한 후 모델이 관련 함수를 호출한 후 답변을 생성한다.

EX. 사용자 질문: 강릉에서 아이와 함께 갈 수 있는 장소 알려줄래?
    -> 모델은 여행지 추천 함수를 작동 후 작동 결과를 다음 단계로 전달

👉 LangChain의 AgentExecutor를 통해 가능 👉 LangGraph의 사전 구축된 create_react_agent 를 통해 가능

3. 항상 JSON 형식으로 답변을 생성해야 한다. LLM의 답변을 JSON 형식으로 만들어 웹 서버로 전달해야 하며 FastAPI를 사용한다. 웹 서버에서는 JSON을 받아 필요한 값을 추출해 질문에 가장 효과적인 방식으로 답변을 재구성한다. JSON의 key는 answer와 place가 존재한다.

👉 LangChain with_structured_output() 메서드를 이용한 모델의 답변을 구조적 데이터로 반환 가능 👉 JSON 스키마 설정을 출력 형태로 전달해 구조적 출력이 가능한 모델이 있음 (EX. OpenAI)

4. 상황에 따라 생성해야 하는 JSON 값이 다르다. 간단하게 예를 들어 설명하면,

1. 여행지 추천의 경우 장소의 그림과 정보를 전달해 카드 형식으로 사용자에게 보여줘야 하기 때문에 다음과 같이 구성된다. answer 값으로 짧은 추천의 말과, place에는 장소의 정보가 전달된다.

    {
        "answer" : "AI의 짧은 추천의 말",
        "place" :   
            [
                {
                     "place_name" : "장소 이름",
                    "description" : "장소 설명",
                    "redirection_url" : "장소의 상세페이지 이동 경로"
                },
                ...
            ]
    }

2. 여행 계획 생성의 경우 사용자에게 전달되는 답변은 긴 문자열이기 때문에 여행 계획의 긴 내용이 answer 키의 값으로 들어가고, place 값은 null로 된다.

    {
        "answer" : "1박 2일의 긴 여행 계획 내용",
        "place" : null
    }

👉 Pydantic Class , TypedDict, Json-Schema 등의 방식으로 원하는 형태의 답변 출력을 의도할 수 있음

어떻게 구현해볼까?

1. LangChain과 LangGraph의 사전 구축 Agent 사용

import getpass
import os

from typing import Optional
from langgraph.prebuilt import create_react_agent
from langchain_core.pydantic_v1 import BaseModel, Field
from available_functions import callable_tools


os.environ["OPENAI_API_KEY"] = getpass.getpass()

from langchain_openai import ChatOpenAI

# Pydantic
class Joke(BaseModel):
    """Joke to tell user."""

    setup: str = Field(description="The setup of the joke")
    punchline: str = Field(description="The punchline to the joke")
    rating: Optional[int] = Field(
        default=None, description="How funny the joke is, from 1 to 10"
    )

llm = ChatOpenAI(model="gpt-4o-mini")
structured_llm = llm.with_structured_output(Joke)

# 호출할 함수 리스트 가져오기
tools_of_list = callable_tools

agent = create_react_agent(
    model=structured_llm,
    tools=tools_of_list
)

response = agent.invoke({"messages": [("human", "안녕?")]})
print(response["messages"])

간단하게 코드를 짜서 가능성만 확인해본다. create_react_agent 메서드는 Langgraph의 사전 구축 Agent로 이를 통해 쉽게 도구를 호출하고 작동시킬 수 있다. 또한 LLM의 답변이 JSON 형식으로 나오도록 with_structured_output()를 사용하고 싶었다.

하지만 이 코드는 오류가 발생해 작동될 수 없다. 구조적 출력이 가능한 모델을 create_react_agent의 model 파라미터로 전달할 수 없기 때문이다. 오류 메시지는 다음과 같다.

** 무슨 오류인가?**

AttributeError: 'RunnableSequence' object has no attribute 'bind_tools 

Create_react_agent는 내부적으로 모델의 bind_tools 메서드를 호출한다. with_structured_output()의 리턴값은 RunnableSequence 객체이기 때문에 이 객체는 bind_tools 메서드를 작동시킬 수 없기 때문에 일어나는 오류이다.

LangChain의 기존 AgentExecutor를 사용해도 같은 문제가 일어나게 되었다.

create_react_agent의 결과를 이용한 후처리 단계를 chain에 달면 되지만 우리 프로그램의 복잡한 로직을 구현하기에 한계가 있다고 생각했다.

2. LangGraph를 통해 나만의 Agent를 만들어보자

LangChain의 AgentExecutor와 사전 구축되어 제공되는 agent에 한계를 느낄 때쯤, 구조적 출력이 가능한 agent 생성 글을 보게 되었다.

How to return structured output with a ReAct style agent

LangChain은 보통 도구 호출 후 결과가 LLM에게 다시 반환되어 이 결과를 참고하여 답변을 생성한다. 이 과정이 일정하게 묶여있어 도구 호출 결과에 따라 다른 후처리 과정을 구현하기가 어렵다고 생각되었다.

LangGraph를 사용하면 도구 호출 결과에 따라 다른 과정을 구현하고 검토하기 편리하며, 에이전트의 흐름을 직접 지정하고 제어할 수 있다는 점에서 유연하다고 판단했다.

또한 좋은 점은 프로그램 각 단계의 상태를 저장하고 확인할 수 있기 때문에 데이터의 일관성과 올바른 실행 순서를 보장할 수 있다는 점이다.

나만의 Agent는?

현재 흐름은 단순하다.

1. 언어 모델이 로드되어 있는 agent 노드에 사용자 메시지 전달

2. 사용자 메시지가 호출할 수 있는 함수와 관련이 있는지 판단 후 다음 단계 진입, 관련이 있는 경우 AI가 tool_calls 변수를 가진 메시지 생성

3. 호출 가능한 함수가 존재한다면, tools 노드로 진입 후 작동, 이때 질문과 관련된 정보 검색도 호출된 함수 안에서 실행 호출 가능한 함수가 없다면, AI의 일반적인 답변을 생성한 후 다음 노드로 전달

4. 호출 결과는 respond 노드로 전달되어 호출된 함수에 따라 다른 지시사항과 함께 적용되어 답변 생성 (답변은 JSON 형태로 생성)

5. 생성된 답변이 지정된 JSON 형식으로 잘 생성되고, 올바른 JSON인지 확인하는 단계를 json-processing 노드에서 진행 후 답변을 사용자에게 전달한다.

이제 이러한 흐름의 자세한 코드 내용은 다음 게시물에서 이어진다.

이 챗봇은 로컬 여행 관련 웹 메인 페이지에 위치해 Java Spring 웹서버와 FastAPI를 활용해 통신한다.

구현 프로그램 내용

로컬 여행 챗봇으로 사용자에게 필요한 정보를 친숙한 대화 형식의 방식으로 지원해준다. 이 챗봇이 가진 기능은 크게 세가지이다.

프로그램 기능

1. 여행지 추천 사용자가 여행하고자 하는 지역을 파악한 후 사용자 질문과 관련된 장소를 최대 5곳 추천해준다. 아래의 그림 처럼 장소명, 설명, 장소의 그림과 함께 사용자에게 장소를 추천한다.

2. 여행 계획 생성 여행 지역, 여행 기간을 파악한 후 사용자가 원하는 장소들로 구성된 계획을 생성해준다.

   • 여행 계획 생성 조건
   1. 하루에 방문할 여행지의 카테고리는 모두 달라야 한다.
   2. 마지막 날에 숙박 업소는 추천하지 않는다.
   3. 여행 계획에 포함된 각 장소들의 거리가 10km 이내가 되도록 구성한다.
   4. 추천해주는 각 장소는 장소의 이름, 설명, 위치를 제공한다.

3. 특정 장소 정보 제공 사용자가 특정 장소의 정보를 원할 시, 특정 여행지 이름을 파악하여 관련 장소의 카드를 전달한다. 이 카드를 누르면 장소의 상세정보를 보여주는 페이지로 이동할 수 있다.

프로그램 시스템 구성

• 활용 라이브러리: LangChain, LangGraph
• 주요 적용 기술: RAG (Retrieval-Argumented Generation)
• 사용된 모델: GPT-4o-mini (OpenAI), BGE-M3-Korean (Transformers)

시스템 흐름 설명

• 챗봇이 작동되면 사용자의 질문이 LangGraph Agent에 전달된다. 질문의 의도를 파악하여 질문과 관련된 외부 도구(= 함수)를 호출할 수 있다.

• 이 도구가 호출될 경우, 사용자 질문과 관련된 함수가 작동되며 질문을 벡터화 시킨 후 관련된 자료를 검색한다. ( RAG 로직이 여기에 이 부분에서 쓰인다.)

• 호출된 함수에 지정된 지시사항과 검색 참고 자료를 이용해 자연어 처리 모델이 답변을 생성하고, 알맞은 JSON 형식으로 답변을 후처리 후 사용자에게 전달
  
  

프로그램 특장점

1. 로컬 크리에이터 기반의 특수 데이터 제공 로컬 크리에이터는 지역의 자산을 활용해 사업을 하고 계신 분들을 의미하며, 정부에서 지역 활성화 프로젝트로 이들을 지원하기도 한다. 마케팅 역량이 부족한 로컬 크리에이터 위주로 장소를 소개하여 일반 여행자가 알아내기 힘든 장소들 위주로 정보를 제공한다.

2. 여행 가이드봇 제공

   챗봇을 통해 정보 접근 편의성을 증대시키며 개인의 관심사와 취향을 실시간으로 반영해 여행 정보를 제공 받을 수 있다. 설문 형태의 데이터 군집화 결과가 아닌 사용자와 소통하는 간단한 방식으로 장소를 추천한다.

3. 장소의 구체적인 정보 체크 가능

   강원도 로컬 크리에이터 위주로 설문조사를 진행하여 장소의 예약 가능 여부, 반려동물 동반 가능, 분위기, 휠체어 접근 가능 여부 등 디테일한 정보들을 제공함.

   다른 여행 사이트에서 제공하지 않는 정보들을 제공하며 챗봇과의 대화와 메인페이지의 장소 검색 기능을 통해 이러한 정보를 손쉽게 얻을 수 있다.

개발하면서 까다로웠던 부분

왜 LangGraph 라이브러리를 사용했는지에 대한 답변도 포함한다.

*프로그램에 사용되는 LLM은.. *

1. 이전 대화 내용을 기억할 수 있어야 한다. 자연어 처리 모델를 이용해 대화를 하다 보면 대화 내용이 쌓이게 되는데, 앞선 대화 속에 사용자의 요구사항을 모델이 기억하고 알맞은 답변을 하기 위해서는 대화 내용을 기억할 수 있어야 한다.

2. 도구를 호출할 수 있어야 한다. 사용자의 다양한 요구에 맞는 능동적인 답변을 생성하기 위해서 사용자의 질문과 관련된 함수를 작동시킬 수 있도록 구성하였다. 따라서 위에서 설명한 여행지 추천, 여행 계획 생성, 특정 장소 검색 기능은 각각 다른 함수들로 구성되어 있고, 사용자 질문의 의도를 파악한 후 모델이 관련 함수를 호출한 후 답변을 생성한다.

EX. 사용자 질문: 강릉에서 아이와 함께 갈 수 있는 장소 알려줄래?
    -> 모델은 여행지 추천 함수를 작동 후 작동 결과를 다음 단계로 전달

3. 항상 JSON 형식으로 답변을 생성해야 한다. LLM의 답변을 JSON 형식으로 만들어 웹 서버로 전달해야 하며 FastAPI를 사용한다. 웹 서버에서는 JSON을 받아 필요한 값을 추출해 질문에 가장 효과적인 방식으로 답변을 재구성한다. JSON의 key는 answer와 place가 존재한다.

4. 상황에 따라 생성해야 하는 JSON 값이 다르다. 간단하게 예를 들어 설명하면,

1. 여행지 추천의 경우 장소의 그림과 정보를 전달해 카드 형식으로 사용자에게 보여줘야 하기 때문에 다음과 같이 구성된다. answer 값으로 짧은 추천의 말과, place에는 장소의 정보가 전달된다.

    {
        "answer" : "AI의 짧은 추천의 말",
        "place" :   
            [
                {
                     "place_name" : "장소 이름",
                    "description" : "장소 설명",
                    "redirection_url" : "장소의 상세페이지 이동 경로"
                },
                ...
            ]
    }

2. 여행 계획 생성의 경우 사용자에게 전달되는 답변은 긴 문자열이기 때문에 여행 계획의 긴 내용이 answer 키의 값으로 들어가고, place 값은 null로 된다.

    {
        "answer" : "1박 2일의 긴 여행 계획 내용",
        "place" : null
    }

보완해야 하는 부분

1. 웹 메인 페이지에서 여러 사용자들이 챗봇을 각각 이용할 수 있어야 한다. 현재 병렬적으로 작동하지 않아 한 사용자가 답변을 받은 후 다른 사용자가 이용할 수 있다.

2. 사용자의 여행 취향을 파악할 수 있는 조금 더 효율적인 방법이 필요하다. 사용자와의 대화를 통해 취향에 맞는 장소를 대부분 알려주지만 안정성과 속도 측면에서 좋은 편은 아니라고 판단된다. (가끔 취향과 맞지 않는 장소를 추천해줄 때도 있다) 대화를 시작하기 전 사용자의 여행 유형을 전달하여 이를 기반으로 답변을 생성하는 방식으로 재구성해 보고 싶다.

3. Self-RAG / Corrective-RAG 방식을 적용해 답변의 검색 정확도와 완성도를 높이는 방법으로 구성하고 싶다.

다음 게시물은 왜 LangGraph 라이브러리를 사용했는지에 대한 구체적인 내용과 구현 내용을 적어볼 것이다 😉

하지만, 학습 시킬 수 있는 데이터의 개수가 부족하기 때문인지 정확도가 좋지 않았고 학습하는 시간도 너무 오래 걸리기 때문에 시간 낭비인듯한 느낌이 들었다. 또한, 프롬프트 엔지니어링(prompt-engineering) 만으로 우리가 원하는 AI 기능을 개발할 수 없었기 때문에 RAG를 도입하기로 결정했다.

만약 나처럼 챗봇을 처음 개발해야 한다면 아래의 단계대로 챗봇 구현 가능성에 대해 파악하면 좋겠다.

1. 프롬프트 엔지니어링(prompt-engineering)을 시도해 본다. 프롬프트 엔지니어링은 언어 모델에게 원하는 결과를 얻기 위해서 언어 모델에게 전달될 텍스트들을 효율적으로 설계하는 것이라고 볼 수 있다. OpenAI 공식 문서에서 프롬프트 엔지니어링을 위한 전략들과 예시들을 확인해 볼 수 있다.

   Prompt Engineering

2. 프롬프트 엔지니어링으로 해결되지 않는다면 RAG(Retrieval-Augmented Generation)를 도입한다. RAG가 무엇인지 아래에서 자세하게 살펴보겠다.

3. RAG를 사용해도 해결되지 않을 때 파인 튜닝을 이용해본다. 파인 튜닝으로 데이터 학습의 효과를 나타내고 싶다면 학습 데이터의 개수가 최소 10만건이 존재해야 한다.

RAG란?

RAG(Retrieval-Augmented Generation)는 LLM의 단점을 개선시키기 위해 학습 데이터 소스 이외에 신뢰할 수 있는 외부 지식 베이스를 참조하도록 하는 기술이다.

간단하게 예를 든다면, 대규모 언어 모델이 시험을 치룰 때 자신이 알고 있는 지식으로만 시험을 보는 것이 아닌 참고할 수 있는 오픈북 하나를 가지고 시험을 볼 수 있는 것이다.

참고할 수 없는 오픈북이 없을 때 LLM이 발생시킬 수 있는 문제점들이 존재한다.

• 자신이 잘 모르는 부분이라면 허위 정보를 제공한다.
• 구체적이고 최신의 정보가 아닌 오래되었거나 일반적인 정보를 제공한다.
• 신뢰할 수 없는 출처의 정보를 제공한다.
• 대화의 맥락을 잘 이해하지 못한다.

RAG를 이용한다면 이러한 문제점들을 개선 시킬 수 있다. 실제 프로젝트에 RAG를 도입하면서 모델의 답변 정확도와 신뢰성이 향상되었기 때문에 매우 효율적인 방법으로 자연어 처리 어플리케이션을 구현할 수 있었고, 모델을 학습시키는데에 큰 비용과 시간이 들지 않았다.

RAG 작동 방식

RAG는 아래와 같은 방식으로 동작하기 때문에 RAG를 구현하기 위해선 벡터 데이터베이스, 텍스트 임베딩 모델, LLM 이 세가지 구성 요소가 꼭 필요하다.

1. 사용자가 질문을 입력한다.
2. 사용자의 질문을 임베딩 모델이 벡터로 변환하여 질문과 관련된 정보를 벡터 데이터 베이스에서 찾는다. 이 벡터 데이터베이스에선 모델이 참고할 수 있는 외부 지식(웹 문서, 기업 내부 문서) 등이 담겨져 있다.
3. 검색된 정보는 LLM의 입력 데이터가 되고, 사용자의 질문과 검색된 데이터를 기반으로 LLM이 답변을 생성한다.
4. 생성된 답변을 사용자에게 전달한다.

내가 선택한 세가지 구성요소는 아래와 같다. DB 생성 과정은 이전 게시물을 참고 하길 바란다.

• 벡터 DB : Milvus
• 텍스트 임베딩 모델: text-embedding-3-small (OpenAI)
• LLM : gpt-4o (OpenAI)

RAG 구현

이제 작동 방식을 알아보았으니 이와 같은 방식으로 RAG를 간단하게 구현해보고자 한다. 구현 테스트는 colab에서 진행 했으며 LLM 챗봇에게 어떤 특정한 장소를 추천 받는 상황으로 가정한다.

1. 필요 라이브러리 및 환경 준비

필요한 라이브러리를 임포트 하고 LLM이 참고할 데이터가 저장된 벡터 DB 테이블을 로드시켜 준비한다.

import pandas as pd
import numpy as np
import time
import os

from pymilvus import MilvusClient
from pymilvus import FieldSchema, DataType
from pymilvus import FieldSchema, CollectionSchema
from openai import OpenAI
from google.colab import userdata

#openai api 쓰기 위한 환경변수 설정
EMBEDDINGS_KEY = userdata.get('EMBEDDINGS_KEY')
os.environ["OPENAI_API_KEY"] = EMBEDDINGS_KEY

openAI_api = OpenAI(
    # This is the default and can be omitted
    api_key=os.environ.get("OPENAI_API_KEY"),
)


# 벡터 데이터베이스 연결
url = userdata.get("URL")
client = MilvusClient(url)


# 컬렉션(=테이블)이름 리스트
collection_list = ['myStartup_travel_sites', 'nowlocal_travel_sites', 'nature_travel_sites']

# 컬렉션 로드하기
for collection_name in collection_list:
    client.load_collection(
        collection_name=collection_name,
      )
    # 로드된 상태인지 확인
      res = client.get_load_state(
        collection_name=collection_name
      )
     print(res)

출력 결과는 아래와 같다.

DEBUG:pymilvus.milvus_client.milvus_client:Created new connection using: 117e983a7c4
{'state': <LoadState: Loaded>}
{'state': <LoadState: Loaded>}
{'state': <LoadState: Loaded>}

2. 사용자 질문 입력 후 벡터화

# 질문 벡터화 함수
def embed_question(question):
    response = openAI_api.embeddings.create(
        input=question,
        model="text-embedding-3-small",
        dimensions=768
    )
    embedding = response.data[0].embedding
    return embedding


# 사용자 질문 저장
example_question = "서울 카페 추천해줘"

# 질문 임베딩
embedding = embed_question(example_question)

사용자는 챗봇에게 "서울 카페를 추천해줘" 라고 질문한다. embedding 변수에 사용자 질문이 768 차원의 벡터로 변환되어 저장된다.

3. 벡터 DB 검색

# 단일 테이블 검색 함수
def search_table(table_name, embedding, top_k):
    search_params = {"metric_type": "IP", "params": {}}
    results = client.search(
        collection_name=table_name,
        data=[embedding],
        anns_field="embedding",
        search_params=search_params,
        output_fields=["id", "text"],
        limit=top_k
    )
    return results

# 여러 테이블 검색 함수
def search_all_tables(embedding):
    results_myCreator = search_table('myStartup_travel_sites', embedding, top_k=6)
    results_nowLocal = search_table('nowlocal_travel_sites', embedding, top_k=5)
    results_nature = search_table('nature_travel_sites', embedding, top_k=4)

    return results_myCreator, results_nowLocal, results_nature


# 테이블 검색
results_myCreator, results_nowLocal, results_nature = search_all_tables(embedding)

테이블 세개에 각각 쿼리를 날려 사용자 질문에 기반한 관련 정보를 가져온다. 각각의 쿼리 결과로 데이터의 기본 키 id와 text 문서를 리턴한다. 벡터 DB 검색 시 유사도 metric 유형은 IP이며 데이터 베이스에 존재한 벡터와 사용자 질문 벡터의 거리를 측정하여 가까운 거리의 데이터들을 반환한다.

Milvus 벡터 검색 함수들을 구체적으로 알고 싶다면 single vector search 공식 문서를 참고해주길 바란다.

4. 검색 결과 재구성

# 쿼리 검색 결과 하나의 문자열로 수정
def format_results(results_myCreator, results_nowLocal, results_nature):

    list_of_results = [results_myCreator[0], results_nowLocal[0], results_nature[0]]
    formatted_results = ""

    for result in list_of_results:
      length = len(result)
      for num in range(length):
        formatted_results += result[num]['entity']['text'] + "\n"

    return formatted_results


# 쿼리 결과 하나의 문자열로 묶기
formatted_results = format_results(results_myCreator, results_nowLocal, results_nature)
formatted_results

쿼리 결과로 얻은 데이터를 하나의 텍스트로 구성하여 묶는다. 아래와 같은 형식으로 쿼리 결과가 하나의 텍스트로 만들어 진다.

장소명: 구욱희씨 서울숲 본점
카테고리: 카페/디저트
장소 키워드: 디저트 카페, 카페
위치: 서울 성동구 성수동1가 685-271 1, 2층

장소명: 새서울
카테고리: 주류
장소 키워드: 추억여행, 바
위치: 서울 종로구 돈화문로11길 28-5

장소명: 서울앵무새
카테고리: 카페/디저트
장소 키워드: 디저트, 카페, 앵무새, 볼거리
위치: 서울 성동구 성수동1가 685-213 B1~2F

장소명: 성수동대림창고갤러리
카테고리: 카페/디저트
장소 키워드: 카페, 디저트
위치: 서울 성동구 성수동2가 322-32

....

5. LLM에 참고 자료 전달 및 답변 생성

# LLM 답변 생성 함수
def sendLLM(question, results):
    response = openAI_api.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "- 당신은 여행을 계획하는데 도움을 주는 챗봇 TBTI입니다. \n-당신의 역할은 사용자의 질문에 reference를 바탕으로 답변하는 것 입니다. \n- 만약 사용자의 질문이 reference와 관련이 없다면, {제가 가지고 있는 정보로는 답변할 수 없습니다.}라고만 반드시 말하세요."},
            {"role": "user", "content":f"사용자 질문: {question} \n reference: {results}" }
        ],
    )
    return response.choices[0].message.content


#LLM에 전달
LLM_response = sendLLM(example_question, formatted_results)
LLM_response

LLM에 참고 자료를 전달해 답변을 생성했다. 출력 결과를 확인해보면 참고 자료의 정보만을 이용해 답변을 생성한 것을 볼 수 있다.

[ 출력 결과 ]

서울에서 추천할 만한 카페는 아래와 같습니다:

1. **구욱희씨 서울숲 본점**
   - 위치: 서울 성동구 성수동1가 685-271 1, 2층
   - 키워드: 디저트 카페, 카페

2. **서울앵무새**
   - 위치: 서울 성동구 성수동1가 685-213 B1~2F
   - 키워드: 디저트, 카페, 앵무새, 볼거리

3. **성수동대림창고갤러리**
   - 위치: 서울 성동구 성수동2가 322-32
   - 키워드: 카페, 디저트

4. **LCDC SEOUL**
   - 위치: 서울 성동구 성수동2가 275-28
   - 키워드: LCDC,카페, bar, 베이커리, 문화공간

이 카페들은 성수동 일대에 위치하고 있어 접근성도 좋고 다양한 경험을 할 수 있습니다.

• System :

  • 당신은 여행을 계획하는데 도움을 주는 챗봇입니다.
  • 당신의 역할은 사용자의 질문에 reference를 바탕으로 답변하는 것 입니다.
  • 만약 사용자의 질문이 reference와 관련이 없다면, {제가 가지고 있는 정보로는 답변할 수 없습니다.} 라고만 반드시 말하세요.

• User:

  • 사용자 질문: {사용자 질문} reference: {참고 자료}

이렇게 간단하게 RAG를 구현하여 LLM의 환각 증세를 줄이고 원하는 답변 결과를 얻을 수 있었다. 만약 챗봇의 말투를 다른 방식으로 변화 시키고 싶다면 사용 언어 모델의 파라미터나 파인 튜닝 기법으로 변화 시킬 수 있을 것이다.

더 유연한 챗봇 개발을 위해 langchain 프레임 워크를 활용하여 간편하고 빠른 어플리케이션을 구현해볼 수 있다.

[ 참고 자료 ] https://aws.amazon.com/ko/what-is/retrieval-augmented-generation/ https://www.ncloud-forums.com/topic/277/ https://modulabs.co.kr/blog/retrieval-augmented-generation/ https://milvus.io/docs/build-rag-with-milvus.md https://velog.io/@pearl1058/Fine-tuning-%ED%8C%81%EB%93%A4AWSKRUG-%ED%9B%84%EA%B8%B0

Rag 시스템을 만들기 위한 테스트 과정에서의 기록입니다.

이전 게시물에서 Milvus Lite를 이용해 간단하게 데이터베이스를 구축하고 컬렉션을 생성했다. 이번에는 생성한 컬렉션에 데이터를 입력해보는 과정을 알아보겠다.

Milvus Lite가 무엇인지 알고싶다면 이전 게시물이나 공식 문서를 참고해주길 바란다.

Run Milvus Lite

1. DB 연결 및 컬렉션 로드

# Milvus Lite DB 연결
from pymilvus import MilvusClient

client = MilvusClient("milvus_demo.db")

출력) 
DEBUG:pymilvus.milvus_client.milvus_client:Created new connection using: 117e983a7c4f4e0eb80454971bbde4a2

Milvus Lite DB 데이터베이스 파일명을 이용해 데이터베이스에 연결한다.

# 만들어진 컬렉션 확인
client.list_collections()

출력)
['myStartup_travel_sites', 'nature_travel_sites', 'nowlocal_travel_sites']

만들어진 컬렉션 리스트를 확인해서 컬렉션 이름을 확인한다.

# 컬렉션 이름과 가져올 데이터 파일명 저장
data_list = {
    # key   : 컬렉션 이름 
    # value : 가져올 데이터 파일이름
    'myStartup_travel_sites' : 'data_myStartup_time.xlsx',
    'nowlocal_travel_sites' : 'data_nowlocal_time.xlsx',
    'nature_travel_sites' : 'data_natural_attractions.xlsx'
}


# 컬렉션 로드하기
for collection_name in data_list.keys():
    client.load_collection(
      collection_name=collection_name,
  )

MilvusClient.load_collection() 전달인자에 컬렉션 이름을 넣고 컬렉션을 로드한다. 로드 이후에 각 컬렉션에 저장될 데이터를 가져오기 위하여 컬렉션 이름과 데이터 파일명을 저장해두었다.

# 컬렉션 로드 된지 확인
for collection_name in data_list.keys():
  res = client.get_load_state(
    collection_name=collection_name
  )
  print(res)

출력)
{'state': <LoadState: Loaded>}
{'state': <LoadState: Loaded>}
{'state': <LoadState: Loaded>}

마지막으로 컬렉션이 로드 되었는지 MilvusClient.get_load_state()를 이용해 확인한다.

2. 데이터 생성 후 입력

- 생성할 입력 데이터 예시

# 입력 데이터 예시
data=[
    {"text": '장소명: 경복궁\n카테고리: 자연명소 ... ', "embedding": [0.3580376395471989, -0.6023495712049978, 0.18414012509913835, -0.26286205330961354, 0.9029438446296592]},
    {"text": '장소명: 남산타워\n카테고리: 자연명소 ... ', "embedding": [0.19886812562848388, 0.06023560599112088, 0.6976963061752597, 0.2614474506242501, 0.838729485096104]},
    {"text": '장소명: 남산타워\n카테고리: 자연명소 ... ', "embedding": [0.43742130801983836, -0.5597502546264526, 0.6457887650909682, 0.7894058910881185, 0.20785793220625592]} 
]

먼저 입력 데이터의 예시를 보도록 하겠다. Milvus DB에서는 입력할 데이터를 딕셔너리 목록으로 구성하고, 각 딕셔너리는 엔티티를 의미한다.

딕셔너리 구성을 보면 text와 embedding 키만 존재하는 데, id는 스키마 생성할 때 automatic id로 설정했기 때문에 데이터베이스가 알아서 id를 지정하고 데이터에 추가해준다. ( automatic id 가 아니라면 id도 딕셔너리에 포함하여 설정해줘야 한다. )

[ 필드값 설명 ]

• text : 장소의 정보(장소명, 위치, 카테고리, 장소키워드)가 담긴 문서

    text 예시)
    장소명: 오일차
    카테고리: 카페/키워드
    장소 키워드: 추억여행, 디저트
    위치: 서울 성동구 성수동1가 680-219 1층 ...

• embedding : 위의 문서를 벡터화 시킨 데이터

- 데이터 생성 과정

import os

#openai api 쓰기 위한 환경변수 설정
os.environ["OPENAI_API_KEY"] = EMBEDDINGS_KEY

openAI_api = OpenAI(
    # This is the default and can be omitted
    api_key=os.environ.get("OPENAI_API_KEY"),
)

# 컬렉션 이름과 가져올 데이터 파일명 저장
data_list = {
    # key   : 컬렉션 이름 
    # value : 가져올 데이터 파일이름
    'myStartup_travel_sites' : 'data_myStartup_time.xlsx',
    'nowlocal_travel_sites' : 'data_nowlocal_time.xlsx',
    'nature_travel_sites' : 'data_natural_attractions.xlsx'
}

# 임베딩 함수
def embed_string(string):
    response = openAI_api.embeddings.create(
        input=string,
        model="text-embedding-3-small",
        dimensions=768
    )
    embedding = response.data[0].embedding
    return embedding

# 문서 생성 함수
def make_dataset(row):
    string_data = f"""장소명: {row['상호명']}
카테고리: {row['카테고리']}
장소 키워드: {row['키워드']}
위치: {row['주소2']}
"""

    vector = embed_string(string_data)

    return {'text': string_data, 'embedding':vector}


for collection_name, file_name in data_list.items():
  # 엑셀 파일 읽어오기
  original_df = pd.read_excel(file_name, engine='openpyxl')

  # 필요없는 열 버리기
  columns_to_drop = ['설명','평점','운영정보']
  original_df = original_df.drop(columns=columns_to_drop)

  # 원본 파일 복사
  copy_df = original_df.copy()
  #copy_df = copy_df.head(2)

  dataset = copy_df.apply(make_dataset, axis=1).tolist()

  client.insert(
      collection_name=collection_name,
      data=dataset
  )

위의 코드는 데이터 생성 후 입력까지의 전체 코드이며 임베딩 함수는 openAI의 text-embedding-3-small 모델을 이용했다.

1. 입력할 데이터들을 엑셀 파일에서 읽어온 후 필요없는 열을 없앤다. 아래와 같은 형태로 데이터가 남게 된다.

2. 각 행을 기준으로 데이터 프레임을 make_dataset 함수에 적용한다.

3. make_dataset 함수에서 데이터 프레임 각 행의 값들을 이용해 문자열을 만들어 text 키에 저장한다. 그 후에, 만든 문자열을 벡터화 시켜 embedding 키에 값을 집어 넣는다.

4. make_dataset 함수에서 리턴된 딕셔너리 데이터를 모두 리스트에 포함시켜 입력 데이터를 완성한다.

5. MilvusClient.insert() 이 함수에 컬렉션 이름과 데이터셋을 지정하여 입력을 수행한다.

3. 입력된 데이터 개수 확인

# 컬렉션 삽입 데이터 개수 확인
for collection_name, file_name in data_list.items():

  print(f'-- {collection_name} --')
  res = client.query(
      collection_name=collection_name,
      filter="",
      output_fields=["count(*)"]
  )
  print(f'삽입 데이터 개수: {res}')
  print()



출력)
-- myStartup_travel_sites --
삽입 데이터 개수: data: ["{'count(*)': 244}"] , extra_info: {'cost': 0}

-- nowlocal_travel_sites --
삽입 데이터 개수: data: ["{'count(*)': 154}"] , extra_info: {'cost': 0}

-- nature_travel_sites --
삽입 데이터 개수: data: ["{'count(*)': 68}"] , extra_info: {'cost': 0}

MilvusClient.query() 를 이용해 다양한 매개변수 설정으로 입력된 데이터 정보를 확인해 볼 수 있다. 입력하고자 한 데이터의 개수와 입력된 데이터 개수가 같다는 것을 확인해 데이터가 올바르게 잘 들어간 것을 볼 수 있었다.

query(
    collection_name: str,
    filter: str,
    output_fields: Optional[List[str]] = None,
    timeout: Optional[float] = None,
    partition_names: Optional[List[str]] = None,
    **kwargs,
) -> List[dict]

Milvus 데이터 검색의 자세한 정보는 공식 문서를 참고해주길 바란다.

Query

Rag 시스템을 만들기 위한 테스트 과정에서의 기록입니다.

Milvus Lite를 이용해 간단하게 데이터베이스를 구축해보고 테이블 생성을 진행해본다.

Milvus Lite란?

벡터 임베딩과 유사도 검색 기능을 갖춘 오픈 소스 데이터베이스 Milvus의 경량 버전이다. Milvus Lite는 Milvus의 핵심 벡터 검색 기능을 제공하고, 대부분의 기능을 포함한다. 따라서, 백만개 미만의 벡터에 대한 빠른 데모를 실행하거나 프로토타입을 빌드하는데 좋다.

Milvus Lite 구동의 자세한 내용은 아래 공식 문서를 참고하시길 바란다.

Run Milvus Lite

입력 데이터 속성

1. 장소 ID (기본 키)
2. text : 장소의 정보(장소명, 위치, 카테고리, 장소키워드)가 담긴 문서

    text 예시)
    장소명: 오일차
    카테고리: 카페/키워드
    장소 키워드: 추억여행, 디저트
    위치: 서울 성동구 성수동1가 680-219 1층

3. embedding vector : 위의 문서를 벡터화 시킨 데이터
   

1. Milvus 설치 및 모듈 임포트

# pymilvus 설치
!pip install -U pymilvus

# 모듈 임포트
import pandas as pd
import numpy as np
import time
import os

from openai import OpenAI
from pymilvus import MilvusClient
from pymilvus import FieldSchema, DataType
from pymilvus import FieldSchema, CollectionSchema

• 구글 코랩에서 Milvus의 Python SDK 라이브러리인 pymilvus를 통해 milvus를 사용하였다.
• Milvus Lite는 pymilvus와 함께 패키징 되어있기 때문에 설치를 진행해줘야 한다.

2. 벡터 데이터베이스 설정

client = MilvusClient("milvus_demo.db")

출력) 
DEBUG:pymilvus.milvus_client.milvus_client:Created new connection using: 117e983a7c4f4e0eb

• MilvusClient 클래스를 통해 로컬 Milvus Lite 데이터베이스 생성을 해준다.
• *"Milvus_demo.db"** 라는 이름으로 모든 데이터를 저장할 데이터베이스 파일이 생성된다.

3. 컬렉션 생성

# 인덱스, 벡터 차원, 유사도 메트릭 설정
INDEX_TYPE = "FLAT"
DIMENSION = 768
METRIC_TYPE = "IP"

# 컬렉션 생성하는 클래스
class MakeCollections:
  def __init__(self, client, index_type, metric_type, dimension):
    self.client = client
    self.index_type = index_type
    self.metric_type = metric_type
    self.dimension = dimension


  # 스키마 생성
  def create_schema(self):
    fields = [
      FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
      FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=500, description="elements of travel sites"),
      FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=self.dimension, description="vector")
    ]
    schema = CollectionSchema(fields=fields, auto_id=True, description="travel sites")
    return schema

  # 인덱스 생성
  def create_index(self):
    index_params = self.client.prepare_index_params()

    index_params.add_index(
      field_name="embedding",
      index_type=self.index_type,
      metric_type=self.metric_type
    )
    return index_params

  # 컬렉션 생성
  def create_collection(self, collection_name):
    self.client.create_collection(
      collection_name=collection_name,
      schema=self.create_schema(),
      index_params=self.create_index()
    )

    time.sleep(2)

    res = self.client.get_load_state(
      collection_name=collection_name
    )
    print(res)

    return self.client



collection = MakeCollections(client, INDEX_TYPE, METRIC_TYPE, DIMENSION)
kstartup_collection = collection.create_collection("myStartup_travel_sites")
nowlocal_collection = collection.create_collection("nowlocal_travel_sites")
nature_collection = collection.create_collection("nature_travel_sites")

• 아래는 컬렉션을 생성하는 전체 코드이며 스키마와 인덱스 정보가 같은 3개의 컬렉션을 생성하였다.
• 컬렉션은 기존 데이터베이스에서 테이블과 같은 의미이고 컬렉션을 만들 때 저장되는 벡터의 차원, 인덱스 유형, 유사도 메트릭 (similarity metric) 유형을 지정할 수 있다.
• 컬렉션을 생성하기 전 스키마와 인덱스 매개변수를 설정하고 지정한 설정 정보를 이용하여 컬렉션을 생성한다.

3-1. 스키마 생성 메서드

  # 스키마 생성
  def create_schema(self):
    fields = [
      FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
      FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=500, description="elements of travel sites"),
      FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=self.dimension, description="vector")
    ]
    schema = CollectionSchema(fields=fields, auto_id=True, description="travel sites")
    return schema

• 생성 필드: id, text, embedding
• automatic id를 사용하여 id값이 자동으로 지정되고 증가한다.
• 동적 필드는 사용하지 않으며 데이터 입력 시 지정되지 않은 새로운 필드의 추가되는 것을 막는다.

스키마는 컬렉션의 구조를 정의하며 데이터 속성과 유형, 기본 키, 동적 필드 사용여부 등 지정할 수 있다. 스키마를 생성하는 방식은 다양하기 때문에 자세한 내용은 아래의 문서를 참고하길 바란다.

Manage Schema

3-2. 인덱스 파라미터 설정 메서드

  # 인덱스 생성
  def create_index(self):
    index_params = self.client.prepare_index_params()

    index_params.add_index(
      field_name="embedding",
      index_type=self.index_type,
      metric_type=self.metric_type
    )
    return index_params

• 인덱스 선정 필드: embedding
  • index type : FLAT 컬렉션의 데이터셋의 규모가 크지않으며 검색 결과의 높은 정확도를 위해 지정
  • metric type : IP 정규화 되지않은 벡터 데이터이고 IP는 주로 자연어 처리 분야에서 텍스트 유사성 검색에 자주 사용됨

인덱스와 메트릭 유형을 지정하여 효율적으로 데이터를 정렬하고 필요한 데이터를 빠르게 검색할 수 있도록 한다.

Milvus에서는 FLAT, IVF_FLAT, IVF_SQ8, IVF_PQ 등 다양한 인덱스 유형을 제공하며 구체적인 인덱스 알고리즘과 메트릭에 대해선 다른 블로그나 공식 문서를 참고하길 바란다.

In-memory Index Similarity Metrics

3-3. 컬렉션 생성 메서드

  # 컬렉션 생성
  def create_collection(self, collection_name):
    self.client.create_collection(
      collection_name=collection_name,
      schema=self.create_schema(),
      index_params=self.create_index()
    )

    time.sleep(2)

    res = self.client.get_load_state(
      collection_name=collection_name
    )
    print(res)

    return self.client

• MilvusClient.create_collection() 함수를 통해 생성할 컬렉션 이름을 지정하고 위에 단계에서 만든 스키마와 인덱스 파라미터를 전달해준다.
• MilvusClient.get_load_state() 함수를 통해 컬렉션이 생성 후 로드되었는지 확인할 수 있다.
• 위의 함수 호출을 통한 컬렉션 생성 후 출력값 예시

출력)

DEBUG:pymilvus.milvus_client.milvus_client:Successfully created collection: nowlocal_travel_sites
DEBUG:pymilvus.milvus_client.milvus_client:Successfully created an index on collection: nowlocal_travel_sites
{'state': <LoadState: Loaded>}

4. 생성한 컬렉션 정보 확인

4-1. 구체적인 정보 확인하기

MilvusClient.describte_collection()에 컬렉션 이름을 지정하면 해당 컬렉션의 필드 정보 및 설정 등 구체적인 정보를 확인할 수 있다.

# 만든 컬렉션 확인
res = client.describe_collection(
    collection_name="nowlocal_travel_sites"
)
res


출력)
{'collection_name': 'nowlocal_travel_sites',
 'auto_id': True,
 'num_shards': 0,
 'description': 'travel sites',
 'fields': [{'field_id': 100,
   'name': 'id',
   'description': '',
   'type': <DataType.INT64: 5>,
   'params': {},
   'auto_id': True,
   'is_primary': True},
  {'field_id': 101,
   'name': 'text',
   'description': 'elements of travel sites',
   'type': <DataType.VARCHAR: 21>,
   'params': {'max_length': 500}},
  {'field_id': 102,
   'name': 'embedding',
   'description': 'vector',
   'type': <DataType.FLOAT_VECTOR: 101>,
   'params': {'dim': 768}}],
 'aliases': [],
 'collection_id': 0,
 'consistency_level': 0,
 'properties': {},
 'num_partitions': 0,
 'enable_dynamic_field': False}

4-2. 생성된 컬렉션 리스트 확인하기

MilvusClient.list_collections()를 통해 만들어진 컬렉션들을 확인할 수 있다.

#만들어진 컬렉션 리스트 확인
client.list_collections()

출력)
['myStartup_travel_sites', 'nature_travel_sites', 'nowlocal_travel_sites']

이제 다음 게시물에서 생성된 컬렉션에 맞는 데이터를 입력해 보는 과정을 진행해 보겠다.

교내 데이터베이스 수업에서 진행한 프로젝트를 기록한 게시물 입니다.

외부 데이터를 활용하여 약물의 성분을 검색하고 해당 성분에 대한 효과를 조사했다. 이 부분은 팀원들이 맡았던 부분으로 조사 과정과 결과를 설명하고자 간략하게 기록한다.

약물의 ndc 코드 조회

약물에 포함된 구체적인 성분을 알아내기 위해 미국 식품의약국(FDA)가 배정한 국가 약물 코드(NDC)를 조회한다. 이후 FDA api를 이용하여 약물의 성분을 구체적으로 조사할 수 있다. 활용 테이블 : hops 모듈의 prescriptions

prescriptions 테이블 공식 문서

< 팀원 코드 첨부 >

import pymysql
import pandas as pd
import requests

# 데이터베이스 접속 정보 설정
connection = pymysql.connect(
    host='호스트',
    user='유저이름',
    password='비밀번호',
    db='데이터베이스'
)

try:
    with connection.cursor() as cursor:
        # sql_mode 설정 변경
        sql = "SET sql_mode = '';"
        cursor.execute(sql)

        # 약물을 그룹화하고 각 약물의 NDC 코드를 조회하는 쿼리
        query = """
        SELECT
            p.drug,
            p.ndc
        FROM
            mimic4_fri_7.hosp_prescriptions p
        GROUP BY
            p.drug, p.ndc

        """

        cursor.execute(query)
        result = cursor.fetchall()
        df = pd.DataFrame(result, columns=['drug', 'ndc'])

• prescriptions 테이블 내의 약물과 NDC를 그룹화하고 각 약물의 NDC 코드를 조회한다.

FDA Drug API 활용한 약물 성분 검색

openFDA는 공공 FDA 데이터를 제공하는 API로, openFDA drug NDC Directory를 이용해 NDC 코드가 포함된 약품 정보들을 검색할 수 있다. 위의 단계에서 알아낸 각 약물의 NDC 코드를 이용하여 약물 성분 정보를 검색했다.

Drug API는 웹 브라우저를 통하여 호출할 수 있으며 자세한 내용은 공식 문서를 확인하길 바란다.

https://open.fda.gov/apis/drug/ndc/how-to-use-the-endpoint/

검색 결과

자주 쓰이는 10가지 약물에 대한 성분을 조사해본 결과, INSULIN LISPRO, SODIUM CHLORIDE, POTASSIUM CHLORIDE 등이 있었다.

아래의 4가지 성분들은 영양 성분 자체가 약물로 사용되며 각각의 성분이 특정 질환에 대해 어떤 치료효과와 예방효과를 가지고 있는지 연구했다.

1. SODIUM CHLORIDE
2. POTASSIUM CHLORIDE
3. DEXTROSE MONOHYDRATE
4. MAGNESIUM SULFATE HEPTAHYDRATE

약물 성분 효과 연구

1. SODIUM CHLORIDE (염화 나트륨)

• 치료 효과 전해질 균형 유지 ; 급성 및 만성 류마티스 심장질환, 허혈 심장질환, 폐성 심장병, 뇌혈관 질환 등에 도움됨

• 예방 효과 효과 없음 ; 상기 표시된 질환에서 예방 효과 없음

  단, 고혈압성 질환에는 염화 나트륨 사용 제한이 권장됨

2. POTASSIUM CHLORIDE (염화 칼륨)

• 치료 효과 저칼륨혈증 예방 및 치료 ; 급성 류마티스열, 만성 류마티스 심장질환, 고혈압성 질환, 허혈 심장질환, 뇌혈관 질환 등에 도움됨

• 예방 효과 혈압 관리 ; 고혈압성 질환 예방에 도움됨 심혈관 건강 ; 허혈심장질환, 뇌혈관질환 예방에 일정 부분 도움됨

3. DEXTROSE MONOHYDRATE (포도당 일수화물)

• 치료 효과 에너지 공급원 ; 모든 질환에서 에너지 공급원으로 작용

• 예방 효과 효과 없음 ; 대부분의 질환에서 예방 효과 없음

4. MAGNESIUM SULFATE HEPTAHYDRATE (황산 마그네슘 7수화물)

• 치료 효과 항염증 및 혈압 조절 ; 급성 류마티스열, 만성 류마티스심장질환, 고혈압성 질환, 허혈심장질환, 뇌혈관질환 등에서 도움됨

• 예방 효과 심혈관 건강 유지 ; 만성 류마티스심장질환, 고혈압성 질환, 허혈심장질환, 뇌혈관질환 등에서 예방 효과

연구 결론

• 특정 영양성분들이 질환 관리와 예방에 다양한 방식으로 기여할 수 있다는 점을 확인할 수 있다.
• 각 질병에 따라 다르게 작용하는 효과를 고려하여, 약물의 적절한 사용과 보충이 중요함을 알아볼 수 있었다.

프로젝트 후기..

• 조사하는 과정을 수월하게 하기 위해선 분석을 통해 도달할 수 있는 목표치를 구체적으로 정하는 것이 좋다고 느꼈다.
• MIMIC-IV 데이터와 데이터베이스 활용의 이해도를 높일 수 있었다.
• 다양한 데이터 분석 알고리즘과 데이터 시각화 방법을 공부해보고 싶다는 생각이 들었다.

aws와 docker 사용법에 아직 익숙치 않지만 경험을 기록하기 위한 글입니다 😶

직접 조금 사용해보고 느낀 바는 DB 쿼리의 기능이 다양하며, docker와 kubernetes를 활용한 어플리케이션 관리와 배포의 효율성이 매우 좋다고 느꼈다.

Milvus의 자세한 내용은 공식 문서를 참고해보길 바라며, AWS cloud에 Milvus의 설치는 어떻게 이루어지고 접속하는지 바로 알아보겠다.

What is Milvus

나는 ec2 인스턴스에 Milvus DB를 설치하고 public 서버 ip와 지정 port를 통해 DB에 외부 접속하는 방식으로 진행 했다. 그 이유는 프로젝트를 진행을 위해 클라우드로 데이터베이스를 공유해야 했고, 리눅스 서버에서의 Milvus DB의 설치는 매우 간단했기 때문이었다. Milvus의 DB 설치 가이드도 모두 리눅스 명령어로 제공되고 있다.

EC2 인스턴스 연결

aws ec2 인스턴스 유형은 t2.micro이고 OS image는 Amazon Linux 2023 AMI로 선택해 생성했다. 그 다음, 생성한 인스턴스 연결을 진행한다.

나는 Amazon Ec2 콘솔을 사용하여 연결했다.

위의 화면에서 하단의 연결 버튼을 누르면

이렇게 브라우저 기반 클라이언트를 사용하여 Linux 인스턴스에 연결 할 수 있다. 이 방법 이외에도 ssh 키를 이용해서 인스턴스에 연결하는 방법이 있다.

다른 연결 방법은 공식 가이드 문서나 다른 블로그를 참고해주길 바란다.

https://docs.aws.amazon.com/ko_kr/AWSEC2/latest/UserGuide/ec2-instance-connect-methods.html

Docker 설치

docker를 통해 Milvus를 설치하고 이용할 것이기 때문에 먼저 docker를 설치해줘야 한다.

1. 일단 인스턴스에 접속 한 후 최신 버전으로 모든 패키지를 업데이트 한다.

    $ sudo yum update -y

2. 그 다음 Docker를 설치한다.

    $ sudo yum install docker -y

3. 설치 후 Docker를 실행해본다.

    $ sudo service docker start

4. 현재 사용자를 docker 그룹에 추가한다. Docker 그룹에 사용자를 추가하면 Docker 명령어를 실행할 때 마다 'sudo'를 사용할 필요가 없다.

    $ sudo usermod -aG docker ec2-user

5. 인스턴스 재접속 후 Docker 명령어를 실행해본다.

    $docker run hello-world

이제 docker에 Milvus를 설치할 수 있는 사전 작업이 완료되었다.

Milvus 설치

Milvus는 docker container로서 구동되기 위해 설치 스크립트를 제공한다. 스크립트에는 아래의 내용들이 있다.

https://raw.githubusercontent.com/milvus-io/milvus/master/scripts/standalone_embed.sh

1. 아래의 명령어를 입력한다.

    # Download the installation script
    curl -sfL https://raw.githubusercontent.com/milvus-io/milvus/master/scripts/standalone_embed.sh -o standalone_embed.sh

   위의 설치 스크립트를 다운받고 현재 위치한 디렉토리의 내용을 보면 세개의 파일이 다운받아져 있다.

   $ ls
   embedEtcd.yaml standalone_embed.sh user.yaml

   
   

2. 이제 docker 컨테이너를 실행시킨다.

    # Start the Docker container
    $ bash standalone_embed.sh start

아래의 그림 처럼 컨테이너가 생성되고 실행된다.

3. 실행 되고있는 컨테이너 정보를 확인한다.

    $ docker ps

• 정보를 확인해보면 milvus 이름의 docker 컨테이너가 19530 port로 실행되고 있는 것을 알 수 있다.
• 기본 Milvus의 구성을 바꾸기 위해서는 user.yaml 파일에 설정을 추가한 후 서비스를 재시작하면 된다.
• Milvus의 데이터 볼륨은 현재 폴더의 volumes/milvus에 매핑된다.

컨테이너 중지 및 삭제

# Stop Milvus
$ bash standalone_embed.sh stop

# Delete Milvus data
$ bash standalone_embed.sh delete

이제 컨테이너의 실행까지 확인 했기 때문에, AWS 인스턴스 서버의 Milvus DB를 외부 접속 하기 위해서 추가적인 설정을 진행한다.

aws 인바운드 규칙 추가

aws 인스턴스에 접속하는 트래픽을 제어하는 인바인드 규칙을 추가하여 정해진 경로로만 접속 할 수 있도록 한다. 해당 인스턴스의 보안 그룹을 선택하여 인바운드 규칙 편집을 클릭하여 추가한다.

예시

• Type : 사용자 지정 TCP
• 프로토콜: TCP
• 포트 범위: 19530
• Source: 0.0.0.0/0 (모든 IP 주소 허용, 원래 보안을 위해선 특정 ip로만 접속 할 수 있도록 해야함)

기본적으로 Milvus 컨테이너는 port 19530으로 설정되기 때문에 인스턴스의 port도 19530으로 지정했다.

Docker 포트포워딩

aws 인스턴스에 위의 19530 port로 접속했을 시에 내부 도커 컨테이너와 접속하기 위해선 포트 포워딩이 필요하다. 호스트의 포트를 컨테이너 포트와 연결하여 밖에서 온 통신을 컨테이너 포트로 전달하는 것이다.

먼저 기존의 Milvus 컨테이너를 멈춘다.

$ bash standalone_embed.sh stop

아래의 명령어 처럼 컨테이너를 생성하고 실행하면 되는데 standalone_embed.sh에 이 코드가 존재하기 때문에

$ docker run -d -p 19530:19530 milvusdb/milvus:태그명

그냥 쉘 스크립트를 지운 후, 다시 한번 실행해주었다.

$ bash standalone_embed.sh delete
$ bash standalone_embed.sh start

Milvus 서버 접속

포트 포워딩 후 Milvus 컨테이너를 실행하여 DB를 시작해두고 외부에서 접속해본다.

아래의 코드를 실행했을 때 연결된 정보를 가져다주면 접속에 성공한 것이다.

from pymilvus import connections

conn = connections.connect(host="aws public ip", port=19530)
connections.get_connection_addr(alias="default")

# Output
# {'address': 'aws public ip 주소:19530', 'user': ''}

참조글 https://milvus.io/docs/install_standalone-docker.md https://chati.tistory.com/124 https://jinjinyang.tistory.com/46

교내 데이터베이스 수업에서 진행한 프로젝트를 기록한 게시물 입니다.

지난 게시물에서 MIMIC-IV 데이터와 선정한 연구 주제에 대해서 설명했다. 연구의 순서대로 최다 처방 약물 10가지를 추출하고 해당 약물들이 처방되는 질환에는 어떠한 것이 있는지 알아보자.

처방된 약물 검색

MIMIC-IV 데이터셋에서 hosp 모듈의 prescriptions 테이블을 활용했다. prescriptions 테이블은 처방된 약물에 대한 정보를 제공하고 약물 이름 및 국가 약물 코드(= ndc), 처방약 복용량 및 투여 경로 등이 포함된다.

prscriptions 구체적 설명 문서 https://mimic.mit.edu/docs/iv/modules/hosp/prescriptions/

<팀원 코드 첨부>

import pymysql
import pandas as pd

# 데이터베이스 접속 정보 설정
connection = pymysql.connect(
    host='호스트',
    user='유저이름',
    password='비밀번호',
    db='데이터베이스이름'
)

try:
    with connection.cursor() as cursor:
        # SQL 쿼리 작성
        query = """
        SELECT drug, COUNT(*) as count
        FROM mimic4_fri_7.hosp_prescriptions
        GROUP BY drug
        ORDER BY count DESC;
        """

        # 쿼리 실행 및 결과 가져오기
        cursor.execute(query)
        result = cursor.fetchall()

        # 결과를 데이터프레임으로 변환
        df = pd.DataFrame(result, columns=['drug', 'count'])

        # 결과 출력
        print(df)

        # 엑셀 파일로 저장
        df.to_excel('all_prescribed_drugs.xlsx', index=False)

except Exception as e:
    print(f"An error occurred: {e}")
    connection.rollback()  # 오류 발생 시 롤백

finally:
    # 데이터베이스 연결 닫기
    if connection:
        connection.close()

1. SQL 쿼리문을 보면 prescriptions 테이블에서 약물 처방 횟수를 계산한다.
2. 각 약물을 기준으로 그룹화하여 처방 횟수를 정리했다.

검색 결과

결과를 살펴보니 자주 쓰이는 약물 10가지에는 Insulin, 0.9% Sodium Chloride, Sodium Chloride 0.9% Flush 등이 존재하는 것을 확인할 수 있었다.

약물이 처방된 질환 검색

이제 앞서 조사한 약물 10가지가 어떤 질환에 주로 처방되는지 알아보고 이를 질병코드별로 분류하겠다. 이 부분에서는 앞 단계에서 사용했던 prescriptions 테이블과 hosp 모듈의 diagnoses_icd 테이블을 활용했다. diagnoses_icd 테이블은 치료기간 동안 환자의 모든 진단 기록들이 들어 있기 때문에 질병 코드를 얻을 수 있다.

diagnoses_icd 구체적 설명 문서 https://mimic.mit.edu/docs/iv/modules/hosp/diagnoses_icd/

# 데이터베이스 접속 정보 설정
connection = pymysql.connect(
    host='호스트',
    user='유저이름',
    password='비밀번호',
    db='데이터베이스이름'
)

try:
    with connection.cursor() as cursor:
        # SQL 쿼리 작성
        query = """
            WITH drugs_of_patients AS(
                SELECT subject_id, GROUP_CONCAT(drug) AS drugs
                FROM hosp_prescriptions
                GROUP BY subject_id
            )
            SELECT icd_code, drugs
            FROM hosp_diagnoses_icd A JOIN drugs_of_patients B ON A.subject_id = B.subject_id
            WHERE icd_version = 10
            GROUP BY icd_code;
        """

        # 쿼리 실행 및 결과 가져오기
        cursor.execute(query)
        result = cursor.fetchall()

        # 결과를 데이터프레임으로 변환
        df = pd.DataFrame(result, columns=['icd_code', 'drugs'])

1. 먼저 prescriptions 테이블에서 환자별로 처방된 약품의 목록을 만들어 서브 테이블을 생성한다.
2. 서브 테이블과 diagnoses_icd 테이블을 동등 조인하여 각 환자들의 질병 코드에 따른 약품 목록을 추출한다. 질병 코드에는 ICD-9와 ICD-10이 있는데 ICD-10 질병 코드만 추출하였다.

ICD-10 버전의 질병 코드는 첫번째 문자인 알파벳을 통해 질병을 분류한다.

출처: KOICD 질병분류정보센터 https://www.koicd.kr/kcd/kcd.do

검색 결과

각 환자들의 질병 코드에 따라 처방된 약품 목록을 확인할 수 있다.

질병 코드 분류

위의 코드와 이어지는 나머지 코드이다.

        top_10_drugs = ['Insulin', '0.9% Sodium Chloride', 'Potassium Chloride', 'Sodium Chloride 0.9%  Flush', 'Furosemide', '5% Dextrose', 'Magnesium Sulfate', 'Metoprolol Tartrate', 'Acetaminophen', 'HYDROmorphone']


        for drugs_name in top_10_drugs:
            print("약물 이름: ", drugs_name)

            #drugs_name이 들어간 행들만 추출
            df_drugs = df[df['drugs'].str.contains(drugs_name)]

            #icd_code(ICD-10)의 첫글자 알파벳을 뽑아 'code_category'열에 새로 저장
            df_drugs = df_drugs.copy()
            df_drugs.loc[:, 'code_category'] = df_drugs['icd_code'].str[0]

            # 'Z'인 행 삭제
            df_drugs = df_drugs[df_drugs['code_category'] != 'Z']

            # 상위 3개의 icd_code 종류 출력
            icd_code = df_drugs['code_category'].value_counts().head(3)
            icd_code = pd.DataFrame(icd_code).reset_index()
            icd_code.columns = ['top_3_icdCode', 'count']
            icd_code.index += 1
            print(icd_code)
            print()



except Exception as e:
    print(f"An error occurred: {e}")
    connection.rollback()  # 오류 발생 시 롤백

finally:
    # 데이터베이스 연결 닫기
    if connection:
        connection.close()

1. 반복문을 통해 Insulin, 0.9% Sodium Chloride, Potassium, Chloride 등 자주 쓰이는 약물이 포함된 행들을 각각 추출하여 해당 약물들이 처방된 질병코드들을 알아낸다.
2. 질병코드를 분류하기 위해 icd_code의 첫번째 알파벳을 분리해 'code_category' 새로운 열에 저장한다.
3. 첫번째 알파벳이 'Z'인 행은 삭제한다. Z로 시작하는 코드는 질환이 아닌 진단이나 치료에 관한 내용이라 제외했다.
4. 각 약물들이 가장 많이 처방된 경우의 질병 코드를 상위 3개만 출력한다.

출력 결과

약물 이름:  Insulin
  top_3_icdCode  count
1             I     73
2             R     57
3             E     47

약물 이름:  0.9% Sodium Chloride
  top_3_icdCode  count
1             I     97
2             R     67
3             K     58

약물 이름:  Potassium Chloride
  top_3_icdCode  count
1             I     91
2             R     60
3             K     58

약물 이름:  Sodium Chloride 0.9%  Flush
  top_3_icdCode  count
1             I     97
2             R     67
3             K     58

약물 이름:  Furosemide
  top_3_icdCode  count
1             I     78
2             R     60
3             K     53

약물 이름:  5% Dextrose
  top_3_icdCode  count
1             I     92
2             R     64
3             K     56

약물 이름:  Magnesium Sulfate
  top_3_icdCode  count
1             I     98
2             R     67
3             K     58

약물 이름:  Metoprolol Tartrate
  top_3_icdCode  count
1             I     80
2             R     54
3             K     38

약물 이름:  Acetaminophen
  top_3_icdCode  count
1             I     91
2             R     64
3             K     56

약물 이름:  HYDROmorphone
  top_3_icdCode  count
1             I     63
2             R     53
3             E     48

출력 결과를 확인해 봤을 때, 공통적으로 알파벳 'I'로 시작하는 질병 코드가 가장 많이 존재한다는 것을 확인할 수 있었다.

분석 결과 1

질병 코드가 I로 시작하는 경우의 질환은 순환계통 질환(Diseases of the circulatory system, I00-I99)이다.

따라서 자주 쓰이는 약물 10가지는 심장 질환, 혈관 질환, 고혈압 및 저혈압 질환에 주로 처방된다는 것을 알 수 있다.

추가적 조사

순환계통 질환 중 어떤 질환일 때 위의 약물 10가지가 많이 처방되는 건지 구체적으로 궁금해서 더 알아보았다.

import numpy as np

def classification_of_ICode(x):
    ranges = [
        (0, 2),
        (5, 9),
        (10, 15),
        (20, 25),
        (26, 28),
        (30, 52),
        (60, 69),
        (70, 79),
        (80, 89),
        (95, 99)
    ]

    for i, (start, end) in enumerate(ranges):
        if start <= x <= end:
            count_code[i] += 1

top_10_drugs = ['Insulin', '0.9% Sodium Chloride', 'Potassium Chloride', 'Sodium Chloride 0.9%  Flush', 'Furosemide', '5% Dextrose', 'Magnesium Sulfate', 'Metoprolol Tartrate', 'Acetaminophen', 'HYDROmorphone']

for drugs_name in top_10_drugs:
    print(drugs_name)

    #drugs_name이 들어간 행들만 추출
    df_drugs = df[df['drugs'].str.contains(drugs_name)]

    #icd_code(ICD-10)의 첫글자 알파벳을 뽑아 'code_category'열에 새로 저장
    df_drugs = df_drugs.copy()
    df_drugs.loc[:, 'code_category'] = df_drugs['icd_code'].str[0]

    #icd_code가 I로 시작하는 행들만 추출
    df_drugs = df_drugs[df_drugs['code_category'] == 'I']

    # 'Z'인 행 삭제
    df_drugs = df_drugs[df_drugs['code_category'] != 'Z']
    #print(df_drugs)

    #icd_code 코드 숫자 추출
    I_code_df = df_drugs.copy()
    I_code_df['icd_code'] = I_code_df['icd_code'].str[1:3]
    I_code = I_code_df['icd_code'].astype(int)

    count_code = np.zeros(10, dtype=int)

    I_code.apply(classification_of_ICode)

    print(f'순환 계통 질환 10분류 개수: {count_code}')
    print(f'가장 많이 처방받는 질환 : {count_code.argmax() + 1}분류 질환')
    print()

1. 각 약물들이 처방된 경우의 질병 코드 중에서 icd_code가 알파벳 'I'로 시작하는 행들만 추출한다.
2. 앞서 추출한 행들의 데이터 프레임에서 I로 시작하는 질병 코드의 숫자 부분을 추출한다. 즉, 질병 코드가 I654라면 65인 부분만 뽑아내어 기본 코드를 알아낸다. * I654 질병 코드는 I65.4로 표현할 수 있는데, 소수점 부분은 일반적인 질환 범주에서 더 구체적인 진단 세부사항을 나타낸다.
3. 기본 코드를 추출하여 순환 계통 질환 분류 범위에 따라 데이터프레임 각 행의 코드를 분류해 개수를 센다.

분석 결과 2

Insulin
순환 계통 질환 10분류 개수: [ 0  3  4  9  3 26  9  4  7  8]
가장 많이 처방받는 질환 : 6분류 질환

0.9% Sodium Chloride
순환 계통 질환 10분류 개수: [ 0  4  6  9  4 34 13  6 11 10]
가장 많이 처방받는 질환 : 6분류 질환

Potassium Chloride
순환 계통 질환 10분류 개수: [ 0  3  4  9  4 30 13  7 11 10]
가장 많이 처방받는 질환 : 6분류 질환

Sodium Chloride 0.9%  Flush
순환 계통 질환 10분류 개수: [ 0  4  6  9  4 34 13  6 11 10]
가장 많이 처방받는 질환 : 6분류 질환

Furosemide
순환 계통 질환 10분류 개수: [ 0  3  5  8  3 29  9  6  8  7]
가장 많이 처방받는 질환 : 6분류 질환

5% Dextrose
순환 계통 질환 10분류 개수: [ 0  4  6  9  4 33  9  7 11  9]
가장 많이 처방받는 질환 : 6분류 질환

Magnesium Sulfate
순환 계통 질환 10분류 개수: [ 0  4  6  9  4 34 13  7 11 10]
가장 많이 처방받는 질환 : 6분류 질환

Metoprolol Tartrate
순환 계통 질환 10분류 개수: [ 0  4  6  9  2 29 11  7  7  5]
가장 많이 처방받는 질환 : 6분류 질환

Acetaminophen
순환 계통 질환 10분류 개수: [ 0  4  6  9  2 32 13  6  9 10]
가장 많이 처방받는 질환 : 6분류 질환

HYDROmorphone
순환 계통 질환 10분류 개수: [ 0  3  6  6  2 28  3  4  8  3]
가장 많이 처방받는 질환 : 6분류 질환

위의 결과를 보면 각 약물들은 순환계통의 질환에서 6번째 범위에 속하는 기타 형태의 심장병으로 급성 심장막염, 비류마티스성 승모판 장애, 심근병증 등의 질병에 가장 많이 처방된 것을 알 수 있었다.

교내 데이터베이스 수업에서 진행한 프로젝트를 기록한 게시물 입니다.

MIMIC-IV 데이터는 무엇인가?

MIMIC-IV(Medical Information Mart for Intensive Care IV)는 미국 메사추세츠주 보스턴에 있는 Israel Deaconess Medical Center에 입원한 환자들의 실제 병원 체류 기록을 포함하는 관계형 데이터베이스이다. 2008년부터 2019년까지의 기간 동안 수집된 자료를 포함하고 환자의 생체 신호, 투여된 약물, 입원 및 퇴원 등 포괄적인 정보가 포함되어 있다. 따라서 총 7.2GB의 매우 큰 데이터셋으로 몇가지 과정만 거치면 무료로 사용할 수 있다.

PhysioNet에 들어가면, 데이터셋에 대한 설명을 볼 수 있고 제일 아래로 내려가면 *Files *라는 부분이 나온다. 이 부분의 설명대로 여러 조건을 맞추게 된다면 제한된 정보에 접근할 수 있다.

• https://baeseongsu.github.io/posts/mimiciii/

• https://www.notion.so/laplacian/2-2-ad9d27f8044a46a4af2a9bf4754511cd

선정 주제

데이터베이스에 필요한 MIMIC-IV 데이터들을 입력한 후에 이를 활용한 데이터 분석을 진행하였다. 우리팀은 최다 처방약물 10가지의 영양소 및 효능을 분석하는 것을 주제로 삼았다.

연구 목적

1. 약물의 영양 성분이 대상 질환에 어떤 효용성을 갖는지 조사
2. 영양 성분을 미리 섭취하는 것으로 대상 질환을 예방할 수 있는지 연구

연구 순서

1. 처방된 약물 리스트 검색 - 최다 처방 약물 10가지를 추출
2. 약물에 따른 질환 검색 및 분류 - 앞서 추출한 약물이 처방되는 질환을 질병 코드별로 분류
3. 약물에 포함된 영양 성분 조사 - 외부 데이터 활용 약물 성분 검색 및 해당 성분에 대한 효과 조사

연구할 내용에 대해서는 팀원 모두가 기획했고, 나는 두번째 단계를 맡아 이 부분을 구체적으로 설명할 예정이다.

백준 알고리즘을 풀며 알게된 점을 메모한 게시물입니다.

처음 백준을 풀기 시작해보니 코드 처리 속도가 중요하다는 것을 알게 되었다. 오늘은 Scanner와 BufferedReader 차이점과 StringTokenizer와 split의 차이점에 대해 메모한다. 문제는 두 수의 입력을 받고 두 수의 차이값을 출력하는 간단한 문제이다.

문제

백준 문제 1001번 : 두 정수 A와 B를 입력받은 다음, A-B를 출력하는 프로그램을 작성하시오.

입력 : 첫째 줄에 A와 B가 주어진다. (0 < A, B < 10)

출력 : 첫째 줄에 A-B를 출력한다.

Scanner 사용했을 때

import java.util.*; 

public class Main{
    public static void main(String args[]){
        Scanner sc = new Scanner(System.in);
        int a,b;
        a = sc.nextInt();
        b = sc.nextInt();
        System.out.println(a-b);
    }
}

Scanner 클래스를 이용해서 두 수를 nextInt()로 간단하게 읽어왔고, 속도는 212ms 정도 걸렸다.

BufferedReader 사용했을 때

import java.io.*;

public class Main{
    public static void main(String args[]) throws IOException{
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        String[] s = br.readLine().split(" ");
        int a = Integer.parseInt(s[0]);
        int b = Integer.parseInt(s[1]);

        System.out.println(a-b);
    }
}

BufferedReader를 이용해 입력한 값을 읽어왔고, 입력한 값을 분리하는 과정을 거쳤다. 이 경우 처리속도가 120ms 정도이다.

왜 이런 속도차이가 나는 것일까?

Scanner와 BufferedReader의 차이점

1. 데이터 가공

   • Scanner는 값을 정수, 문자열, 불리안 등 다양하게 읽어들일 수 있어 간편하다.
   • BufferedReader는 입력 받은 데이터가 텍스트이기 때문에 문자나 문자열로 읽어들인 후 값을 변환해야 한다.

2. 버퍼 사용

   • Scanner는 키보드의 입력이 키를 누르는 즉시 전송한다.

   • BufferedReader는 8192 byte 크기의 버퍼에 입력 받은 값을 담아두었다가 한번에 전송한다.

     👉 데이터를 하나하나 전달하는 것보다 한번에 모아 데이터를 전송하는 것이 성능적인 측면에서 더 효율적이다. 따라서 버퍼의 사용이 더 빠른 처리 속도를 낼 수 있다.

BufferedReader 사용할 때 주의점

1. java.io 패키지를 import 해준다.
2. throws IOException으로 예외처리를 해준거나 try, catch문 으로 예외처리를 해줘야한다.
3. 여러 데이터를 입력 받을 때는 split() 함수를 사용해주거나 StringTokenizer 클래스를 이용한다. 위의 경우는 split () 을 이용하여 문자열을 구분해주었다.

StringTokenizer 클래스 사용해서 문자열을 구분할 때

import java.util.*;
import java.io.*;

public class Main{
    public static void main(String args[]) throws IOException{
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        StringTokenizer st = new StringTokenizer(br.readLine());

        int a = Integer.parseInt(st.nextToken());
        int b = Integer.parseInt(st.nextToken());

        System.out.println(a-b);
    }
}

StringTokenizer 클래스는 java.util 패키지에 있고, split () 처럼 문자열을 분리해준다. 분리된 문자열을 nextToken()으로 가져올 수 있다. 분리된 모든 문자열을 확인하기 위해서는 hasMoreTokens() 을 이용하여 반복문을 돌면서 확인하는 방법이 있다.

StringTokenizer를 생성하는 방법 3가지도 알아보자.

1. StringTokenizer st = new StringTokenizer(문자열); 👉 띄어쓰기를 기준으로 문자열을 분리해준다.

2. StringTokenizer st = new StringTokenizer(문자열, 구분자); 👉 명시한 구분자로 문자열을 분리해준다.

3. StringTokenizer st = new StringTokenizer(문자열, 구분자, true/false); 👉 세번째 매개변수의 true/false는 구분자를 토큰에 포함할지 여부를 결정한다. true면 구분자를 토큰에 포함되고, false면 포함되지 않는다.

Split과 StringTokenizer의 차이점

• StringTokenizer는 빈문자열을 토큰으로 인식하지 않지만, Split은 빈문자열을 토큰으로 인식한다.
• StringTokenizer는 결과값이 문자열이라면 Split은 결과값이 문자열 배열이다.
• StringTokenizer는 문자 또는 문자열로 입력값을 구분하고, split은 정규 표현식으로 구분한다.

참고 게시물

• https://dev-coco.tistory.com/94
• https://lasbe.tistory.com/48

gitignore 파일을 이용해서 git에 올릴 필요 없는 파일들을 지정해 git에서 제외 시켜보자

.gitignore 파일 생성

이미 존재했던 프로젝트에서 .gitignore을 생성해보겠다.

1. Add file을 누르고 Create new file을 눌러준다.

2. 프로젝트 명 옆에 /(슬래시) 하고 Name your file 칸이 있다. 이 칸에 .gitignore 라고 입력한다.

3. .gitignore 라고 입력하면 .gitignore 파일의 template을 선택할 수 있다. 여기에서 나는 Unity를 선택했다.

자동적으로 Unity에 대한 ignore 파일들을 설정해주었다. 적혀있는 내용 이외에 제외하고 싶은 파일이 있다면 .gitignore 파일에서 패턴에 맞게 추가 작성해주면 된다.

.gitignore 파일 적용

.gitigore 파일을 생성 후 push 했지만 이 파일이 작동하지 않을 수 있다. 중요한 조건이 있다.

• .gitignore 파일은 프로젝트 최상단에 위치해야 한다.
• Unity 프로젝트의 경우 Assets 폴더가 존재하는 곳에 .gitignore 파일이 존재해야 한다.

이 조건에 잘 해당하지만, 또 작동하지 않는 경우는 .gitignore 파일을 git 프로젝트를 생성할 때가 아닌 나중에 추가했을때 그럴 수 있다.

이때는 git의 캐시를 지워줘야 한다.

git rm -r --cached .
git add. 
git commit -m "커밋메세지"
git push origin {브랜치명}

push 후 확인해 봤을 때 불필요한 파일들이 Git에서 제외된 것을 확인할 수 있었고, 작업하면서 프로젝트에 불필요한 파일이 다시 생성되어도 Git에 업데이트 되지 않았다.

Gemma는 어떤 모델인가

젬마(Gemma)는 최근 구글이 발표한 새로운 인공지능 모델로 대규모 언어 모델이다. Google이 작년에 발표했던 제미나이(Gemini)모델의 핵심 기술과 연구를 기반으로 제작된 최점단 경량 오픈 모델로, 모델은 매개변수가 20억개인 젬마 2B와 70억개인 젬마 7B이 존재한다. 젬마(Gemma)의 가장 큰 장점은 다른 오픈 LLM보다 주요 벤치마크에서 더 뛰어난 성능을 보여주고 있으며, 노트북, 데스크탑, 모바일 및 클라우드를 포함한 범용적인 디바이스에서 실행될 수 있어 폭넓게 이용 가능하다.

또한, 코랩(Colab), 캐글 노트북(Kaggle notebooks), 허깅 페이스(Hugging Face) 등 다양한 환경에서 젬마를 이용할 수 있다. 이러한 다양한 환경에서 쉽게 이용해 볼 수 있도록 가이드가 모두 존재하고, 새로운 데이터들로 파인 튜닝된 모델도 쉽게 구할 수 있다.

더 자세한 내용은 구글의 문서를 참고하길 바란다.

• 구글 코리아 블로그

• Gemma 모델 개요
• 구글 클라우드 젬마 웹사이트

Gemma를 사용해보자

구글 젬마(Gemma) 가이드를 보면서 Gemma를 사용해 보았다. KerasNLP는 TensorFlow, Jax, PyTorch에서 기본적으로 작동하는 자연어 처리 라이브러리다. 위의 가이드에서는 KerasNLP를 사용하여 Gemma를 사용해본다.

1. Gemma Setup

Gemma를 사용하기 전에 Kaggle을 통해 모델에 대한 엑세스를 요청해야 한다. Kaggle에 로그인 후 Gemma 모델 카드에서 Request Acess를 눌러 접근 요청을 한다.

엑세스 요청 후 Gemma를 사용하기 위해 Kaggle 사용자 이름과 Kaggle API Key가 있어야 한다. Kaggle 사용자 계정 👉 Settings 로 이동한 후 아래의 API 메뉴에서 Create New Token을 누르게 되면, 사용자 인증 정보가 담긴 Kaggle.json 파일이 다운로드 된다.

Kaggle.json 파일은 다음과 같은 내용이 존재한다.

{"username":"your_username","key":"012345678abcdef012345678abcdef1a"}

코랩의 런타임 유형도 설정해줘야 한다. 코랩 상단메뉴에 런타임 👉 런타임 유형 변경을 선택하고, 젬마 2B를 사용하기 위해 T4 GPU *하드웨어 가속기를 선택한다. 젬마 7B를 사용하기 위해선 *A100 GPU같은 유료 프리미엄 GPU를 설정해줘야 한다.

2. 환경 변수 설정

Gemma 설정을 완료하고 colab 환경의 환경변수를 설정한다.

import os
from google.colab import userdata

# Note: `userdata.get` is a Colab API. If you're not using Colab, set the env
# vars as appropriate for your system.
os.environ["KAGGLE_USERNAME"] = userdata.get('KAGGLE_USERNAME')
os.environ["KAGGLE_KEY"] = userdata.get('KAGGLE_KEY')

위의 코드를 보면 userdata.get() 함수를 이용해 환경 변수를 설정하는 것을 볼 수 있다. 이는 코랩에서 비밀키로 지정한 값을 가져오는 역할을 한다.

코랩 보안 비밀 칸에서 Kaggle.json 파일의 정보들을 입력해 Kaggle 사용자 이름과 Kaggle API Key를 추가할 수 있다.

위의 그림을 보면 KAGGLE_USERNAME 이름으로 username 사용자 이름을 저장하고, KAGGLE_KEY 이름으로 API Key를 저장한 것을 볼 수 있다.

3. 필요 라이브러리 설치

Keras와 KerasNLP를 설치한다.

# Install Keras 3 last. See https://keras.io/getting_started/ for more details.
pip install -q -U keras-nlp
pip install -q -U keras>=3

이런 식으로 설치되는 것이 보이고 밑에 빨간 오류 메시지는 무시하고 지나갔다.(keras 3.3.3을 사용하기 때문에 무시해도 된다...)

4. 백엔드 설정

Keras 3는 새로운 대규모 모델 학습과 배포기능을 제공하며 tensorflow, PyTorch, Jax 중에 하나를 설정하여 사용할 수 있다. 가장 적합한 프레임워크를 선택하고 상황에 따라 프레임워크를 전환할 수 있다.

import os

os.environ["KERAS_BACKEND"] = "jax"  # Or "tensorflow" or "torch".
os.environ["XLA_PYTHON_CLIENT_MEM_FRACTION"] = "0.9"

5. 패키지와 모델 가져오기

kerasNLP은 다양한 모델 아키텍처들을 제공하고 있고, from_preset( ) 메서드를 통해서 모델을 불러올 수 있다.

import keras
import keras_nlp

gemma_lm = keras_nlp.models.GemmaCausalLM.from_preset("gemma_1.1_instruct_2b_en")

위의 목록 처럼 다양한 Gemma 모델 중 하나를 지정하면 모델을 가져올 수 있고, 처음에는 가이드에 따라 gemma_2b_en 모델을 사용했다. gemma_2b_en 모델은 한국어에 대한 이해력이 좋지 않고, 만족할 만한 답변을 얻지 못해서 gemma_1.1_instruct_2b_en 모델을 가져와 사용해 보았다.

gemma의 다른 버전 모델로 바꿔 사용하기 위해서는 코랩의 런타임 연결을 해제 및 삭제해야 한다. gemma 모델의 용량이 크기 때문에 코랩 리소스를 초기화 시킨 후 다른 모델을 받아 올 수 있다.

summary() 메서드를 통해서 모델의 자세한 정보를 확인할 수 있다.

gemma_lm.summary()

6. 텍스트 생성하기

모델에는 프롬프트를 기반으로 텍스트를 생성하는 Generate 메서드가 있다. Generate 메서드 파라미터로 질문을 전달할 수 있고, 선택사항인 max_length 인수는 모델이 생성할 수 있는 텍스트의 최대길이를 지정할 수 있다.

gemma_lm.generate("What is the meaning of life?", max_length=64)

결과)

What is the meaning of life?

The question is one of the most important questions in the world.

It’s the question that has been asked by philosophers, theologians, and scientists for centuries.

And it’s the question that has been asked by people who are looking for answers to their own lives

한국 여행지를 추천해달라는 질문을 해보았다. gemma_1.1_instruct_2b_en 모델은 한국어 지원이 잘 되었고, 나름 구체적인 답변을 얻을 수 있었다.

gemma_lm.generate("한국 여행지 좀 추천해줄래?", max_length=500)

결과)

한국 여행지 추천 좀 해줄래?

**목표:** 한국 여행지 추천

**요건:**

* 관광지와 자연 경관이 풍부한 지역
* 문화적 가치가 높은 지역
* 다양한 활동과 경험 제공
* 편리한 이동 환경

**추천 지역:**

**1. 경기도:**

* 경주: 경기도의 전통 문화와 전통 음식
* 평택: 한국의 최고의 수산물 중 하나인 평택 생선의 가치
* 부산: 한국의 대표적인 해안 도시

**2. 충청도:**

* 천안: 천안의 천연 풍경과 역사적 가치
* 홍천: 한국의 최고의 농경지 중 하나인 홍천 농경지
* 제주도: 한국의 제주도는 자연 풍경과 문화적 가치가 풍부한 지역

**3. 전라도:**

* 전주: 전주는 전통 문화와 전통 음식의 가치
* 강릉: 강릉은 한국의 최고의 수산물 중 하나인 강릉 생선의 가치
* 울산: 울산은 한국의 대표적인 산악 지역


**추천 활동:**

* 관광지 방문
* 자연 경관 관찰
* 수산물 구매
* 전통 음식 경험
* 활동과 경험 제공

**추천 경험:**

* 한국의 전통 무용수 경험
* 한국의 전통 음식 경험
* 한국의 수산물 구매 경험
* 한국의 자연 경관 관찰 경험

질문을 묶어서 한꺼번에 물어볼 수 있고, 그에 대한 답변도 한번에 받을 수 있다.

gemma_lm.generate(
    ["What is the meaning of life?",
     "How does the brain work?"],
    max_length=64)

['What is the meaning of life?


The question is one of the most important questions in the world.

It’s the question that has been asked by philosophers, theologians, and scientists for centuries.

And it’s the question that has been asked by people who are looking for answers to their own lives',

'How does the brain work?

The brain is the most complex organ in the human body. It is responsible for controlling all of the body’s functions, including breathing, heart rate, digestion, and more. The brain is also responsible for thinking, feeling, and making decisions.

The brain is made up']

발생한 문제

유니티에서 수신한 데이터를 아바타에 적용했을 때, 전달받은 데이터대로 사용자의 동작을 따라하지만 렉이 걸린듯이 움직였다. 또한, 아바타에 지정해둔 초기 자세로 자꾸 돌아가는 듯한 느낌이 들었다.

아바타의 초기 자세👇

문제 예측

아바타의 동작에 렉이 걸린다는 점과, 초기 자세로 동작이 자꾸 바뀐다는 점에서

1. 송신한 데이터들이 분실된다.
2. 송신하는 속도보다 수신하는 속도가 빨라 읽어들일 데이터가 부족하다.

문제 해결

데이터를 읽어들이는 속도를 조절해보기 위해서 코루틴(Coroutine)을 사용했다.

코루틴이란? 코루틴을 사용하면 작업을 다수의 프레임에 분산할 수 있다. 코루틴은 실행이 일시 정지되거나 중단된 부분에서 다음 프레임을 계속해서 처리할 수 있는 메서드이다.

유니티 코루틴 개념 참고 https://docs.unity3d.com/kr/2021.3/Manual/Coroutines.html

코드 수정

public class FinalTest : MonoBehaviour
{
    public static Socket sock;
    string serverIP = "123.45.67.89";
    int port = 9999;
    bool socketConnect = false;
    string startPoint = "start"; 

    public Animator anim;
    StoreJointData storeData;
    MoveAvatar moveAvatar;

    bool poseUpdateCheck = true;

    private void Start()
    {
        anim = GetComponent<Animator>();

        storeData = new StoreJointData(anim);
        moveAvatar = new MoveAvatar();

        try
        {
            // TCP 소켓 객체 생성
            sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);

            // 서버에 연결하기
            IPEndPoint serverEP = new IPEndPoint(IPAddress.Parse(serverIP), port);
            sock.Connect(serverEP);

            socketConnect = true;
            Debug.Log("Connect Success");

            // "Start" 문자열 전송
            byte[] buff = Encoding.UTF8.GetBytes(startPoint);
            sock.Send(buff);



        }
        catch (Exception e)
        {
            Debug.Log("오류: " + e);
        }

        StartCoroutine(PoseUpdate());
    }

    private IEnumerator PoseUpdate()
    {
        while (poseUpdateCheck)
        {
            Vector3[] realJoint = new Vector3[13];
            string[] textLine;
            string[] splitXYZ;

            try
            {
                if (socketConnect)
                {
                    // 전송 받을 바이트 수 먼저 받아오기
                    byte[] buffer = new byte[4];
                    int byteCount = sock.Receive(buffer);
                    Array.Reverse(buffer);
                    int dataByteCount = BitConverter.ToInt32(buffer, 0);

                    // 받으려고 하는 데이터의 수만큼 byte 배열 선언 후 데이터 저장
                    byte[] receivedBuffer = new byte[dataByteCount];
                    byteCount = sock.Receive(receivedBuffer);

                    string msg = Encoding.UTF8.GetString(receivedBuffer, 0, byteCount);

                    textLine = msg.Split("\n");

                    for (int i = 0; i < textLine.Length - 1; i++)
                    {

                        string line = textLine[i];

                        splitXYZ = line.Split(' ');

                        if (splitXYZ.All(x => x is string))
                        {
                            realJoint[i].x = float.Parse(splitXYZ[0]);
                            realJoint[i].y = float.Parse(splitXYZ[1]);
                            realJoint[i].z = float.Parse(splitXYZ[2]);
                        }
                        else
                        {
                            realJoint[i].x = 0f;
                            realJoint[i].y = 0f;
                            realJoint[i].z = 0f;
                        }

                    }
                }
            }
            catch (Exception e)
            {
                Debug.Log("오류 : " + e);
                socketConnect = false;
                sock.Close();
            }

            // 파일에서 받은 데이터 저장
            storeData.SetTrackJointData(realJoint);

            // 아바타의 팔다리, 몸통 관절데이터 저장
            storeData.Store();

            //움직이기 위한 관절 데이터 전달
            moveAvatar.SetRequiredData(storeData.limbsJointData, storeData.torsoJointData);

            //아바타 팔다리, 몸통 움직이는 함수
            moveAvatar.MoveLimbs();
            moveAvatar.MoveTorso();

            //파이썬 서버와의 통신 속도를 맞추기 위한 딜레이
            yield return new WaitForSeconds(0.048f);

            // 데이터 청소
            storeData.ClearAllData();
            moveAvatar.ClearAllData();
        }

    }

}

핵심 코드

1. Start( ) 함수를 보면 통신을 연결하고 StartCoroutine(PoseUpdate()); 함수를 사용하여 코루틴을 시작한다.
2. PoseUpdate() 함수는 Update( ) 함수가 하던 작업을 똑같이 포함하고 있고, 업데이트 되는 프레임의 속도를 조절한다.

      // 파이썬 서버와의 통신 속도를 맞추기 위한 딜레이
      yield return new WaitForSeconds(0.048f);

이 yield 문은 실행이 일시 정지되고 다음 프레임에서 다시 시작되는 부분이다. WaitForSeconds()를 사용하여 0.048초간 일시정지 하게되어 다음 프레임이 시작되기 전 시간을 약간 지연시켰다. (통신 상태에 따라 일시정지가 필요한 시간은 다를 수 있음...)

Update()를 코루틴 함수로 바꾸어 데이터를 수신하는 속도를 약간 늦추었다. 이렇게 수정한 결과, 아주 약간의 속도차이가 있지만 아바타가 사용자의 동작을 매우 매끄럽게 잘 따라하는 것을 확인할 수 있었다.

사용자의 동작을 따라하는 짧은 영상...

아쉬운점

1. 이제와서 코드를 다시 확인해보니 코드가 효율적이지 못하고, 메모리의 낭비가 있다고 생각한다.
2. 아바타의 관절을 13개만 사용하여, 아바타가 부자연스럽게 움직이고, 동작의 정확성 또한 떨어진다고 생각한다.

   의의

3. 아바타의 관절 위치값을 이용해 아바타의 새로운 동작을 구현할 수 있다.
4. 실시간으로 사용자의 동작을 트래킹하여 UI상으로 사용자의 동작을 확인해 볼 수 있다.
5. 실시간 통신을 직접 구현해보고, 흐름 제어의 중요성을 깨닫게 되었다.

파이썬 TCP 소켓 통신 코드

import mediapipe as mp
import socket
import time
import cv2


mpPose = mp.solutions.pose
pose = mpPose.Pose()

# 0은 웹캡
cap = cv2.VideoCapture(0)
# 이전 프레임 처리 시간 저장 변수 초기화
pTime = 0
# 출력하고 싶은 관절들의 인덱스
desired_indices = [0, 11, 12, 13, 14, 15, 16, 23, 24, 25, 26, 27, 28]

# TCP 서버 생성
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('localhost', 포트번호))
server_socket.listen(1)  # 최대 1개의 연결을 기다림

print("서버가 연결을 대기 중입니다.")

while True:

    # 클라이언트 연결 대기
    try:
        client_socket, addr = server_socket.accept()
        print(f"{addr}에서 연결이 수락되었습니다.")

        # 문자열 대기
        start_message = client_socket.recv(5).decode()
        if start_message == 'start':
            start_message = ''


        # 실시간을 위해 무한 루프 돌려 줍니다
        while True:

            # 바이트 배열 초기화
            coordinates_bytes = bytearray()

            # 성공여부, 이미지
            success, img = cap.read()

            # 이미지 크기 축소
            new_width = 480  # 가로 크기
            new_height = 640  # 세로 크기
            img_small = cv2.resize(img, (new_width, new_height))

            #rgb로 변경 (mediapipe는 rgb 이미지를 사용)
            imgRGB = cv2.cvtColor(img_small, cv2.COLOR_BGR2RGB)
            results = pose.process(imgRGB)
            # print("시작")

            if results.pose_world_landmarks:

                for desired_idx in desired_indices:
                    landmark = results.pose_world_landmarks.landmark[desired_idx]
                    x = landmark.x
                    y = landmark.y
                    z = landmark.z

                    # 선택한 관절 지점의 정보 출력
                    print(f"{x:.5f} {y:.5f} {z:.5f}".format(x,y,z))

                    # 좌표값을 바이트 배열에 추가
                    coordinates_bytes += f"{x:.5f} {y:.5f} {z:.5f}\n".format(x, y, z).encode()

                coordinates_bytes = coordinates_bytes.rstrip(b'\0')

                # 바이트 배열로 변환된 좌표값을 Unity 서버로 전송
                coordinates_length = len(coordinates_bytes)
                length_bytes = coordinates_length.to_bytes(4, byteorder='big') 

                # 길이 정보 전송

                client_socket.sendall(length_bytes)

                # 바이트 배열로 변환된 좌표값을 Unity 서버로 전송
                client_socket.sendall(coordinates_bytes)



            cTime = time.time()
            fps = 1 / (cTime - pTime)
            pTime = cTime

            # 이미지에 프레임 속도 표시
            # cv2.putText(img, str(int(fps)), (70, 50), cv2.FONT_HERSHEY_PLAIN, 3, (255, 0, 0), 3)
            cv2.imshow("image", img)

            if cv2.waitKey(10) & 0xFF == ord('q'):  # 'q' 키를 누르면 루프 종료
                break

    except Exception as e:
        print(f"연결 수락 중 오류 발생: {e}")


# 연결 종료
client_socket.close()
server_socket.close()
cap.release()
cv2.destroyAllWindows()

유니티 소켓 통신 코드

public class FinalTest : MonoBehaviour
{
    public static Socket sock;
    string serverIP = "123.45.67.89";
    int port = 9999;
    bool socketConnect = false;
    string startPoint = "start"; 

    public Animator anim;
    StoreJointData storeData;
    MoveAvatar moveAvatar;

    Vector3[] realJoint = new Vector3[13];
    string[] textLine;
    string[] splitXYZ;

    private void Start()
    {
        anim = GetComponent<Animator>();

        storeData = new StoreJointData(anim);
        moveAvatar = new MoveAvatar();

        try
        {    
            // TCP 소켓 객체 생성
            sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);

            // 서버에 연결하기
            IPEndPoint serverEP = new IPEndPoint(IPAddress.Parse(serverIP), port);
            sock.Connect(serverEP);

            socketConnect = true;
            Debug.Log("Connect Success");

            // 서버에 "Start" 문자열 전송
            byte[] buff = Encoding.UTF8.GetBytes(startPoint);
            sock.Send(buff);

        }
        catch (Exception e)
        {
            Debug.Log("오류: " + e);
        }
    }

    private void Update()
    {   

        try {

            if (socketConnect)
            {    
                // 전송 받을 바이트 수 먼저 받아오기
                byte[] buffer = new byte[4];
                int byteCount = sock.Receive(buffer);
                Array.Reverse(buffer);
                int dataByteCount = BitConverter.ToInt32(buffer, 0);

                // 받으려고 하는 데이터의 수만큼 byte 배열 선언 후 데이터 저장
                byte[] receivedBuffer = new byte[dataByteCount]; 
                byteCount = sock.Receive(receivedBuffer);

                string msg = Encoding.UTF8.GetString(receivedBuffer, 0, byteCount);

                textLine = msg.Split("\n");

                for(int i = 0; i < textLine.Length - 1; i++)
                {

                    string line = textLine[i];

                    splitXYZ = line.Split(' ');

                    if (splitXYZ.All(x => x is string))
                    {
                        realJoint[i].x = float.Parse(splitXYZ[0]);
                        realJoint[i].y = float.Parse(splitXYZ[1]);
                        realJoint[i].z = float.Parse(splitXYZ[2]);
                    }
                    else
                    {
                        realJoint[i].x = 0f;
                        realJoint[i].y = 0f;
                        realJoint[i].z = 0f;
                    }

                }
            }
        }
        catch(Exception e) 
        { 
            Debug.Log("오류 : " + e);
            socketConnect = false;
            sock.Close();
        }

    }

}

통신하면서 겪은 오류 1

13개의 관절 좌표값을 올바르게 전달 받지 못했다.

-0.02801599 1.530401 0.01000143
0.1409548 1.397012 0.02304516
-0.1969829 1.397008 0.01986888
0.37999 1.397016 0.02304507
-0.4362946 1.397007 0.01986932
0.6318882 1.39702 0.02304519
-0.688216 1.397006 0.01986977
0.0739904 0.884761 0.002631317
-0.1300061 0.884761 0.002718598
0.06866307 0.4908463 0.009188599
-0.124666 0.4908559 0.007917952
0.07400247 0.1169901 0.0270884
-0.1299936 0.1169791 0.02709265

통신할 때 13개의 관절 x y z 좌표값이 위와 같은 형태로 전달 되어야 배열에 값이 잘 저장될 수 있다.

-0.02801599 1.530401 0.0100
0.1409548 1.397012 0.02304516345
-0.1969829 1.397008 0.019868883
0.37999 1.397016 0.023045073535
-0.4362946 1.397007 0.0198
0.6318882 1.39702 0.02304519
-0.6 1.397006 0.01986977
0.0739904 0.884761 0.002631317
-0.1300061 0.884761 0.002718598
0.06866307 0.4908463 

이처럼 x, y, z값 중에 z값이 없거나, 13개의 관절 위치값을 받아오지 못하는 일이 발생했다.

*원인 : * mediaPipe 모듈로 관절의 위치가 매우 미세하게 측정되어 3차원 좌표값의 길이가 매 프레임마다 달라졌기 때문에 일정한 데이터의 길이로 데이터를 전송 받기 어려웠다.

*해결 : * 전송 받을 바이트 수를 먼저 받아온 후, 그 바이트 수만큼의 배열을 생성해 데이터를 저장했다.

  // 전송 받을 바이트 수 먼저 받아오기
  byte[] buffer = new byte[4];
  int byteCount = sock.Receive(buffer);
  Array.Reverse(buffer);
  int dataByteCount = BitConverter.ToInt32(buffer, 0);

  // 받으려고 하는 데이터의 수만큼 byte 배열 선언 후 데이터 저장
  byte[] receivedBuffer = new byte[dataByteCount]; 
  byteCount = sock.Receive(receivedBuffer);

그런데 이 방식은 크기가 다른 배열을 매 프레임마다 새로 생성하기 때문에 힙 메모리를 낭비할 수 있는 문제가 있다. 코드를 수정한다면, 파이썬 서버에서 관절 데이터를 보내기 전에 일정한 길이의 데이터만 보내고, 유니티에서는 배열을 하나만 생성해서 생성한 메모리를 재사용 할 수 있도록 하는 것이 좋을 듯 하다.

통신하면서 겪은 오류 2

원인 : 바이트의 순서가 바뀌어 올바른 값을 얻지 못했다.

파이썬 측에서 보내려고 하는 바이트 수를 먼저 보낸다.

# 바이트 배열로 변환된 좌표값을 Unity 서버로 전송
coordinates_length = len(coordinates_bytes)
length_bytes = coordinates_length.to_bytes(4, byteorder='big') 

# 길이 정보 전송
client_socket.sendall(length_bytes)

# 바이트 배열로 변환된 좌표값을 Unity 서버로 전송
client_socket.sendall(coordinates_bytes)           

to_bytes()로 전송하려고 하는 바이트의 길이를 빅 엔디안(big endian)으로 저장해서 보냈지만, 유니티 측에서는 리틀 엔디안(little endian)으로 받게 되어 올바른 값을 확인하기 힘들었다.

해결 :

// 전송 받을 바이트 수 먼저 받아오기
byte[] buffer = new byte[4];
int byteCount = sock.Receive(buffer);
Array.Reverse(buffer);

유니티에서 전송받은 데이터를 Array.Reverse() 로 이용해 받은 바이트를 역순으로 바꾼다. 바꾼 값을 확인해 봤을 때 파이썬에서 전송했던 값을 올바르게 받을 수 있었다.

이 텍스트 파일의 값들을 아바타에 적용시켜서 아바타가 동작을 잘 수행하는지 최종적으로 점검 했다.

아바타 관절을 움직이는 코드는 저번 게시물 참고🙂 저번 게시물

코드 실행 후 문제점

저번 게시물에서 텍스트 파일에 기록되어있는 관절값을 가져와 아바타에 적용한 후 아바타가 잘 움직이는 것을 확인했다. 그런데, mediaPipe 모듈로 측정한 데이터를 아바타에 적용했을 때 아바타의 동작이 잘 수행되지 않고, 아바타의 몸도 이상해졌다(보기 흉함..)

문제 예측

1. 텍스트 파일에 관절 순서가 뒤바뀌어 기록되었기 때문에 아바타의 관절 위치가 잘못되었나? 👉 확인해보니 아바타의 관절 순서에 맞게 잘 기록되어있음
2. 아바타의 관절을 움직이는 코드에 문제가 있나? 👉 확인해보니 관절 위치를 수정해야 하는 부분이 있었다!

문제 해결

mediaPipe의 Pose Landmark model은 두가지의 Output을 리턴한다. Output 하나는 Landmarks 이고, 다른 하나는 WorldLandmarks 이다. 나는 WorldLandmarks 결과를 사용해 관절의 위치를 텍스트 파일에 기록했다. WorldLandmarks는 관절의 x, y, z값이 모두 엉덩이 사이의 중심(24번과 23번의 중심값)을 원점으로 하는 상대좌표이다.

따라서 mediaPipe 모듈로 측정한 관절의 좌표값을 이용해 아바타를 움직이기 위해선 코드의 수정이 필요하다.

다른 클래스의 코드는 그대로 두고 StoreJointData.cs의 MakeVirtualData 함수를 수정한다.

    // 가상의 관절 만들기, 관절 위치 재조정
    public void MakeVirtualData()
    {
        // 가상의 목 관절의 위치 구하기
        virtualNeck = (trackJoint[1] + trackJoint[2]) / 2.0f;
        virtualNeck.y += 0.05f;

        // 가상의 힙 관절의 위치 구하기
        virtualHips = (trackJoint[7] + trackJoint[8]) / 2.0f;
        virtualHips.y += 0.95f;

        //가상의 UpperChest 관절 위치 구하기
        virtualUpperChest = (trackJoint[1] + trackJoint[2]) / 2.0f;
        virtualUpperChest.y -= 0.1f;

        virtualNeck += virtualHips;
        virtualUpperChest += virtualHips;

        for (int i = 0; i < 13; i++)
        {
            trackJoint[i].y *= -1f; // 트래킹한 조인트 값의 y좌표가 땅과 반대로 되어있음
            trackJoint[i] += virtualHips; // pose_world_landmarks는 엉덩이 중간 포인트를 기준으로 상대좌표이므로 Hips의 위치를 더해 절대 좌표를 구해준다.
        }

        anim.GetBoneTransform(HumanBodyBones.Hips).position = virtualHips;
    }

• virtualHips(엉덩이 사이의 중간 지점)의 위치는 절대 좌표로 잡아두고, 다른 관절들은 virtualHips 위치를 더해 3차원 절대 좌표로 바꾸어준다.

• mediaPipe의 모델이 사용하는 3차원 좌표계와 유니티에서 사용하는 좌표계의 차이가 있어 y좌표값이 반대로 되어 있음을 발견했다.

  trackJoint[i].y *= -1f;

  트래킹한 관절의 y좌표 값에 -1을 곱해 방향을 바꾸어 아바타의 몸이 반대로 뒤집어 지지 않도록 코드를 수정한다.

전체코드

코드를 수정한 후 다시 데이터를 적용했을 때 아바타가 잘 움직인다😁 이제 노트북 웹캠을 통해 실시간 관절 데이터를 유니티 서버로 전달하고, 실시간으로 아바타를 움직여 볼 것이다.