본문 바로가기
Agentic AI/AI_AGENT

AI Agent(1/3): Function Calling(도구 호출) 에이전트의 정석

by 아톨 2026. 10. 3.

OpenAI의 최신 Responses API와 Open-Meteo API를 활용해, 사용자의 질문에서 도시명을 추출하고 날씨 정보를 조회해 자연어로 변환해주는 Function Calling(도구 호출) 에이전트의 정석적인 구조입니다.

전체적인 코드 흐름과 스키마 설계, 그리고 실제 동작 구조까지 핵심을 짚어 설명해 드립니다.

 

1. 전체 실행 흐름 (Execution Lifecycle)

[사용자 질문] "서울특별시의 오늘 날씨 알려줘"
       │
       ▼
[1차 LLM 호출] client.responses.create()
       │ ──> LLM 판단: "직접 답할 수 없으니 get_weather 도구를 호출하자"
       ▼
[도구 실행] get_weather(location="서울특별시")
       │ ├── 1. Geocoding API: '서울특별시' -> (위도: 37.566, 경도: 126.978)
       │ └── 2. Forecast API: 위경도 기반 날씨 데이터 조회 및 parse_weather_code() 변환
       ▼
[2차 LLM 호출] 도구 실행 결과([Tool output...])를 input_messages에 추가하여 재호출
       │
       ▼
[최종 답변 생성] "현재 서울특별시의 날씨는 맑음이며, 기온은 18.5°C..."

2. 주요 블록별 상세 설명

① 도구 정의 (tools 스키마)

LLM에게 "너는 필요할 때 이 함수를 쓸 수 있어"라고 알려주는 명세서입니다.

tools = [
    {
        "type": "function",
        "name": "get_weather",
        "description": "지정된 위치의 현재 날씨와 예보를 가져온다.",
        "parameters": { ... }
    }
]
  • JSON Schema 구조: LLM은 description과 parameters 내부의 description을 읽고 도구를 실행할지 여부 및 어떤 인자를 넘길지 판단합니다.
  • enum 지정: unit 매개변수에 ["celcius", "fahrenheit"]를 지정하여 LLM이 허용되지 않은 문자열을 생성하지 못하도록 가이드했습니다. (참고: Schema의 celcius 철자는 내부 로직의 celsius와 매칭되는지 체크하면 더 완벽해집니다.)

② 외부 API 연동 함수 (get_weather, parse_weather_code)

실제 외부에 요청을 보내고 LLM이 읽을 수 있는 형태로 데이터를 가공합니다.

# 1단계: 지명 -> 위도/경도 변환 (Geocoding API)
geo_url = f"https://geocoding-api.open-meteo.com/v1/search?name={location}&count=1&language=ko&format=json"

# 2단계: 위도/경도 -> 실시간 날씨 데이터 조회 (Forecast API)
weather_url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&current=..."
  • 2단계 API 파이프라인: Open-Meteo는 위도/경도 좌표를 요구하므로, 사용자가 입력한 "서울특별시"를 지오코딩 API로 먼저 변환한 뒤 날씨 API를 호출하는 2단계 구조를 매끄럽게 처리했습니다.
  • WMO 날씨 코드 변환: parse_weather_code()를 통해 0, 1, 61 같은 숫자로 된 WMO 코드를 "맑음", "비" 등 사람이 이해하기 쉬운 한국어 텍스트로 치환하여 모델의 이해도를 높였습니다.
  • JSON 직렬화: json.dumps(result, ensure_ascii=False)로 반환하여 한글이 깨지지 않고 LLM에게 전달되도록 구성되어 있습니다.

③ 오케스트레이션 및 루프 제어 (call_llm)

에이전트의 핵심 판단 및 메시지 관리 루프입니다. 

# 1차 호출: LLM이 Tool Call 여부 결정
response = client.responses.create(
    model=model,
    input=input_messages,
    tools=tools,
    tool_choice="auto",
    temperature=0.2,
)

# function_call 아이템 필터링
tool_calls = [item for item in response.output if item.type == "function_call"]
  • Responses API 활용: 기존 Chat Completions API와 달리 client.responses.create 및 response.output을 활용하는 최신 SDK 규격을 사용하셨습니다.
  • 메시지 컨텍스트 누적:
    1. Assistant의 도구 요청 내역(role: assistant)을 input_messages에 추가.
    2. 파이썬에서 실제 함수를 수행한 결과(role: user 또는 tool role)를 input_messages에 append.
  • 2차 재호출: 도구 실행 결과를 가지고 LLM을 재호출(second_response)하여 최종 사용자용 응답 텍스트(output_text)를 얻어냅니다.

3. 코드의 장점 & 완성도 높은 포인트

  1. 에러 핸들링: try-except 블록으로 API 통신 실패나 검색 결과 없음(not geo_res.get("results")) 상황에서 예외를 잡고, JSON 형태로 에러 메시지를 LLM에 돌려주어 LLM이 "위치 정보를 찾을 수 없습니다"라고 유연하게 답변할 수 있게 설계되었습니다.
  2. 디버깅 가시성: print("="*100) 패턴을 통해 각 단계별 API 응답, tool_calls 객체 구조, 함수 반환값 등을 실시간으로 모니터링할 수 있어 개발 및 문제 해결에 용이합니다.
  3. 섭씨/화씨 단위 변환: unit=="fahrenheit" 요청 시 자동 계산 로직까지 포함되어 있어 단순 데이터 조회를 넘어 비즈니스 로직 가공 역할을 갖췄습니다.

💡 소소한 개선 팁 (Refactoring Checkpoint)

  1. 스키마 철자 오타 확인:
    • tools 스키마 내부: "enum": ["celcius", "fahrenheit"]
    • get_weather 내부: unit == "fahrenheit" / default "celsius"
    • 스키마의 celcius를 celsius로 맞추면 LLM이 정확한 기본값을 추론하는 데 도움이 됩니다.
  2. OpenAPI Responses API의 Tool Message 형식:
    • 현재 role: user에 [Tool output for get_weather]: ... 형식으로 메시지를 넣어 완성하셨으며, 실제 동작상 잘 작동합니다.
    • Responses API 규격에 맞게 파라미터를 더욱 세밀하게 다루고 싶다면 tool_call_id 기반 전달 방식을 유지하는 것도 확장성 면에서 좋은 선택입니다.

전반적으로 OpenAI Function Calling의 원리와 파이프라인 흐름을 매우 정확하게 이해하고 작성.

 

 
    import os
    from dotenv import load_dotenv
    from openai import OpenAI
    import json
    import requests

    load_dotenv()
    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    model = os.getenv("OPENAI_MODEL", "gpt-5.4-mini")

    tools = [
        {
            "type":"function",
            "name":"get_weather",
            "description":"지정된 위치의 현재 날씨와 예보를 가져온다.",
            "parameters":{
                "type":"object",
                "properties":{
                    "location":{"type":"string","description":"도시 이름(예: 서울, Newyork...)"},
                    "unit":{"type":"string", "enum":["celcius","fahrenheit"], "default":"celsius"},
                },
            },
        },
    ]

    def get_weather(location, unit="celsius"):
        print(f"[시스템] 현재 {location}의 날씨 데이터 조회중")
        try:
            geo_url=f"https://geocoding-api.open-meteo.com/v1/search?name={location}&count=1&language=ko&format=json"
            geo_res= requests.get(geo_url, timeout=5).json()
            print("="*100)
            print(geo_res)
            print("="*100)
           
            if not geo_res.get("results"):
                return json.dumps({"error":f"'{location}의 위치정보를 찾을 수 없습니다.."})
           
            lat = geo_res["results"][0]["latitude"]
            lon = geo_res["results"][0]["longitude"]
            location_name = geo_res["results"][0].get("name",location)
            print("="*100)
            print("lat, lon, location_name: ", lat, lon, location_name)
            print("="*100)
           
            weather_url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&current=temperature_2m,relative_humidity_2m,weather_code&timezone=auto"
            w_res = requests.get(weather_url, timeout=5).json()
            # print(w_res)
           
            current = w_res.get("current",{})
            temp = current.get("temperature_2m")
            humidity = current.get("relative_humidity_2m")
            code = current.get("weather_code")
           
            if unit=="fahrenheit" and temp is not None:
                temp = round((temp*9/5)+32, 1)
                temp_unit_str = "°F"
            else:
                temp_unit_str = "°C"
           
            condition = parse_weather_code(code)
            result = {
                "location":location_name,
                "condition":condition,
                "temperature":f"{temp}{temp_unit_str}",
                "humidity":f"{humidity}%"
            }
            print("="*100)
            print(result)
            print("="*100)
            return json.dumps(result, ensure_ascii=False)
       
        except Exception as e:
            return json.dumps({"error":f"날씨 정보를 가져오는중 에러발생: {str(e)}"}, ensure_ascii=False)
       
    def parse_weather_code(code):
        if code == 0:
            return "맑음"
        elif code in [1,2,3]:
            return "구름 조금 / 흐림"
        elif code in [45, 48]:
            return "안개"
        elif code in [51,53,55,61,63,65,80,81,82]:
            return "비"
        elif code in [71,73,75,77,85,86]:
            return "눈"
        elif code in [95,96,99]:
            return "뇌우"
        return "정보없음"    

    available_functions = {"get_weather":get_weather}

    def call_llm(system_prompt, user_prompt):
        input_messages = [
            {"role":"system", "content":system_prompt},
            {"role":"user", "content":user_prompt},
        ]
       
        response = client.responses.create(
            model=model,
            input=input_messages,
            tools=tools,
            tool_choice="auto",
            temperature=0.2,
        )
       
        tool_calls = [item for item in response.output if item.type=="function_call"]

        print("="*100)
        print("response: ",response)
        print("="*100)    
        print("tool_calls: ",tool_calls)
        print("="*100)  

        if tool_calls:
            input_messages.append(
                {
                    "role":"assistant",
                    "content":response.output_text or "도구 호출 요청",
                }
            )
            for tool_call in tool_calls:
                function_name= tool_call.name
                function_to_call=available_functions[function_name]
                function_args = json.loads(tool_call.arguments)
               
                function_response = function_to_call(
                    location = function_args.get("location"),
                    unit=function_args.get("unit","celsius")
                )
                print("="*100)
                print("function_response: ",function_response)
                print("="*100)
                input_messages.append(
                    {
                        "role":"user",
                        "content":f"[Tool output for {function_name}]: {function_response}",
                    }
                )
            second_response = client.responses.create(
                model=model,
                input=input_messages,
            )
            print("="*100)
            print("second_response: ",second_response)
            print("="*100)  
            return second_response.output_text
        return response.output_text

    def main():
        system_ask = "당신은 친절하고 정확한 날씨 안내 도우미입니다."
        user_ask = "서울특별시의 오늘 날씨 알려줘."

        answer_5 = call_llm(system_ask,user_ask)
        print("="*100)
        print("질문: ", user_ask)
        print("="*100)
        print("답변: ", answer_5)
        print("="*100)

    if __name__ =="__main__":
        main()
       
반응형
저작자표시 비영리 변경금지 (새창열림)

'Agentic AI > AI_AGENT' 카테고리의 다른 글

AI Agent(3/3): MCP(Model Context Protocol) 서버와 MCP 클라이언트(LLM 에이전트)  (0) 2026.10.03
AI Agent(2/3): 다단계 도구 호출(Multi-step Tool Calling) 및 자율적 루프(Autonomous Re-act Loop)  (0) 2026.10.03
XVII. AI Agent Engine 기본 완성: Python · Agent Loop · Tool Dispatcher · Function Call · Context Block · History Trim  (0) 2026.08.30
XVI. 작은 AI Agent에 5가지 기능을 추가한 프로그램(끝)  (0) 2026.08.26
XV. 작은 AI Agent 프로그램(실제 LLM -OpenAI 사용)  (0) 2026.08.26

관련글

티스토리툴바