우원 기술 블로그

안녕하세요. 개발하는 우원입니다. 평소처럼 업무를 보던 중에 문득 GPT는 본인 모습을 어떻게 생각하는지 궁금해졌습니다. 물어본 결과는 이미 보셨다시피... 저는 외면할 수 밖에 없었지만요 🤣 재미로 시작한 질문이었지만, GPT의 답변이 꽤나 재밌었고, 한편으로 멀티 모달이 마치 어떤 함수를 호출하는 것처럼 느껴졌습니다.

제가 이번 글에서 다루고자 하는 내용은 "Function Calling"입니다. Function Calling 외에도 약간의 프롬프트 엔지니어링에 대한 내용도 포함하고 있습니다. 좋은 기회로 데보션 오픈랩 스터디에 참가하게 되었고, 그 과정에서 Function Calling 활용해 기능을 구현하게 되었습니다. 개인적으로 얕게만 알고 있던 부분이었는데, 심층적으로 공부하며 제 자신도 돌아보고, 더 나은 개발자(?)가 된 것만 같은 시간이었습니다!

자, 이제 시작하겠습니다.

Function Calling

이미 들어보셨을지 모르겠지만, 함수 호출은 꽤나 최근에 나온 기능입니다. 2023년 6월에 발표한 이후로, 많은 개발자들이 이 기능을 활용하고 있습니다. 저도 기능이 발표되고 나서 문서를 읽어보며, 대략 이런 기능이 있구나 정도로만 알고 있었습니다.

구체적인 정의는 다음과 같습니다.

GPT API 호출에서 호출 함수에 대한 설명을 제공하면, GPT가 하나 이상의 함수를 호출하기 위한 인수가 포함된 JSON 개체를 출력합니다. Chat Completions API가 직접적으로 함수를 호출하지 않지만, 사용자의 코드에서 함수를 호출하는 데 사용할 수 있는 JSON을 생성합니다.

기존 번역이 조금 어려운 부분이 있어, 조금 쉽게 풀어보았습니다. 요약하면, GPT가 함수를 호출하기 위한 인수를 포함한 JSON을 출력한다는 것입니다.

정말 간단하죠? 이제 실제로 사용해보기 전에 연속적 사고를 통해 어떻게 사용할지 생각해보겠습니다.

Chain Of Thought

일단 우리는 GPT가 함수를 호출한다는 사실을 알게 되었습니다. 그런데 원격의 모델이 어떻게 함수를 호출할 수 있을까요? 아마도 대부분의 사람들은 "함수 호출"이라는 용어에 집중하고 모델의 직접적인 호출을 그리고 계실지 모릅니다.

물론 저도 그랬습니다. "아...🧐 함수를 호출해 주겠구나"라고 생각했지만, 실제로는 "대리자" 혹은 "매개자" 역할을 사용자가 부여 받은 것과 다름없습니다.

더 쉬운 예를 들면 virtual class를 예를 들 수 있습니다. 다음 코드를 들여다 봅시다.

class Base {
public:
    virtual void display() {
        std::cout << "출력을 의도하고 있습니다." << std::endl;
    }
};

class Derived : public Base {
public:
    void display() override {
        std::cout << "나는 XXX를 출력하고 싶습니다." << std::endl;
    }
};

위 코드는 Base 클래스와 Derived 클래스를 정의하고 있습니다. Derived 클래스는 Base 클래스를 상속받아 display() 함수를 재정의하고 있죠. Base 클래스는 이미 display라는 함수를 '출력'의 의도 내재한 함수로 정의하고 있습니다. 그렇기에 Derived 클래스에서 어떤 출력을 하고 싶다면, display() 함수를 재정의하는 것입니다.

함수 호출도 이와 비슷합니다. 어떤 함수가 있고, 어떤 기능을 수행하고, 어떤 인자가 들어가는지 의도를 정리해서 GPT에게 전달하면, GPT는 그 의도를 이해하고 상속 받아 상황에 맞게 함수를 호출합니다. 저는 이러한 상태를 "함수 동기화 상태(Function Synchronization State)"라고 정의하고 앞으로 사용하겠습니다. 또한 함수 동기화 상태에서 사용 가능한 함수를 "알고 있는 함수"라고 부르겠습니다.

함수 동기화 상태(Function Synchronization State)

함수 동기화 상태가 되었다면 GPT는 평소와 같이 사용자의 질문을 받기 위해 대기 상태로 들어갑니다. 그리고 사용자가 질문을 하면, GPT는 사용자의 질문을 이해하고, 알고 있는 함수 중에서 사용자의 의도에 맞는 함수를 호출합니다. 여기서 호출이라는 말은 JSON의 형태로 알고 있는 함수의 이름을 반환하는 것을 의미합니다.

좀 더 쉬운 그림으로 표현하자면 다음과 같습니다.

대리자 || 매개자

대리자로서의 사용자는 이제 역으로 GPT의 요청에 응답해야합니다. AIMessage 타입으로 들어온 요청에서 "function_call" 혹은 "tool_calls"이라는 키워드가 있다면 "function_name"을 확인하고 반드시 FunctionMessage or ToolMessage 타입으로 응답해야 합니다. 물론 "function_call", "tool_calls" 키워드가 없다면 일반적인 대화이겠지요?

추출한 function_name은 곧바로 key로 사용되어야 합니다. 그리고 key에 대한 value는 해당 함수의 주소를 가르키는 변수입니다.

간단하게 예를 들어 보겠습니다.

# 함수 매핑 정보
function_map_info = {"get_fruit_price": get_fruit_price, "calc_total_price": calc_total_price}

# 만약 GPT가 "get_fruit_price"를 호출한다면
function_map_info["get_fruit_price"](...args)

이렇게 함수를 호출하면 됩니다. 인자도 놓치지 말고 전달해야 합니다. 정상적으로 함수를 호출하고 나면 곧바로 결과를 string 형태로 넘겨주도록 합니다.

최종 응답

GPT는 응답 결과를 대리자에게 받고나서 마치 직접 함수를 실행한 것처럼 결과 컨텍스트를 활용해 가장 처음 사용자의 질문에 대한 답변을 생성합니다. 이것이 바로 함수 호출의 전체적인 흐름입니다.

강력한 예시

위의 전체적인 흐름을 관통하는 가장 좋은 예시를 작성해보겠습니다. 저는 편의를 위해서 langchain을 적극적으로 활용하겠습니다. (아뇨... 헤어나오지 못하겠습니다... 😅)

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage, AIMessage
from typing import List
import pandas as pd

@tool
def get_fruit_price() -> str:
    """Get the price of the fruits. column [name, price]

    Args:
        None

    Returns:
        str: The price of the fruits.
    """

    fruits_csv_path = "fruits.csv"
    fruits_df = pd.read_csv(fruits_csv_path)
    return str(fruits_df)

@tool
def calc_total_price(fruits_list: List[str]) -> str:
    """calculate the total price of the fruits.

    Args:
        fruits_list (List[str]): The list of fruits.

    Returns:
        str: The total price of the fruits.
    """

    fruits_csv_path = "fruits.csv"
    fruits_df = pd.read_csv(fruits_csv_path)
    result = 0
    for q in query:
        result += fruits_df[fruits_df["name"] == q]["price"]

    return str(result)


def main():
  # 키 선언
  API_KEY = 'sk-XXX...'

  # 모델 선언, 저는 gpt-4o를 사용합니다.
  model = ChatOpenAI(
      model="gpt-4o",
      openai_api_key=API_KEY
  )

  # 컨텍스트 선언
  context = [HumanMessage(content="바나나, 사과, 딸기 각각의 가격과 총 가격을 알려줘.")]

  # 툴 선언, GPT가 사용할 강력한 도구를 불러옵니다.
  tools = [get_fruit_price, calc_total_price]

  # 도구를 GPT에게 쥐어줍니다.
  model_with_tools = model.bind_tools(tools)

  # 대화를 시작합니다.
  ai_msg = model_with_tools.invoke(context)

  # ai의 응답을 컨텍스트에 추가합니다.
  context.append(ai_msg)

  # 결과를 출력합니다.
  print(response)
  """
  content=''
  additional_kwargs={
    'tool_calls': [
      {
        'id': 'call_HU0V8xqccCHDPeDqbHoVYkNv',
        'function': {
          'arguments': '{}',
          'name': 'get_fruit_price'},
          'type': 'function'
      },
      {
        'id': 'call_ISwN3tv9v3zuBB8My6h4Z3F8',
        'function': {
        'arguments': '{
        "fruits_list": ["바나나", "사과", "딸기"]
      }',
      'name': 'calc_total_price'
    },
    'type': 'function'
    }]
  }
  """

  tool_calls = ai_msg.additional_kwargs["tool_calls"]

  # 무조건 tool_calls 속성이 온다고 가정하고, 이를 처리합니다.
  # 빈 딕셔너리를 넣어줍니다. -> 이는 BaseModel이 딕셔너리를 요구하기 때문입니다.
  get_fruit_price_result = get_fruit_price({})

  # 합을 미리 하드코딩합니다.
  calc_total_price_result = 450

  # 결과를 컨텍스트에 추가합니다.
  context.append(ToolMessage(content=get_fruit_price_result, tool_call_id=tool_calls[0]["id"]))
  context.append(ToolMessage(content=calc_total_price_result, tool_call_id=tool_calls[1]["id"]))

  print(context)

  """
  [
    HumanMessage(content='바나나, 사과, 딸기 각각의 가격과 총 가격을 알려줘.'), 
    AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_xWkPPBz5DpN02bJojRbycId8', 'function': {'arguments': '{}', 'name': 'get_fruit_price'}, 'type': 'function'}, {'id': 'call_JK66YU69cWbwCP0FRhGMyswQ', 'function': {'arguments': '{"fruits_list": ["바나나", "사과", "딸기"]}', 'name': 'calc_total_price'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 56, 'prompt_tokens': 131, 'total_tokens': 187}, 'model_name': 'gpt-4o-2024-05-13', 'system_fingerprint': 'fp_c4e5b6fa31', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-e52c3275-22c5-46f6-9556-d4551239553c-0', tool_calls=[{'name': 'get_fruit_price', 'args': {}, 'id': 'call_xWkPPBz5DpN02bJojRbycId8'}, {'name': 'calc_total_price', 'args': {'fruits_list': ['바나나', '사과', '딸기']}, 'id': 'call_JK66YU69cWbwCP0FRhGMyswQ'}], usage_metadata={'input_tokens': 131, 'output_tokens': 56, 'total_tokens': 187}), 
    ToolMessage(content='  name  price\n0   사과    100\n1  바나나    150\n2   딸기    200', tool_call_id='call_xWkPPBz5DpN02bJojRbycId8'), 
    ToolMessage(content='450', tool_call_id='call_JK66YU69cWbwCP0FRhGMyswQ')
  ]
  """
  response = model_with_tools.invoke(context)

  print(response)

  """
  content='### 과일 가격\n- **바나나**: 150원\n- **사과**: 100원\n- **딸기**: 200원\n\n### 총 가격\n총 가격은 450원입니다.' 
  response_metadata={'token_usage': {'completion_tokens': 51, 'prompt_tokens': 249, 'total_tokens': 300}, 'model_name': 'gpt-4o-2024-05-13', 'system_fingerprint': 'fp_c4e5b6fa31', 'finish_reason': 'stop', 'logprobs': None} 
  id='run-155443eb-3ff3-4bf9-815c-7094114e5e9d-0' 
  usage_metadata={'input_tokens': 249, 'output_tokens': 51, 'total_tokens': 300}
  """

if __name__ == "__main__":
    main()

하나씩 천천히 뜯어 보면서 이해해보도록 하죠.

도구 선언부

from langchain_core.tools import tool

@tool
def get_fruit_price() -> str:

@tool
def calc_total_price(fruits_list: List[str]) -> str:

@tool 데코레이터를 활용해서 해당 함수가 도구임을 명시합니다. 데코레이터 형식으로 사용하는 것 외에도 객체 선언 후 name, description, args_schema를 입력하는 것도 가능합니다.

모델 선언

# 모델 선언, 저는 gpt-4o를 사용합니다.
model = ChatOpenAI(
    model="gpt-4o",
    openai_api_key=API_KEY
)

# 컨텍스트 선언
context = [HumanMessage(content="바나나, 사과, 딸기 각각의 가격과 총 가격을 알려줘.")]

# 툴 선언, GPT가 사용할 강력한 도구를 불러옵니다.
tools = [get_fruit_price, calc_total_price]

# 도구를 GPT에게 쥐어줍니다.
model_with_tools = model.bind_tools(tools)

# 대화를 시작합니다.
ai_msg = model_with_tools.invoke(context)

# ai의 응답을 컨텍스트에 추가합니다.
context.append(ai_msg)

이미 주석을 통해서 흐름 자체는 이해하셨겠지만, 한 가지 중요한 포인트가 있습니다. 그것은 'context' 배열에 대화를 차곡차곡 쌓아주는 것입니다. 이것은 function call이 발생했다는 것을 가정했을 때, 반드시 이렇게 돌려달라는 요청에 의해서 수행하게 되는 것입니다. 최초 요청에 function call을 요청한 ai의 메세지, 그리고 어떤 툴을 요청했는지 구별할 수 있는 id와 함께 함수 실행 결과를 포함하면 완벽한 응답이 완성됩니다.

좀 더 구체적으로 살펴보도록 하겠습니다.

나의 요청 쌓기

context = [HumanMessage(content="바나나, 사과, 딸기 각각의 가격과 총 가격을 알려줘.")]

GPT 응답 쌓기

context.append(ai_msg)

함수 실행 결과 쌓기

messages.append(ToolMessage(content=get_fruit_price_result, tool_call_id=tool_calls[0]["id"]))
messages.append(ToolMessage(content=calc_total_price_result, tool_call_id=tool_calls[1]["id"]))

최종 형태

"""
[
  HumanMessage(content='바나나, 사과, 딸기 각각의 가격과 총 가격을 알려줘.'), 
  AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_xWkPPBz5DpN02bJojRbycId8', 'function': {'arguments': '{}', 'name': 'get_fruit_price'}, 'type': 'function'}, {'id': 'call_JK66YU69cWbwCP0FRhGMyswQ', 'function': {'arguments': '{"fruits_list": ["바나나", "사과", "딸기"]}', 'name': 'calc_total_price'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 56, 'prompt_tokens': 131, 'total_tokens': 187}, 'model_name': 'gpt-4o-2024-05-13', 'system_fingerprint': 'fp_c4e5b6fa31', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-e52c3275-22c5-46f6-9556-d4551239553c-0', tool_calls=[{'name': 'get_fruit_price', 'args': {}, 'id': 'call_xWkPPBz5DpN02bJojRbycId8'}, {'name': 'calc_total_price', 'args': {'fruits_list': ['바나나', '사과', '딸기']}, 'id': 'call_JK66YU69cWbwCP0FRhGMyswQ'}], usage_metadata={'input_tokens': 131, 'output_tokens': 56, 'total_tokens': 187}), 
  ToolMessage(content='  name  price\n0   사과    100\n1  바나나    150\n2   딸기    200', tool_call_id='call_xWkPPBz5DpN02bJojRbycId8'), 
  ToolMessage(content='450', tool_call_id='call_JK66YU69cWbwCP0FRhGMyswQ')
]
"""

최종 형태는 tool이 두 개이지만, 상황에 따라서 다를 수 있습니다.

function call 결과

이제 최종적으로 모델에 context를 돌려주는 마지막 부분입니다.

response = model_with_tools.invoke(messages)

print(response)

"""
content='### 과일 가격\n- **바나나**: 150원\n- **사과**: 100원\n- **딸기**: 200원\n\n### 총 가격\n총 가격은 450원입니다.' 
response_metadata={'token_usage': {'completion_tokens': 51, 'prompt_tokens': 249, 'total_tokens': 300}, 'model_name': 'gpt-4o-2024-05-13', 'system_fingerprint': 'fp_c4e5b6fa31', 'finish_reason': 'stop', 'logprobs': None} 
id='run-155443eb-3ff3-4bf9-815c-7094114e5e9d-0' 
usage_metadata={'input_tokens': 249, 'output_tokens': 51, 'total_tokens': 300}
"""

markdown이라 조금 알아보기 힘들 수 있지만, 대략 내용을 정리하면 다음과 같습니다.

과일 가격\n

바나나: 150원\n

사과: 100원\n

딸기: 200원\n\n

총 가격\n

총 가격은 450원입니다.

야호! GPT가 사용자와 상호 작용하며 원하는 결과를 돌려주었습니다!

그래서...뭐?

결국 GPT가 알아서 해주는게 아니네? 이게 과연 쓸모가 있는거야?

네, 맞습니다. 분명히 위와 같은 의문이 발생할 수 있습니다. 또한 RAG(Retrieval-Augmented Generation)와 같은 기술과 약간의 혼동이 발생할 가능성도 충분합니다.

그러나 function call은 분명히 다릅니다. 언제나 같은 결과를 받거나 정확한 수치 정보를 다룬다면 function call을 고려해볼 필요가 있습니다. 또한 사용자가 마치 대리자로서 결과를 제공한다는 것이 가장 큰 차이점입니다.

RAG를 생각해보죠.

사용자의 질문: "마리 퀴리의 주요 업적에 대해 알려줘."
정보 검색: 마리 퀴리의 업적에 관한 정보를 데이터베이스와 문서에서 검색합니다.
응답 생성: 검색된 정보 (예: 노벨상 수상, 방사능 연구 등)를 바탕으로 응답을 생성합니다.
최종 응답: "마리 퀴리는 폴란드 출신의 과학자로, 방사능 연구로 두 번의 노벨상을 수상했습니다. 그녀의 주요 업적에는 폴로늄과 라듐의 발견이 포함됩니다."

중간에 질문을 한 것 외에 사용자가 끼어들지 않았습니다. 그리고 최종 응답의 결과는 매번 시행마다 다를 것입니다.

반대로 function call입니다.

사용자의 질문: "서울의 날씨는 어때?"
정보 검색: 시스템은 특정 API를 호출하여 서울의 현재 날씨 정보를 가져옵니다.
응답 생성: API는 서울의 현재 온도, 습도, 날씨 상태 등을 반환합니다.
최종 응답: 시스템은 반환된 데이터를 사용자에게 전달합니다.

정보 검색 부분에서 사용자가 대리하지만, API는 항상 정해진 형식으로 데이터가 반환됩니다. 그렇기 때문에 GPT의 최종 응답을 기대할 수 있습니다. 수치만 바뀔 것이기 때문이죠.

Function Call VS RAG

function call과 RAG의 차이점을 한 눈에 보기 위한 매트릭스입니다.

LENS 응용

앞서 데보션 오픈랩 스터디에 참여했다고 말씀드렸습니다. 그리고 저는 "토이프로젝트" 주제에 참여해 갓한성님의 리드로 LENS라는 서비스를 개발하는 프로젝트를 진행했습니다.

LENS 개발 초기에는 데이터 분석가를 위한 데이터 추출 및 EDA 자동화 솔루션을 지향했지만, 프로젝트 후반에는 SQL로 데이터를 분석하는 프로 일잘러가 되고 싶은 모든 직장인을 타겟으로 고객층을 확대했습니다. 그리고 이를 위해 이미 등록된 테이블 스키마 정보, 쿼리 실행 여부 확인을 위한 함수 사용이 필요했습니다.

처음에는 프롬프트 엔지니어링 기법을 활용해서 매번 스키마 정보를 컨텍스트로 넣어주고 쿼리 실행 여부를 확인하지는 않지만 DB명과 "실행 가능해야 한다"라는 제약을 걸어 구현하는 방향으로 가닥을 잡았습니다.

하지만 이런 구현 방식은 사용자의 자유로운 응답에 있어 유연한 결과를 내어 놓지 못하는 점, 토큰 수의 비대함, 실행이 불가능한 쿼리를 검증하지 못하는 등의 치명적인 문제가 발생했습니다.

그래서 저희는 방법을 찾던 중에 갓한성님의 조언과 영웅님의 인싸력으로 function call로 이 부분을 해결할 수 있는 실마리를 얻고 곧바로 기능 개발에 착수했습니다.

최종적인 코드는 아래와 같습니다.

constraints = """
--- start of constraints ---
- 현재 데이터베이스는 MariaDB를 사용합니다. 쿼리를 생성하면 반드시 MariaDB에서 실행할 수 있는 쿼리로 생성해야 합니다.
- run_sql_query 툴을 사용하여 쿼리를 실행한 결과를 받으면, 데이터 조회 여부를 통해 실행 가능한 쿼리인지 확인해야 합니다.
- 답변에는 데이터 행을 반환하지 않고 쿼리에 대한 설명만 포함합니다.
- 쿼리에 대한 설명에는 쿼리 생성 전략, 실행 효율에 대한 평가 등을 포함해서 최소 3줄 이상으로 작성하고 절대 쿼리를 포함하지 않습니다.
- 사용자에게 답변할 때 조회한 데이터를 절대 반환하지 않습니다.
- 그 외는 자유롭게 질문에 대해 답변을 제공합니다.
--- end of constraints ---

{user_prompt}
"""

messages = [HumanMessage(constraints.format(user_prompt=user_prompt))]
chain = callLLM()

while True:
    ai_msg = chain.invoke(
        messages,
        config={"configurable": {"session_id": session_id}},
    )

    if ai_msg.tool_calls:
        new_messages = []
        for tool_call in ai_msg.tool_calls:
            selected_tool = {
                "get_table_info": get_table_info,
                "run_sql_query": run_sql_query
            }.get(tool_call["name"].lower())

            if selected_tool:
                try:
                    if inspect.iscoroutinefunction(selected_tool.invoke):
                        tool_output = await selected_tool.invoke(tool_call["args"])
                    else:
                        tool_output = selected_tool.invoke(
                            tool_call["args"])
                except Exception as e:
                    tool_output = str(e)

                if tool_call["name"].lower() == "run_sql_query":
                    sql_array.append(tool_call["args"])
                new_messages.append(ToolMessage(content=str(
                    tool_output), tool_call_id=tool_call["id"]))
            else:
                new_messages.append(ToolMessage(
                    content=f"Tool '{tool_call['name']}' not found", tool_call_id=tool_call["id"]))

        messages = new_messages
    else:
        ai_message = AIMessage(content=ai_msg.content)
        session_history = chat_history_store.get(session_id)
        if session_history:
            session_history.add_message(ai_message)
        else:
            new_history = InMemoryChatMessageHistory(messages=[ai_message])
            chat_history_store.set(session_id, new_history)
        return ai_msg.content

코드 전체는 분량으로 인해 보여드리지 못하지만, 코드의 코어 부분을 가져왔습니다.

크게 두 부분으로 해석해 보겠습니다.

선언부

constraints = &quot;&quot;&quot;
--- start of constraints ---
- 현재 데이터베이스는 MariaDB를 사용합니다. 쿼리를 생성하면 반드시 MariaDB에서 실행할 수 있는 쿼리로 생성해야 합니다.
- run_sql_query 툴을 사용하여 쿼리를 실행한 결과를 받으면, 데이터 조회 여부를 통해 실행 가능한 쿼리인지 확인해야 합니다.
- 답변에는 데이터 행을 반환하지 않고 쿼리에 대한 설명만 포함합니다.
- 쿼리에 대한 설명에는 쿼리 생성 전략, 실행 효율에 대한 평가 등을 포함해서 최소 3줄 이상으로 작성하고 절대 쿼리를 포함하지 않습니다.
- 사용자에게 답변할 때 조회한 데이터를 절대 반환하지 않습니다.
- 그 외는 자유롭게 질문에 대해 답변을 제공합니다.
--- end of constraints ---

{user_prompt}
&quot;&quot;&quot;

messages = [HumanMessage(constraints.format(user_prompt=user_prompt))]
chain = callLLM()

처음 사용자의 원본 prompt는 이미 데이터베이스에 적재한 상태로 constraints를 추가해도 실제로 뷰에 보이지 않습니다. constraints는 DB를 특정하고, 자유로운 대화 위에 데이터 관련 질문에 대한 처리 방식을 제약합니다.

그리고 messages라는 컨텍스트를 준비하고 모델을 선언합니다. callLLM시에 모델은 RunnableWithMessageHistory로 ChatOpenAI를 래핑해서 반환합니다. RunnableWithMessageHistory을 사용하기 때문에, 앞서 보여드린 예시처럼 직접 AIMessage를 삽입할 필요가 없습니다.

실행부

while True:
    ai_msg = chain.invoke(
        messages,
        config={"configurable": {"session_id": session_id}},
    )

    if ai_msg.tool_calls:
        new_messages = []
        for tool_call in ai_msg.tool_calls:
            selected_tool = {
                "get_table_info": get_table_info,
                "run_sql_query": run_sql_query
            }.get(tool_call["name"].lower())

            if selected_tool:
                try:
                    if inspect.iscoroutinefunction(selected_tool.invoke):
                        tool_output = await selected_tool.invoke(tool_call["args"])
                    else:
                        tool_output = selected_tool.invoke(
                            tool_call["args"])
                except Exception as e:
                    tool_output = str(e)

                if tool_call["name"].lower() == "run_sql_query":
                    sql_array.append(tool_call["args"])
                new_messages.append(ToolMessage(content=str(
                    tool_output), tool_call_id=tool_call["id"]))
            else:
                new_messages.append(ToolMessage(
                    content=f"Tool '{tool_call['name']}' not found", tool_call_id=tool_call["id"]))

        messages = new_messages
    else:
        ai_message = AIMessage(content=ai_msg.content)
        session_history = chat_history_store.get(session_id)
        if session_history:
            session_history.add_message(ai_message)
        else:
            new_history = InMemoryChatMessageHistory(messages=[ai_message])
            chat_history_store.set(session_id, new_history)
        return ai_msg.content

실행부는 function call 여부에 따라 처리를 위한 무한 루프입니다. function call이 있다면 예시처럼 HumanMessage + AIMessage + ToolMessage 세트를 완전하게 만들어 주고 아니면 자연스러운 대화이기 때문에 그대로 반환합니다.