Function Call 응답 처리

함수 정의

{
    "name": "get_stock_price",
    "description": "주가 조회 (Yahoo Finance 기반)",
    "parameters": {
        "properties": {
            "symbol": {
                "type": "string",
                "description": "종목 코드 (예: 'AAPL' 미국 애플, '005930.KS' 삼성전자)"
            }
        }
    }
}

tools.py

tools.py

from typing import Dict, Any

def get_stock_price(symbol: str) -> Dict[str, Any]:
    """
    주가 조회 (Yahoo Finance 기반)
    Args:
        symbol (str): 종목 코드 (예: 'AAPL' 미국 애플, '005930.KS' 삼성전자)
    Returns:
        Dict[str, Any]: 주가 정보
    """

    try:
        import yfinance as yf

        ticker = yf.Ticker(symbol)
        info = ticker.fast_info

        price = getattr(info, "last_price", None)
        currency = getattr(info, "currency", None)

        if price is None:
            hist = ticker.history(period="1d")
            if hist.empty:
                return {"ok": False, "symbol": symbol, "error": "거래 데이터 없음"}
            price = float(hist["Close"].iloc[-1])

        return {
            "ok": True,
            "symbol": symbol,
            "price": price,
            "currency": currency
        }
    except Exception as e:
        return {"ok": False, "symbol": symbol, "error": str(e)}

test.py

stream 없이stream=True

test.py

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

load_dotenv()
client = OpenAI()

input_message = "현재 삼성전자 주가 알려줘"

response = client.responses.create(
    input=input_message,
    prompt = { 
        "id": os.environ["PROMPT_ID"]
    }
)

# 응답 텍스트 출력
print(response.output_text)

# 함수 호출 응답 처리
follow_up_input = []
for output in response.output:
    if output.type == "function_call":
        try:
            import tools
            func = getattr(tools, output.name)
            args = json.loads(output.arguments)
            func_output = str(func(**args))
        except Exception as e:
            func_output = str(e)
        finally:
            print(f"\n[Function Call: {output.name}]\n{func_output}\n")
            follow_up_input.append({
                "type": "function_call_output",
                "call_id": output.call_id,
                "output": func_output
            })

# 추가 함수 호출 응답이 있으면 실행 결과를 담아서 API 재호출
if follow_up_input:
    response = client.responses.create(
        input=follow_up_input,
        previous_response_id=response.id,
        prompt = { 
            "id": os.environ["PROMPT_ID"]
        }
    )
    print(response.output_text)

test.py

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

load_dotenv()
client = OpenAI()

input_message = "현재 삼성전자 주가 알려줘"

response = client.responses.create(
    input=input_message,
    stream=True,
    prompt = { 
        "id": os.environ["PROMPT_ID"]
    }
)

# 스트림 응답 처리
follow_up_input = []
for event in response:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)
    elif event.type == "response.completed":
        previous_response_id = event.response.id
    elif event.type == "response.output_item.done":
        if event.item.type == "function_call":
            try:
                import tools
                func = getattr(tools, event.item.name)
                args = json.loads(event.item.arguments)
                func_output = str(func(**args))
            except Exception as e:
                func_output = str(e)
            finally:
                print(f"\n[Function Call: {event.item.name}]\n{func_output}\n")
                follow_up_input.append({
                    "type": "function_call_output",
                    "call_id": event.item.call_id,
                    "output": func_output
                })

# 추가 함수 호출 응답이 있으면 실행 결과를 담아서 API 재호출
if follow_up_input:
    response = client.responses.create(
        input=follow_up_input,
        previous_response_id=previous_response_id,
        stream=True,
        prompt = { 
            "id": os.environ["PROMPT_ID"]
        }
    )

    for event in response:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)
        # elif event.type == "response.completed":
        #     previous_response_id = event.response.id