어제 휴가로 인한 보충이론 기록 start
Prompt Engineering
Role 역할, Instruction 지시, Context 맥락, Format 형식.
위 4개를 채울수록 결과가 안정된다. 역할지시 맥락은 system_instruction에 데이터는 contents에 넣는다. 맥락에 모르면 모른다는 안정장치를 넣으면 환각이 줄어든다.
shot 예시는 너무 적으면 모델이 기준을 못잡고 너무 많으면 토큰 비용이 늘고 효과는 점점 줄어든다.
json으로만 답하라고프롬프트로 명령할 수도 있고 response_mime_type으로 강제할 수 있다 .후자가 더 안전하다.
response = client.responses.create(
model="gpt-5",
input="""
다음 정보를 JSON으로만 출력하세요.
이름: 홍길동
나이: 25
설명은 하지 마세요.
"""
)
print(response.output_text)response = client.responses.create(
model="gpt-5",
input="홍길동(25세)을 JSON으로 변환해.",
response_format={
"type": "json_object"
}
)
print(response.output_text)
response_schema에 pydantic 모델이나 스키마를 주면 json의 키이름과 타입까지 강제할 수 있다. 즉 name은 문자열, age는 정수 라는 것까지 모델이 지키도록 한다.
response = model.generate_content(
"홍길동은 25살이다.",
generation_config={
"response_mime_type": "application/json",
"response_schema": Person
}
)
비교 프롬프트 <<< >>> 구분자 방어전과 방어후 데이터를 명확하게 분리해서 LLM이 각각을 비교하도록 하는 역할
LLM은 <<< >>> 안의 내용을 하나의 덩어리로 인식하고 앞을 블록과 뒤 블록을 비교한다.
prompt = f"""
다음 두 결과를 비교하세요.
방어 전
<<<
{before}
>>>
방어 후
<<<
{after}
>>>
변경된 점을 설명하세요.
"""
어제 휴가로 인한 보충이론 기록 end
Reasoning Prompt
LLM은 계산기가 아니라 다음 단어를 예측하는 모델로 한번에 답을 뱉질 못해 다단계 계산을 자주 틀리며 '단계적으로 풀어라' chain of Thougtht가 정답률을 크게 끌어올린다. LLM은 다음에 올 가장 그럴듯한 단어(숫자)를 예측하는것이지 계산을 하는 것이 아니라 한번에 최종숫자를 뱉으라고 하면 중간과정을 건너뛰고 그럴듯한 숫자를 찍는다.
Chain of Thought 생각의 사슬 CoT
프롬프트에 단계적으로 풀어라라는 한줄을 더하라. 중간풀이를 글로 쓰게하면 그 풀이가 다음 토큰을 만들때 작업공간이자 근거가 되어 다음줄에서 그 숫자를 보고 계산을 이어갈 수 있다. 암산이 필산이 되는 셈.
CoT는 토큰을 더 쓰지만 계산, 다단계 문제의 정답률이 높다. 토큰 즉 비용과 시간을 쓰되, 대가로 정확도를 얻는것.
직접답변 혹은 CoT의 정답률을 비교했을때 단순한 질문이거나 함정문제일 경우 효과가 없거나 오히 려 해로운 경우도 있다. 만능이 아니라는 점을 이해하자.
response = client.responses.create(
model="gpt-5",
input="""
487 × 236을 계산해.
단계적으로 풀고 마지막에 최종 답을 제시해.
"""
)
print(response.output_text)
self-confistency
CoT가 정답률을 높이지만 100%은 아니며 단계가 많거나 숫자가 큰 문제에서 더 믿음직하게 하기위해 자기일관성기법, 여러번 풀어 일관되게 답이 나왔을때 가장 많이 나온 답. 다수결을 채택한다. 이는 비용이 배로 든다는 단점이 있다. 핵심은 이전 풀이를 통째로 보여주며 이게 맞는지 다시 계산하라고 시키는 것. verfy() 프롬프트 기법.
현업 에이전트에서 민감한 계산에 검사한번을 기본 정책으로 둔다 비용을 조금 더 내고 사고를 예방하는 것. 보험과 같은 개념.
answers = []
for _ in range(5):
answer = llm(
"487 × 236을 단계적으로 계산하고 최종 답을 알려줘."
)
answers.append(answer)
final_answer = majority_vote(answers)
print(final_answer)answer = llm(
"487 × 236을 단계적으로 계산해."
)
verified = llm(f"""
다음 풀이를 검토해.
{answer}
계산 오류가 있는지 다시 확인하고,
틀렸다면 수정해.
""")
print(verified)
critical
검산은 호출을 한번 더 함으로 토큰 비용이 약 2배이며 모든 질문에 쓰지않고 전략적으로 쓴다. 이 계산이 틀리면 큰일이 나는지를 확인하는것. 환불액, 발주량 같은것에 true, 단순안내는 false를 한다는 것.
def is_critical(task):
critical_tasks = ["환불", "발주", "정산", "세금"]
return any(keyword in task for keyword in critical_tasks)
answer = llm(user_question)
if is_critical(user_question):
answer = verify(answer)
print(answer)
function calling
"이 함수를 인자로 호출하세요"
LLM은 함수를 직접 실행하지못한다 다만 이 함수를 이런 인자로 불러달라는 json 요청을 만들 수 있다. LLM은 토큰을 생성하는 기계일 뿐 파이썬 코드를 돌릴 능력이 없다는 것을 이해하자.
LLM은 함수본문 코드를 볼 수 없고 함수이름, 독스트링, 인자타입만 보고 도구사용을 판단함으로 독스트링을 명확하게 쓰는 것이 도구 성능을 좌우한다.
def funtion():
"""독스트링"""
function calling - 수동루프
사용자 요청 - LLM JSON 생성 - 개발자가 JSON으로 함수실행 - 결과 LLM에게 전달 - LLM 최종응답
from openai import OpenAI
import json
client = OpenAI()
# 1. LLM에게 요청
response = client.responses.create(
model="gpt-5",
input="서울 날씨 알려줘."
)
# 2. LLM이 함수 호출 JSON 생성
tool_call = response.output[0]
args = json.loads(tool_call.arguments)
# 3. 개발자가 함수 실행
result = get_weather(args["city"])
# 4. 결과를 다시 LLM에 전달
final = client.responses.create(
model="gpt-5",
input=[
tool_call,
{
"type": "function_call_output",
"call_id": tool_call.call_id,
"output": result
}
]
)
print(final.output_text)
function calling - 자동실행
최신 OpenAI API에서는 이 과정을 SDK가 자동으로 처리할 수 있다.
사용자 요청 - LLM JSON 생성 - SDK 자동으로 함수실행 - SDK가 결과 LLM에게 전달 - LLM 최종응답
response = client.responses.create(
model="gpt-5",
tools=[get_weather],
input="서울 날씨 알려줘."
)
print(response.output_text)
오류처리
없는 상품, 없는 통화, db다운등에 오류를 메세지로 바꿔 모델에게 돌려주면 에이전트를 멈추지않고 게속해서 사용자에게 안내할 수 있다. 해피패스 Happy path 모든게 잘 되는 경우만 처리하는 에이전트는 장난감과 다름이 없다. 실패를 우아하게 다뤄야하며 예외를 친절한 메세지로 변환하는 패턴은 모든 에이전트에서 반복된다.
def get_price(product):
if product not in products:
return {
"success": False,
"error": "상품을 찾을 수 없습니다."
}
return {
"success": True,
"price": 15000
}죄송합니다. 요청하신 상품을 찾을 수 없습니다. 상품명을 다시 확인해 주세요.
Tool System
LLM이 필요할때 호출하는 외부기능 함수, API, DB, 검색등. 계산, 인터넷검색, 데이터베이스 조회, 메일 발송 등을 직접할 수 없기 때문에 Tools를 이용해 필요한 작업을 수행한다. 실제서비스에서는 Tools가 4~5개가 아니라 수십개일 수 있으며 Ai비서라면더 많을 것이다. 에이전트는 이 tools 리스트에서 스스로 판단해 순서대로 호출한뒤 자연스러운 답변을 생성한다. Agents가 여러 Tools를 오케스트레이션 조합 관리해 작업을 수행하는 방식
| Search | 인터넷 검색 |
| Database | DB 조회 |
| Weather API | 날씨 조회 |
| Calculator | 계산 수행 |
| 메일 발송 | |
| File | 파일 읽기/쓰기 |
| Function | 개발자가 만든 함수 호출 |
from openai import OpenAI
client = OpenAI()
# Tool 1 - 날씨
def get_weather(city: str):
return f"{city} 현재 기온은 29도입니다."
# Tool 2 - 계산기
def calculator(expression: str):
return eval(expression)
# Tool 3 - 주문 조회
def get_order(order_id: int):
return {
"order_id": order_id,
"status": "배송중"
}
# Tool 4 - 상품 조회
def get_product(product_name: str):
return {
"name": product_name,
"price": 25000
}
response = client.responses.create(
model="gpt-5",
input="""
서울 날씨를 알려주고,
123*456을 계산한 뒤,
1024번 주문 상태를 조회하고,
에어팟 가격도 알려줘.
""",
tools=[
get_weather,
calculator,
get_order,
get_product
]
)
print(response.output_text)
if문으로 하드코딩해 직접 정해주는 방식은 if문이 늘고, 표현이 조금만 달라도 (얼마, 값이어때, 가격은?) 놓치거나 유지보수 지옥이 펼쳐진다 if하드코딩이아니라 설명을 잘 쓰는것이 진짜 해법이다. 앞선 독스트링. 설명이 너무 짧거나 입력값, 반환값이 명시되어있지않으면 나쁜 독스트링이다 . 무엇을 하는 함수인지, 매개변수 Args의 의미와 예시, 반환값 Returns의 형태를 설명하고필요시 언제 사용해야 하는 지를 적는것이 좋은 Docstring 작성원칙.
def get_weather(city):
"""날씨"""def get_weather(city: str) -> dict:
"""
지정한 도시의 현재 날씨를 조회합니다.
Args:
city (str): 조회할 도시명 (예: 서울, 부산)
Returns:
dict:
- city (str): 도시명
- temperature (float): 현재 기온(℃)
- condition (str): 날씨 상태
"""def calculator(expression: str) -> float:
"""
수학식을 계산합니다.
Args:
expression (str):
계산식
예) "123+456", "15*8", "100/5"
Returns:
float: 계산 결과
"""
좋은 도구 설계원칙
한도구는 한가지일, 이름이 곧 기능이 되도록, 인자가 명확하고, 설명이 구체적인, 반환이 일관적이라는 원칙을 가지면 모델이 도구를 정확히 고르고 올바르게 쓸 수 있다 . 특히 설명이 비슷하게 겹치는 두 도구가 있다면 모델을 어느것을 써야할지 헷갈린다. 입력형태, id 키워드와 결과건수 1건 여러건을 독스트링에 명확히 못 박아 비슷한 도구도 또렷이 구분되게해야한다.
독스트링 점검 체크리스트
- 무엇을 하는가
- 언제 쓰는가
- 인자에 무엇을 넣어야하는지 알 수 있는가
- 비슷한 다른 도구와 구별되는가
Agent Loop
한번 묻고 끝나지 않는 일, 현업의 일은 보통 여러 단계이다. 행동 - 결과보고 - 다음행동 의 반복하는 구조가 Agent Loop.
one-shot 질문하나, 도구하나, 답하나 깔끔한 원샷은 복합적인 질문에서 한번에 끝나질못해 결과를 봐야 행동을 이어할 수가있다. Agent Loop는 단계적 추론 CoT, 도구호출의 Function Calling, 여러도구중 선택하는 Tool 시스템, 결정 - 실행 - 결과전달의 수동루프가 모여있는것을 while루프로 묶는.
사용자 요청
↓
Reasoning
(무엇을 해야 하지?)
↓
Tool 선택
↓
Function Calling
↓
결과 확인
↓
아직 해야 할 일이 있는가?
↓
Yes ───────────────┐
↓ │
다음 Tool 호출 │
↓ │
결과 확인───────────┘
↓
No
↓
최종 답변while True:
# 1. LLM이 다음 행동 결정
action = llm(messages, tools)
# 2. 더 이상 Tool이 필요 없으면 종료
if action.type == "final_answer":
print(action.content)
break
# 3. Tool 실행
result = run_tool(action.name, action.arguments)
# 4. 결과를 LLM에게 전달
messages.append({
"role": "tool",
"content": result
})
ReAct 패턴
Reason, Act의 합성어, 모델이 Thought, Action, Obervation을 거치고 관찰결과를 다음입력으로 되돌려 반복한다. While True는 무한루프의 위험이있고 토큰을 끝없이 태움으로 while True 대신 MAX_STEPS, max_step을 둔다 . 보통 5~10을 두나 작업의 성격에 맞게 정한다. 재고확인 같은 2~3 스탭에는 6정도가 넉넉하다. 추가로 모델이 진전없이 도구명 인자를 여러번 수행하지않도록 중복 호출을 감지한다.
MAX_STEPS = 6
messages = [{"role": "user", "content": user_question}]
# 중복 호출 감지용
visited = set()
for step in range(MAX_STEPS):
# 1. LLM이 다음 행동 결정
response = llm(messages, tools)
# 2. 최종 답변이면 종료
if response.type == "final_answer":
print(response.content)
break
# 3. 같은 Tool + 같은 인자 반복 호출 방지
tool_key = (
response.tool_name,
tuple(sorted(response.arguments.items()))
)
if tool_key in visited:
print("동일한 Tool을 반복 호출하여 종료합니다.")
break
visited.add(tool_key)
# 4. Tool 실행
result = run_tool(
response.tool_name,
response.arguments
)
# 5. Observation 추가
messages.append({
"role": "tool",
"content": result
})
else:
print("최대 반복 횟수를 초과했습니다.")
Trimming과 Conversation Summarization
ReAct루프는 매 스탭마다 관찰결과를 history에 덧붙임으로 메세지와 토큰이 게속 불어난다. 이는 비용, 지연이 됨으로 오래된 기록을 잘라내는 트리밍이 필요하다. 이때 첫 질문 즉 원래 목표는 반드시 보존해 모델이 목표를 잊지않도록 한다. 오래된 중간 기록은 대개 최종 판단에 덜 주요함으로 최근 몇개만 남기고 잘라낸다. 더 정교한 방법으로는 오래된 대화를 요약해서 압축하는 기법도 있다.
MAX_STEPS = 6
MAX_HISTORY = 10 # 최근 10개 메시지만 유지
messages = [
{"role": "user", "content": user_question}
]
for step in range(MAX_STEPS):
# LLM 호출
action = llm(messages)
# 최종 답변이면 종료
if action.type == "final_answer":
print(action.content)
break
# Tool 실행
result = run_tool(
action.tool_name,
action.arguments
)
# Observation 추가
messages.append({
"role": "tool",
"content": result
})
# History Trimming
if len(messages) > MAX_HISTORY:
messages = [messages[0]] + messages[-9:]MAX_HISTORY = 10
if len(messages) > MAX_HISTORY:
# 오래된 대화
old_messages = messages[:-5]
# 최근 대화
recent_messages = messages[-5:]
# 오래된 내용을 LLM으로 요약
summary = llm(
"다음 대화를 핵심만 5줄 이내로 요약해줘.",
old_messages
)
# 요약 + 최근 대화만 유지
messages = [
{
"role": "system",
"content": f"대화 요약:\n{summary}"
}
] + recent_messages
프로젝트 cot_console_project를 분석해보자.
common.py
import os, patlib, dotenv
path 지정, dotenv값으로 model지정하기
require_key()
한경변수 키가 없으면 raise systemexit()
get_genai_client()
google genai 불러와 genai.Clent() api key로 클라이언트 만들기
get_chat()
provider에 따라 lanchain)google_gemai 를 import해 chatmodel return.
get_embeddings()
provider가 gemini일때 langchain_google_gemai를 import해 embeddings 객체 return. openai도.
__name __main()
각 정보 절리해서 출력
# -*- coding: utf-8 -*-
"""
common.py — 모든 실습 공통 파일
목적:
- .env 의 API 키를 한 곳에서 로드한다.
- Gemini(주력) / OpenAI(보조) 모델 객체를 일관되게 생성한다.
- 실습 데이터(data/) 경로를 쉽게 찾는다.
각 강의 실습 코드 맨 위에서 다음처럼 불러 씁니다.
from common import get_chat, get_genai_client, DATA
"""
import os
import pathlib
from dotenv import load_dotenv
# 프로젝트 루트
ROOT = pathlib.Path(__file__).resolve().parent.parent
DATA = ROOT / "data"
DOCS = DATA / "docs"
# .env 로드 (루트의 .env 를 읽음)
load_dotenv(ROOT / ".env")
GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-flash")
GEMINI_EMBED_MODEL = os.getenv("GEMINI_EMBED_MODEL", "models/gemini-embedding-001")
def require_key(name: str) -> str:
"""환경변수 키가 없으면 종료."""
val = os.getenv(name)
if not val or val.startswith("여기에"):
raise SystemExit(
f"[설정 필요] {name} 가 .env 에 없습니다.\n"
f" 1) cp .env.example .env\n"
f" 2) .env 파일을 열어 {name} 값을 채우세요."
)
return val
# ---------- raw SDK (원리 학습용) ----------
def get_genai_client():
"""google-genai 클라이언트 (from google import genai)."""
from google import genai
return genai.Client(api_key=require_key("GOOGLE_API_KEY"))
# ---------- LangChain Chat 모델 (현업용) ----------
def get_chat(provider: str = "gemini", temperature: float = 0.0):
"""LangChain ChatModel 반환. provider: 'gemini'(기본) | 'openai'."""
if provider == "gemini":
require_key("GOOGLE_API_KEY")
from langchain_google_genai import ChatGoogleGenerativeAI
return ChatGoogleGenerativeAI(model=GEMINI_MODEL, temperature=temperature)
elif provider == "openai":
require_key("OPENAI_API_KEY")
from langchain_openai import ChatOpenAI
return ChatOpenAI(model="gpt-4o-mini", temperature=temperature)
raise ValueError(f"알 수 없는 provider: {provider}")
def get_embeddings(provider: str = "gemini"):
"""LangChain Embeddings 반환."""
if provider == "gemini":
require_key("GOOGLE_API_KEY")
from langchain_google_genai import GoogleGenerativeAIEmbeddings
# gemini-embedding-001 기본 출력은 3072차원. 실습 저장·속도를 위해 768로 고정.
# (768/1536/3072 중 선택 가능)
return GoogleGenerativeAIEmbeddings(
model=GEMINI_EMBED_MODEL, output_dimensionality=768)
elif provider == "openai":
require_key("OPENAI_API_KEY")
from langchain_openai import OpenAIEmbeddings
return OpenAIEmbeddings(model="text-embedding-3-small")
raise ValueError(f"알 수 없는 provider: {provider}")
if __name__ == "__main__":
print("ROOT :", ROOT)
print("DATA :", DATA, "(존재:", DATA.exists(), ")")
print("GEMINI_MODEL :", GEMINI_MODEL)
print("키 로드 상태 — GOOGLE_API_KEY:", bool(os.getenv("GOOGLE_API_KEY")),
"/ OPENAI_API_KEY:", bool(os.getenv("OPENAI_API_KEY")))
llm_client.py
import os, re
common 에서 필요한 함수를 가져온다.
extract_number()
입력받은 str에 re.findall \d 정수형태의 숫자를 모두 찾아 text.replace() 공백을 제거한다. 없으면 none을 리턴, 있다면 콤마를 제거하고 int로 반환.
ask_gemini_direct()
gemai 에서 types import.
client를 만들어 response =client.models.generate_content() strip() 앞뒤 공백을 제거해 return
ask_gemini_cot
맟찬가지로 client를 만든뒤contents에 단계적으로 풀라, 맨마지막줄에 숫자형식으로 답하라고 추가해 return.
verify_gemini()
마찬가지로 client를 만든뒤 문제와 제출한 풀이를 가져다가 위풀이가 맞는지를 다시 계산하라고 명명을 추가해 strip() 앞뒤 공백을 제거해 return.
get_openai_client()
api키가 없으면 common.py의 안내메세지가 호출되며 return key
ask_openai_direct() 위 과정을 openai로 수행, ask_openai_cot(), verify_openai()
# -*- coding: utf-8 -*-
"""
llm_clients.py
역할:
- common.py의 API 키 로딩 함수와 모델 상수를 사용합니다.
- Gemini API와 OpenAI API로 직접 답변, CoT 답변, 자기검증을 실행합니다.
- API 키가 없을 때 프로그램 전체가 중단되지 않도록 메뉴 단위로 예외를 처리합니다.
"""
# os는 .env에서 불러온 모델명 같은 환경변수를 읽기 위해 사용합니다.
import os
# re는 LLM 답변에서 마지막 숫자를 추출하기 위해 사용합니다.
import re
# common.py에서 Gemini 클라이언트 생성 함수와 모델명, 키 확인 함수를 가져옵니다.
from common import GEMINI_MODEL, get_genai_client, require_key
def extract_number(text: str) -> int | None:
"""LLM 답변 문자열에서 마지막 숫자를 정수로 추출합니다."""
# 공백을 제거한 뒤 정규식으로 정수 형태의 숫자를 모두 찾습니다.
nums = re.findall(r"-?\d[\d,]*", text.replace(" ", ""))
# 숫자가 하나도 없으면 None을 반환하여 추출 실패를 표현합니다.
if not nums:
return None
# 마지막 숫자에서 콤마를 제거하고 int로 변환합니다.
return int(nums[-1].replace(",", ""))
def ask_gemini_direct(question: str) -> str:
"""Gemini API로 중간 설명 없이 최종 숫자만 요청합니다."""
# google-genai의 설정 타입을 함수 내부에서 import하여, API 메뉴 실행 시점에만 필요하게 합니다.
from google.genai import types
# common.py의 get_genai_client()를 사용해 Gemini 클라이언트를 생성합니다.
client = get_genai_client()
# Gemini 모델에 직접 답변 프롬프트를 보냅니다.
response = client.models.generate_content(
model=GEMINI_MODEL,
contents=f"{question}\n설명 없이 최종 숫자만 답하라.",
config=types.GenerateContentConfig(temperature=0),
)
# 응답 텍스트를 반환합니다.
return response.text.strip()
def ask_gemini_cot(question: str) -> str:
"""Gemini API로 단계적 풀이 후 마지막 줄에 정답을 요청합니다."""
# google-genai 설정 타입을 가져옵니다.
from google.genai import types
# Gemini 클라이언트를 생성합니다.
client = get_genai_client()
# CoT 프롬프트를 사용해 단계적 풀이를 요청합니다.
response = client.models.generate_content(
model=GEMINI_MODEL,
contents=(
f"{question}\n"
"단계적으로 풀어라. 각 계산을 한 줄씩 쓰고, "
"맨 마지막 줄에 '정답: <숫자>' 형식으로 답하라."
),
config=types.GenerateContentConfig(temperature=0),
)
# 응답 텍스트를 반환합니다.
return response.text.strip()
def verify_gemini(question: str, cot_answer: str) -> str:
"""Gemini API로 CoT 풀이를 다시 검산합니다."""
# google-genai 설정 타입을 가져옵니다.
from google.genai import types
# Gemini 클라이언트를 생성합니다.
client = get_genai_client()
# 문제와 1차 풀이를 함께 제공하여 검산을 요청합니다.
response = client.models.generate_content(
model=GEMINI_MODEL,
contents=(
f"[문제] {question}\n"
f"[제출한 풀이]\n{cot_answer}\n\n"
"위 풀이가 맞는지 다시 계산해 검산하라. "
"틀렸다면 올바른 값을, 맞다면 그대로 '정답: <숫자>'로 확정하라."
),
config=types.GenerateContentConfig(temperature=0),
)
# 검산 결과 텍스트를 반환합니다.
return response.text.strip()
def get_openai_client():
"""common.py의 require_key()로 API 키를 확인한 뒤 OpenAI 클라이언트를 생성합니다."""
# OpenAI 패키지는 OpenAI 메뉴를 실행할 때만 필요하므로 함수 내부에서 import합니다.
from openai import OpenAI
# OPENAI_API_KEY가 없으면 common.py의 안내 메시지와 함께 SystemExit이 발생합니다.
api_key = require_key("OPENAI_API_KEY")
# 확인된 API 키로 OpenAI 클라이언트를 생성합니다.
return OpenAI(api_key=api_key)
def ask_openai_direct(question: str) -> str:
"""OpenAI API로 중간 설명 없이 최종 숫자만 요청합니다."""
# OpenAI 클라이언트를 생성합니다.
client = get_openai_client()
# .env의 OPENAI_MODEL 값을 사용하고, 없으면 gpt-4o-mini를 기본값으로 사용합니다.
model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
# OpenAI Chat Completions API를 호출합니다.
response = client.chat.completions.create(
model=model,
temperature=0,
messages=[
{"role": "system", "content": "너는 계산 문제에 답하는 도우미다."},
{"role": "user", "content": f"{question}\n설명 없이 최종 숫자만 답하라."},
],
)
# 첫 번째 선택지의 메시지 내용을 반환합니다.
return response.choices[0].message.content.strip()
def ask_openai_cot(question: str) -> str:
"""OpenAI API로 단계적 풀이 후 마지막 줄에 정답을 요청합니다."""
# OpenAI 클라이언트를 생성합니다.
client = get_openai_client()
# 사용할 OpenAI 모델명을 환경변수에서 가져옵니다.
model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
# 단계적 풀이를 요청하는 메시지를 보냅니다.
response = client.chat.completions.create(
model=model,
temperature=0,
messages=[
{"role": "system", "content": "너는 계산 과정을 정확히 작성하는 도우미다."},
{
"role": "user",
"content": (
f"{question}\n"
"단계적으로 풀어라. 각 계산을 한 줄씩 쓰고, "
"맨 마지막 줄에 '정답: <숫자>' 형식으로 답하라."
),
},
],
)
# 응답 텍스트를 반환합니다.
return response.choices[0].message.content.strip()
def verify_openai(question: str, cot_answer: str) -> str:
"""OpenAI API로 CoT 풀이를 다시 검산합니다."""
# OpenAI 클라이언트를 생성합니다.
client = get_openai_client()
# 사용할 모델명을 가져옵니다.
model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
# 문제와 기존 풀이를 함께 제공하여 검산을 요청합니다.
response = client.chat.completions.create(
model=model,
temperature=0,
messages=[
{"role": "system", "content": "너는 제출된 계산 풀이를 검산하는 도우미다."},
{
"role": "user",
"content": (
f"[문제] {question}\n"
f"[제출한 풀이]\n{cot_answer}\n\n"
"위 풀이가 맞는지 다시 계산해 검산하라. "
"틀렸다면 올바른 값을, 맞다면 그대로 '정답: <숫자>'로 확정하라."
),
},
],
)
# 검산 결과 텍스트를 반환합니다.
return response.choices[0].message.content.strip()
torch_metrics.py
import pd, torch
common에서 data함수 가져오기
load_math_data()
csv_path를 정의해 없으면 error,
pd.read_csv() 수행해 dataframe으로 return
calculate_accuracy_with_torch()
정답, 예측값을 torch.tensor로 변환 두 크기가 다른지 ,shape을 확인해 다를경우 경고.
correct_tensor 각 위치가 맞는지 true false tensor만들기 이 correct_tensor를 float 숫자로 바꾼뒤 .mean() 평규능ㄹ 구하면 accuracy가 된다 return
run_offline_accuracy_demo()d
csv데이터를 읽어 컬럼을 정수리스트로 변환하고 직접답변 예측 예시리스트를 만든다. calculate_accuracy_with_torch() 답변 정확도를 계산해 변수에 할당, cot_acc도 마찬가지
for문으로 df.iterrows()를 돌려 현재 문제의 정답을 가져오고 답변이 맞았다면 x표시. CoT도 마찬가지., 결과출력, 정확도도 백분률로 출력.
# -*- coding: utf-8 -*-
"""
torch_metrics.py
역할:
- API 호출 없이도 CoT 실습의 핵심인 '정답률 계산'을 확인합니다.
- pandas로 CSV 데이터를 읽고 torch Tensor로 정답률을 계산합니다.
"""
# pandas는 CSV 파일을 읽고 표 형태 데이터를 다루기 위해 사용합니다.
import pandas as pd
# torch는 정답 배열과 예측 배열을 Tensor로 바꾸고 정확도를 계산하기 위해 사용합니다.
import torch
# common.py의 DATA 경로를 공통 데이터 폴더로 사용합니다.
from common import DATA
def load_math_data() -> pd.DataFrame:
"""쇼핑 계산 문제 CSV를 pandas DataFrame으로 읽어 반환합니다."""
# math_word_problems.csv 파일 경로를 생성합니다.
csv_path = DATA / "math_word_problems.csv"
# 파일이 없으면 명확한 오류 메시지를 발생시킵니다.
if not csv_path.exists():
raise FileNotFoundError(f"CSV 파일을 찾을 수 없습니다: {csv_path}")
# CSV 파일을 읽어서 DataFrame으로 반환합니다.
return pd.read_csv(csv_path, encoding="utf-8")
def calculate_accuracy_with_torch(gold_answers: list[int], predictions: list[int]) -> float:
"""정답 리스트와 예측 리스트를 torch Tensor로 변환하여 정확도를 계산합니다."""
# 정답 값을 정수 Tensor로 변환합니다.
gold_tensor = torch.tensor(gold_answers, dtype=torch.long)
# 예측 값을 정수 Tensor로 변환합니다.
pred_tensor = torch.tensor(predictions, dtype=torch.long)
# 두 Tensor의 크기가 다르면 비교할 수 없으므로 오류를 발생시킵니다.
if gold_tensor.shape != pred_tensor.shape:
raise ValueError("정답 개수와 예측 개수가 서로 다릅니다.")
# gold_tensor == pred_tensor는 각 위치가 맞았는지 True/False Tensor를 만듭니다.
correct_tensor = gold_tensor == pred_tensor
# float()로 True/False를 1.0/0.0으로 바꾼 뒤 평균을 구하면 정확도가 됩니다.
accuracy = correct_tensor.float().mean().item()
# 파이썬 float 값으로 반환합니다.
return accuracy
def run_offline_accuracy_demo() -> None:
"""API 없이 고정 예측값으로 직접 답변과 CoT의 정답률 비교를 시연합니다."""
# CSV 데이터를 읽습니다.
df = load_math_data()
# answer 컬럼을 정수 리스트로 변환합니다.
gold = df["answer"].astype(int).tolist()
# 직접 답변 예측 예시입니다. 일부러 몇 개는 틀리게 넣어 정답률 차이를 보여 줍니다.
direct_pred = [213000, 89000, 44000, 107000, 80000, 160000, 168000, 135000]
# CoT 예측 예시입니다. 단계적 풀이를 했을 때 정답이 나온 상황을 가정합니다.
cot_pred = gold.copy()
# torch로 직접 답변 정확도를 계산합니다.
direct_acc = calculate_accuracy_with_torch(gold, direct_pred)
# torch로 CoT 정확도를 계산합니다.
cot_acc = calculate_accuracy_with_torch(gold, cot_pred)
# 결과 표 제목을 출력합니다.
print("\n[Torch 기반 정답률 계산]")
print("-" * 80)
# 각 문제의 정답과 두 방식의 예측을 출력합니다.
for idx, row in df.iterrows():
# 현재 문제의 정답을 가져옵니다.
ans = int(row["answer"])
# 직접 답변이 맞았는지 표시합니다.
d_mark = "O" if direct_pred[idx] == ans else "X"
# CoT 답변이 맞았는지 표시합니다.
c_mark = "O" if cot_pred[idx] == ans else "X"
# 한 줄 결과를 출력합니다.
print(
f"{row['problem_id']} 정답={ans:>7} | "
f"직접={direct_pred[idx]:>7} {d_mark} | "
f"CoT={cot_pred[idx]:>7} {c_mark}"
)
# 전체 정확도를 백분율로 출력합니다.
print("-" * 80)
print(f"직접 답변 정답률: {direct_acc:.0%}")
print(f"CoT 정답률 : {cot_acc:.0%}")
main.py
import
traceback 예외발생시 상세원인을 확인하고 싶을때 사용
common에서 필요한 함수들 가져오기
llm_clients 모듈에서 gemini, openai 호출 함수 가져오기
torch_metrics 모듈에서 API없는 정답률 계산 데모함수를 가져오기
sample 질문 정의
print_project_info()
프로젝트 정보 출력
run_gemini_demo()
실습문제를 출력하고, ask_gemini_direct() 직접 답변을 요청하고, ask_gemini_cot() CoT 답변을 요청해 vertify_gemini() 답변을 검산해 결과 출력
run_openai_demo() 마찬가지.
print_menu()
메뉴 출력
main()
while True 루프를 쓴다.
print_menu() 메뉴출력, input() 사용자 선택값을 입력받아 if choice가 0 일경우 반복분을 종료, 아닐경우 exc return. 예외시 실행 오류원인을 출력한다.
# -*- coding: utf-8 -*-
"""
main.py
실행 방법:
python src/main.py
프로젝트 목적:
- 제공된 common.py를 공통 모듈로 사용합니다.
- Python + Torch + Gemini API + OpenAI API 실습을 메뉴 방식으로 실행합니다.
"""
# traceback은 예외 발생 시 상세 원인을 확인하고 싶을 때 사용할 수 있습니다.
import traceback
# common.py를 직접 실행하지 않고도 경로와 키 상태를 확인하기 위해 필요한 값을 가져옵니다.
from common import DATA, DOCS, GEMINI_MODEL, ROOT
# llm_clients 모듈에서 Gemini/OpenAI 호출 함수를 가져옵니다.
from llm_clients import (
ask_gemini_cot,
ask_gemini_direct,
ask_openai_cot,
ask_openai_direct,
verify_gemini,
verify_openai,
)
# torch_metrics 모듈에서 API 없는 정답률 계산 데모 함수를 가져옵니다.
from torch_metrics import run_offline_accuracy_demo
# 실습에 공통으로 사용할 예제 문제입니다.
SAMPLE_QUESTION = "이어버드를 79000원에 3개 샀는데 10% 쿠폰을 받았습니다. 총 결제액은?"
def print_project_info() -> None:
"""프로젝트 경로와 데이터 폴더 상태를 출력합니다."""
# 프로젝트 루트 경로를 출력합니다.
print("\n[프로젝트 정보]")
print("-" * 80)
print(f"ROOT : {ROOT}")
print(f"DATA : {DATA} / 존재={DATA.exists()}")
print(f"DOCS : {DOCS} / 존재={DOCS.exists()}")
print(f"GEMINI_MODEL : {GEMINI_MODEL}")
def run_gemini_demo() -> None:
"""Gemini API로 직접 답변, CoT 답변, 자기검증을 실행합니다."""
# 실습 문제를 출력합니다.
print("\n[Gemini API 실습]")
print("-" * 80)
print(f"문제: {SAMPLE_QUESTION}")
# 직접 답변을 요청합니다.
direct = ask_gemini_direct(SAMPLE_QUESTION)
# CoT 답변을 요청합니다.
cot = ask_gemini_cot(SAMPLE_QUESTION)
# CoT 답변을 검산합니다.
verified = verify_gemini(SAMPLE_QUESTION, cot)
# 직접 답변 결과를 출력합니다.
print("\n[직접 답변]")
print(direct)
# CoT 답변 결과를 출력합니다.
print("\n[CoT 답변]")
print(cot)
# 검산 결과를 출력합니다.
print("\n[자기검증 결과]")
print(verified)
def run_openai_demo() -> None:
"""OpenAI API로 직접 답변, CoT 답변, 자기검증을 실행합니다."""
# 실습 문제를 출력합니다.
print("\n[OpenAI API 실습]")
print("-" * 80)
print(f"문제: {SAMPLE_QUESTION}")
# 직접 답변을 요청합니다.
direct = ask_openai_direct(SAMPLE_QUESTION)
# CoT 답변을 요청합니다.
cot = ask_openai_cot(SAMPLE_QUESTION)
# CoT 답변을 검산합니다.
verified = verify_openai(SAMPLE_QUESTION, cot)
# 직접 답변 결과를 출력합니다.
print("\n[직접 답변]")
print(direct)
# CoT 답변 결과를 출력합니다.
print("\n[CoT 답변]")
print(cot)
# 검산 결과를 출력합니다.
print("\n[자기검증 결과]")
print(verified)
def print_menu() -> None:
"""콘솔 메뉴를 출력합니다."""
# 메뉴 제목을 출력합니다.
print("\n" + "=" * 80)
print("CoT Reasoning Prompt 콘솔 실습 앱")
print("=" * 80)
# 메뉴 항목을 출력합니다.
print("1. 프로젝트/공통 모듈 정보 확인")
print("2. Torch 기반 정답률 비교 실행(API 키 불필요)")
print("3. Gemini API 직접 답변 + CoT + 자기검증 실행")
print("4. OpenAI API 직접 답변 + CoT + 자기검증 실행")
print("0. 종료")
def main() -> None:
"""사용자 입력에 따라 각 실습 기능을 실행하는 메인 루프입니다."""
# 프로그램이 종료될 때까지 메뉴를 반복해서 보여 줍니다.
while True:
# 메뉴를 출력합니다.
print_menu()
# 사용자 선택값을 입력받습니다.
choice = input("\n메뉴 번호를 입력하세요: ").strip()
# 사용자가 0을 입력하면 반복문을 종료합니다.
if choice == "0":
print("프로그램을 종료합니다.")
break
# 각 메뉴 실행 중 오류가 나더라도 전체 프로그램이 종료되지 않도록 try로 감쌉니다.
try:
# 1번 메뉴는 프로젝트 정보를 출력합니다.
if choice == "1":
print_project_info()
# 2번 메뉴는 Torch 정확도 데모를 실행합니다.
elif choice == "2":
run_offline_accuracy_demo()
# 3번 메뉴는 Gemini API 실습을 실행합니다.
elif choice == "3":
run_gemini_demo()
# 4번 메뉴는 OpenAI API 실습을 실행합니다.
elif choice == "4":
run_openai_demo()
# 정의되지 않은 번호는 안내합니다.
else:
print("[안내] 메뉴에 있는 번호를 입력하세요.")
# common.py의 require_key는 API 키가 없으면 SystemExit을 발생시키므로 별도로 처리합니다.
except SystemExit as exc:
print(exc)
# 그 밖의 모든 예외를 잡아 오류 원인을 출력합니다.
except Exception as exc:
print(f"[실행 오류] {exc}")
print("상세 오류:")
traceback.print_exc()
# 이 파일을 직접 실행했을 때만 main()을 호출합니다.
if __name__ == "__main__":
main()
또다른 실습프로젝트 function_calling_console_project를 분석해보자.
common.py
import os, pathlib, dotenv
env에서 내용을 읽어 변수설정,
require_key() 환경변수 키가 없으면 종료, 있으면 name을 가져다가 return
get_genai_client() genai 를 불러와 genai.Client() client 만들어 return.
get_chat() provider를 확인해 key가있는지 확인 후에 chatgoogleGenerativeAI, ChatOpenAI객체를 만들어 return.
get_embeddings() langchain_google_genai 에 googlegenerateiveAiEmbeddings 객체를 만들어 return, openai도.
# -*- coding: utf-8 -*-
"""
common.py — 모든 실습 공통 파일
목적:
- .env 의 API 키를 한 곳에서 로드한다.
- Gemini(주력) / OpenAI(보조) 모델 객체를 일관되게 생성한다.
- 실습 데이터(data/) 경로를 쉽게 찾는다.
각 강의 실습 코드 맨 위에서 다음처럼 불러 씁니다.
from common import get_chat, get_genai_client, DATA
"""
import os
import pathlib
from dotenv import load_dotenv
# 프로젝트 루트
ROOT = pathlib.Path(__file__).resolve().parent.parent
DATA = ROOT / "data"
DOCS = DATA / "docs"
# .env 로드 (루트의 .env 를 읽음)
load_dotenv(ROOT / ".env")
GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-flash")
GEMINI_EMBED_MODEL = os.getenv("GEMINI_EMBED_MODEL", "models/gemini-embedding-001")
def require_key(name: str) -> str:
"""환경변수 키가 없으면 종료."""
val = os.getenv(name)
if not val or val.startswith("여기에"):
raise SystemExit(
f"[설정 필요] {name} 가 .env 에 없습니다.\n"
f" 1) cp .env.example .env\n"
f" 2) .env 파일을 열어 {name} 값을 채우세요."
)
return val
# ---------- raw SDK (원리 학습용) ----------
def get_genai_client():
"""google-genai 클라이언트 (from google import genai)."""
from google import genai
return genai.Client(api_key=require_key("GOOGLE_API_KEY"))
# ---------- LangChain Chat 모델 (현업용) ----------
def get_chat(provider: str = "gemini", temperature: float = 0.0):
"""LangChain ChatModel 반환. provider: 'gemini'(기본) | 'openai'."""
if provider == "gemini":
require_key("GOOGLE_API_KEY")
from langchain_google_genai import ChatGoogleGenerativeAI
return ChatGoogleGenerativeAI(model=GEMINI_MODEL, temperature=temperature)
elif provider == "openai":
require_key("OPENAI_API_KEY")
from langchain_openai import ChatOpenAI
return ChatOpenAI(model="gpt-4o-mini", temperature=temperature)
raise ValueError(f"알 수 없는 provider: {provider}")
def get_embeddings(provider: str = "gemini"):
"""LangChain Embeddings 반환."""
if provider == "gemini":
require_key("GOOGLE_API_KEY")
from langchain_google_genai import GoogleGenerativeAIEmbeddings
# gemini-embedding-001 기본 출력은 3072차원. 실습 저장·속도를 위해 768로 고정.
# (768/1536/3072 중 선택 가능)
return GoogleGenerativeAIEmbeddings(
model=GEMINI_EMBED_MODEL, output_dimensionality=768)
elif provider == "openai":
require_key("OPENAI_API_KEY")
from langchain_openai import OpenAIEmbeddings
return OpenAIEmbeddings(model="text-embedding-3-small")
raise ValueError(f"알 수 없는 provider: {provider}")
if __name__ == "__main__":
print("ROOT :", ROOT)
print("DATA :", DATA, "(존재:", DATA.exists(), ")")
print("GEMINI_MODEL :", GEMINI_MODEL)
print("키 로드 상태 — GOOGLE_API_KEY:", bool(os.getenv("GOOGLE_API_KEY")),
"/ OPENAI_API_KEY:", bool(os.getenv("OPENAI_API_KEY")))
tools.py
import pd, torch
common에 data가져오기
load_inventory(), load_exchange_rates csv 읽기,
get_stock() db_down이 true이면 실제 장애처럼 보이도록 예외를 강제로 실행시킨. connectionError,
아닐경우 load_inventory() 이후 row에 product_name에 str.contains() 상품명이 포함된 행을 찾아return. empty시 안내.
get_exchange_rate() load_exchange_rates csv 를 읽어 upper() 대문자로 변환하여 비교한다.
없다면 안내, 있으면 float() 값으로 변환해 return.
torch_inventory_summary()
load_inventory 데이터를 읽어 torch.tensor로 변환해 torch.sum() 총 재고 수량을 계산하고 torch.mean() 평균 재고 수량을 계산하며 torch.sum() stock_tensor가 0인 상품수를 계산해 출력한다.
test_tools_without_llm()
stocj, exchange_rate, get_stoch, get_exchange_rate() 호출 및 print()
# -*- coding: utf-8 -*-
"""Function Calling에 연결할 재고 조회/환율 조회 도구 함수 모듈입니다."""
# pandas는 CSV 데이터를 DataFrame으로 읽고 검색하기 위해 사용합니다.
import pandas as pd
# torch는 재고 수량을 텐서로 변환해 기본 통계 계산을 확인하기 위해 사용합니다.
import torch
# common.py에서 실습 데이터 폴더 경로를 가져옵니다.
from common import DATA
# 재고 DB 다운 상황을 실습하기 위한 전역 스위치입니다.
DB_DOWN = False
def load_inventory() -> pd.DataFrame:
"""data/inventory.csv 파일을 읽어 재고 DataFrame으로 반환합니다."""
# DATA 경로 아래 inventory.csv 파일을 UTF-8 기본 인코딩으로 읽습니다.
return pd.read_csv(DATA / "inventory.csv")
def load_exchange_rates() -> pd.DataFrame:
"""data/exchange_rates.csv 파일을 읽어 환율 DataFrame으로 반환합니다."""
# DATA 경로 아래 exchange_rates.csv 파일을 읽습니다.
return pd.read_csv(DATA / "exchange_rates.csv")
def get_stock(product_name: str) -> str:
"""상품명 일부 또는 전체 이름을 받아 현재 재고 수량과 창고 위치를 반환한다."""
# 함수 내부에서 예외를 처리하여 에이전트 전체가 중단되지 않게 합니다.
try:
# DB_DOWN이 True이면 실제 장애처럼 예외를 강제로 발생시킵니다.
if DB_DOWN:
raise ConnectionError("재고 DB 연결 실패(timeout)")
# 재고 CSV를 DataFrame으로 읽습니다.
inv = load_inventory()
# product_name 컬럼에서 사용자가 입력한 상품명이 포함된 행을 찾습니다.
row = inv[inv["product_name"].str.contains(product_name, na=False)]
# 검색 결과가 없으면 예외 대신 안내 문자열을 반환합니다.
if row.empty:
return f"'{product_name}' 상품을 찾을 수 없습니다. 상품명을 다시 확인해 주세요."
# 첫 번째 검색 결과 행을 선택합니다.
r = row.iloc[0]
# 재고 수량, 창고, 가격을 사람이 읽기 좋은 문자열로 반환합니다.
return f"{r['product_name']} 재고 {int(r['stock'])}개 / 창고: {r['warehouse']} / 가격: {int(r['price_krw']):,}원"
# 모든 예외를 잡아 사용자 안내 메시지로 변환합니다.
except Exception as e:
return f"일시적인 시스템 오류로 재고를 조회하지 못했습니다. 잠시 후 다시 시도해 주세요. 사유: {e}"
def get_exchange_rate(currency: str) -> str:
"""통화 코드(USD, EUR, JPY, CNY)를 받아 1단위당 원화(KRW) 환율을 반환한다."""
# 예외 처리를 통해 잘못된 입력이나 파일 오류에도 프로그램이 멈추지 않게 합니다.
try:
# 환율 CSV를 DataFrame으로 읽습니다.
rates = load_exchange_rates()
# 통화 코드를 대문자로 통일하여 비교합니다.
row = rates[rates["currency"].str.upper() == currency.upper()]
# 없는 통화 코드이면 안내 문자열을 반환합니다.
if row.empty:
return f"{currency} 환율 정보가 없습니다. 지원 통화: USD, EUR, JPY, CNY"
# 환율 값을 float으로 변환합니다.
krw = float(row.iloc[0]["krw_per_unit"])
# 환율 정보를 문자열로 반환합니다.
return f"1 {currency.upper()} = {krw:,.2f} KRW"
# 예외가 발생하면 모델이 처리할 수 있는 문자열로 반환합니다.
except Exception as e:
return f"일시적인 시스템 오류로 환율을 조회하지 못했습니다. 사유: {e}"
def torch_inventory_summary() -> None:
"""PyTorch 텐서로 재고 수량의 합계, 평균, 품절 개수를 계산합니다."""
# 재고 데이터를 읽습니다.
inv = load_inventory()
# stock 컬럼을 float32 텐서로 변환합니다.
stock_tensor = torch.tensor(inv["stock"].tolist(), dtype=torch.float32)
# 총 재고 수량을 계산합니다.
total_stock = torch.sum(stock_tensor).item()
# 평균 재고 수량을 계산합니다.
avg_stock = torch.mean(stock_tensor).item()
# 0개인 상품 수를 계산합니다.
zero_count = torch.sum(stock_tensor == 0).item()
# 결과를 출력합니다.
print("[Torch 재고 통계]")
print(f"재고 텐서: {stock_tensor}")
print(f"총 재고 수량: {int(total_stock)}개")
print(f"평균 재고 수량: {avg_stock:.2f}개")
print(f"품절 상품 수: {int(zero_count)}개")
def test_tools_without_llm() -> None:
"""LLM 연결 전에 도구 함수가 정상 작동하는지 직접 호출합니다."""
# 재고 도구를 직접 호출합니다.
print(get_stock("이어버드"))
# 환율 도구를 직접 호출합니다.
print(get_exchange_rate("USD"))
# 없는 상품 처리를 확인합니다.
print(get_stock("존재하지않는상품XYZ"))
# 없는 통화 처리를 확인합니다.
print(get_exchange_rate("ZZZ"))
openai_app.py
import
tools 모델엇이 직접 실행할 도구 함수와 torch 통계 기능을 제공한다. test_tools_without_llm, torch_inventory_summary
작성한 gemini_app, openai_app 불러오기
openai_tools 리스트 정의하기 . 모델에서 함수이름 설명 입력 파라미터 등 구조를 알려주는 문구.
tools 반환한 함수이름에 실제 파이썬 함수에 연결한다.
openai_manual_tool_calling()
key가 있는지 확인, openai() 클라이언트 생성, question 준비, client.chat.completition.create() 도구목록과 질문을 전달해 first에 할당. 응답메세지 가져오기. tool_calls 도구호출이 없다면 message.content 일반 답변을 출력.
messages에 기존 대확기록을 구성한다.
for문을 돌려 각 도구들을 하나씩 실행한다. 함수이름을 가져와 json문자열 인자를 파이썬 dict로 변환하고 모델 출력, 실제 파이썬 함수 실행. 실행결과출력. message.append() 해 final이라는 변수에 총 내용을 다시 보내 최종 자연어답변을 생성하도록 하자. client.chat.completetion.create() 에 append한 message를 추가해 출력한다.
# -*- coding: utf-8 -*-
"""OpenAI API의 도구 호출(tool calling) 수동 루프 예제 모듈입니다."""
# json은 OpenAI가 반환한 tool_call 인자를 파이썬 dict로 변환하기 위해 사용합니다.
import json
# os는 .env에 설정한 OPENAI_MODEL 값을 읽기 위해 사용합니다.
import os
# OpenAI SDK 클라이언트를 사용합니다.
from openai import OpenAI
# common.py의 require_key로 API 키 존재 여부를 확인합니다.
from common import require_key
# 도구로 사용할 파이썬 함수를 가져옵니다.
from tools import get_exchange_rate, get_stock
# OpenAI 모델명을 .env에서 읽고, 없으면 gpt-4o-mini를 기본값으로 사용합니다.
OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
# OpenAI tool schema는 모델에게 함수 이름, 설명, 입력 파라미터 구조를 알려 줍니다.
OPENAI_TOOLS = [
{
"type": "function",
"function": {
"name": "get_stock",
"description": "상품명 일부 또는 전체 이름을 받아 현재 재고 수량, 창고, 가격을 반환한다.",
"parameters": {
"type": "object",
"properties": {"product_name": {"type": "string", "description": "조회할 상품명 또는 상품명 일부"}},
"required": ["product_name"],
"additionalProperties": False,
},
},
},
{
"type": "function",
"function": {
"name": "get_exchange_rate",
"description": "통화 코드(USD, EUR, JPY, CNY)를 받아 1단위당 원화 환율을 반환한다.",
"parameters": {
"type": "object",
"properties": {"currency": {"type": "string", "description": "USD, EUR, JPY, CNY 중 하나의 통화 코드"}},
"required": ["currency"],
"additionalProperties": False,
},
},
},
]
# OpenAI가 반환한 함수 이름을 실제 파이썬 함수에 연결합니다.
TOOLS = {"get_stock": get_stock, "get_exchange_rate": get_exchange_rate}
def openai_manual_tool_calling() -> None:
"""OpenAI API로 모델의 도구 호출 결정과 실제 함수 실행 루프를 확인합니다."""
# OPENAI_API_KEY가 없으면 명확한 안내와 함께 종료합니다.
require_key("OPENAI_API_KEY")
# OpenAI 클라이언트를 생성합니다.
client = OpenAI()
# 사용자 질문을 준비합니다.
question = "이어버드 재고 있어? 그리고 USD 환율도 알려줘."
# OpenAI Chat Completions API에 도구 목록과 질문을 전달합니다.
first = client.chat.completions.create(
model=OPENAI_MODEL,
messages=[{"role": "user", "content": question}],
tools=OPENAI_TOOLS,
tool_choice="auto",
temperature=0,
)
# 첫 번째 응답 메시지를 가져옵니다.
message = first.choices[0].message
# 도구 호출이 없으면 일반 답변을 출력합니다.
if not message.tool_calls:
print(message.content)
return
# 최종 답변 생성을 위해 기존 대화 기록을 구성합니다.
messages = [{"role": "user", "content": question}, message]
# 모델이 요청한 도구 호출을 하나씩 실행합니다.
for tool_call in message.tool_calls:
# 함수 이름을 가져옵니다.
name = tool_call.function.name
# JSON 문자열 인자를 파이썬 dict로 변환합니다.
args = json.loads(tool_call.function.arguments)
# 모델의 결정을 출력합니다.
print(f"모델의 결정 → 함수: {name} | 인자: {args}")
# 실제 파이썬 함수를 실행합니다.
result = TOOLS[name](**args)
# 실행 결과를 출력합니다.
print(f"우리가 실행한 결과: {result}")
# 도구 결과를 OpenAI 메시지 형식으로 추가합니다.
messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": result})
# 도구 결과까지 포함해 모델에 다시 보내 최종 자연어 답변을 생성합니다.
final = client.chat.completions.create(model=OPENAI_MODEL, messages=messages, temperature=0.3)
# 최종 답변을 출력합니다.
print("\n[OpenAI Tool Calling 최종 답변]")
print(final.choices[0].message.content)
gemini_app.py
common 불러오기
genai에서의 types 불러오기, 작성한 tools 가져오기
마찬가지로 tools정의,
gemini_auto_function_calling() client를 만들어 복합질문을 넘겨 답변을 생성시킨다. 함수도구를 전달하면 SDK가 알아서 함수실행, 결과전달을 자동 처리한다 .
gemini_manual_function_calling() client를 만들어 답변생성. automatic_function_calling.disable=True로 설정, 호출결정만 만들고실행하지않도록 한다.
만일 함수호출을 만들어 내지않았다면 text답변을 출력하고
history를 대화기록에 저장해 함수목록을 순회하며 모델이 고른 함수이름과 인자, 함수이름으로실제 함수를 찾아 실행, 실행결과출력, history에 append, 대화기록에 추가한다. 이 내용을 종합해 client.models.generate_content() 최종답변을 생성한다.
gemini_error_handling_demo() tools 를 import해 client 생성, db_down을 true해 억지로 db를 다운시키는 시뮬레이션을 한다. client을 만들어 돌려본다 .대확기록을 초기화하고 마찬가지로 해당 함수 객체를 찾아 실행해 기록하며 오류상황에 정중히 안내하도록 시스템 지시를 추가한다.
print.
# -*- coding: utf-8 -*-
"""Gemini API의 Function Calling 자동 실행/수동 실행 예제 모듈입니다."""
# common.py에서 Gemini 클라이언트 생성 함수와 모델명을 가져옵니다.
from common import GEMINI_MODEL, get_genai_client
# google.genai.types는 도구 설정, 자동 호출 비활성화, 대화 기록 구성을 위해 사용합니다.
from google.genai import types
# Function Calling에 연결할 도구 함수를 가져옵니다.
from tools import get_exchange_rate, get_stock
# 모델이 반환한 함수 이름을 실제 파이썬 함수 객체로 연결하는 매핑입니다.
TOOLS = {"get_stock": get_stock, "get_exchange_rate": get_exchange_rate}
def gemini_auto_function_calling() -> None:
"""Gemini SDK의 자동 Function Calling을 실행합니다."""
# API 키가 설정되어 있지 않으면 common.py의 require_key에서 SystemExit가 발생합니다.
client = get_genai_client()
# 도구가 필요한 복합 질문을 준비합니다.
question = "이어버드 재고 있어? 그리고 가격을 달러로 환산하려면 환율이 얼마야?"
# Gemini에 함수 도구를 전달하면 SDK가 모델 결정, 함수 실행, 결과 전달을 자동 처리합니다.
resp = client.models.generate_content(
model=GEMINI_MODEL,
contents=question,
config=types.GenerateContentConfig(
tools=[get_stock, get_exchange_rate],
temperature=0,
),
)
# 최종 응답을 출력합니다.
print("[Gemini 자동 Function Calling 결과]")
print(resp.text)
def gemini_manual_function_calling() -> None:
"""Gemini 자동 실행을 끄고 모델의 함수 호출 결정을 직접 관찰합니다."""
# Gemini 클라이언트를 생성합니다.
client = get_genai_client()
# 사용자의 질문을 준비합니다.
question = "이어버드 재고 알려줘."
# automatic_function_calling.disable=True로 설정하면 모델은 호출 결정만 만들고 실행하지 않습니다.
resp = client.models.generate_content(
model=GEMINI_MODEL,
contents=question,
config=types.GenerateContentConfig(
tools=[get_stock, get_exchange_rate],
automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True),
temperature=0,
),
)
# 함수 호출이 없으면 일반 텍스트 답변을 출력합니다.
if not resp.function_calls:
print(resp.text)
return
# 최초 사용자 질문을 대화 기록에 저장합니다.
history = [types.Content(role="user", parts=[types.Part(text=question)])]
# 모델이 결정한 함수 호출 목록을 순회합니다.
for fc in resp.function_calls:
# 모델이 고른 함수 이름과 인자를 출력합니다.
print(f"모델의 결정 → 함수: {fc.name} | 인자: {dict(fc.args)}")
# 함수 이름으로 실제 함수를 찾아 실행합니다.
result = TOOLS[fc.name](**dict(fc.args))
# 실행 결과를 출력합니다.
print(f"우리가 실행한 결과: {result}")
# 모델의 함수 호출 결정을 대화 기록에 추가합니다.
history.append(types.Content(role="model", parts=[types.Part(function_call=fc)]))
# 우리가 실행한 함수 결과를 function_response 형태로 대화 기록에 추가합니다.
history.append(
types.Content(
role="user",
parts=[types.Part.from_function_response(name=fc.name, response={"result": result})],
)
)
# 함수 실행 결과까지 포함한 대화 기록을 다시 모델에 전달해 최종 답변을 생성합니다.
followup = client.models.generate_content(model=GEMINI_MODEL, contents=history)
# 최종 답변을 출력합니다.
print("\n[Gemini 수동 루프 최종 답변]")
print(followup.text)
def gemini_error_handling_demo() -> None:
"""도구 오류 문자열을 모델에 돌려 자연스러운 안내문을 만드는 예제입니다."""
# tools 모듈 자체를 import해야 전역 DB_DOWN 값을 바꿀 수 있습니다.
import tools
# Gemini 클라이언트를 생성합니다.
client = get_genai_client()
# DB 다운 시뮬레이션을 켭니다.
tools.DB_DOWN = True
# 장애 상황에서 재고를 묻는 질문을 준비합니다.
question = "이어버드 재고 확인해 주세요."
# 자동 실행을 끄고 수동 루프로 오류 문자열을 관찰합니다.
resp = client.models.generate_content(
model=GEMINI_MODEL,
contents=question,
config=types.GenerateContentConfig(
tools=[tools.get_stock, tools.get_exchange_rate],
automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True),
temperature=0,
),
)
# 대화 기록을 초기화합니다.
history = [types.Content(role="user", parts=[types.Part(text=question)])]
# 모델이 호출하려는 함수가 있으면 직접 실행합니다.
for fc in resp.function_calls or []:
# 함수 이름에 맞는 실제 함수 객체를 찾습니다.
result = {"get_stock": tools.get_stock, "get_exchange_rate": tools.get_exchange_rate}[fc.name](**dict(fc.args))
# 오류 문자열을 콘솔에 먼저 보여 줍니다.
print(f"도구 실행 결과: {result}")
# 모델의 함수 호출 결정을 기록합니다.
history.append(types.Content(role="model", parts=[types.Part(function_call=fc)]))
# 도구 결과를 function_response로 기록합니다.
history.append(types.Content(role="user", parts=[types.Part.from_function_response(name=fc.name, response={"result": result})]))
# 오류 상황에서도 정중히 안내하도록 시스템 지시를 추가합니다.
followup = client.models.generate_content(
model=GEMINI_MODEL,
contents=history,
config=types.GenerateContentConfig(
system_instruction="너는 승승장구몰 CS 상담원이다. 도구 결과가 오류면 사과하고 다음 행동을 안내하라.",
temperature=0.3,
),
)
# 최종 안내문을 출력합니다.
print("\n[Gemini 오류 처리 최종 답변]")
print(followup.text)
# 다음 실행에 영향을 주지 않도록 DB_DOWN을 다시 끕니다.
tools.DB_DOWN = False
다른 프로젝트를 분석해보자. Tool System 콘솔 실습 프로젝트 tool_system_console_project
common.py
import os, path, dotenv
path 설정, env 정보를 읽어 각 변수설정
require_key 키가 없으면 종료한다. 안내문구 출력
get_genai_client genai를 ipmort해 genai.client 만들기
get_genai_client provider를 인식한뒤에 key 가 require됬는지 확인해 langchain_google_genai혹은 langchain_openai 를 거쳐 chat 객체 만들기
get_embeddings() 마찬가지로 provider를 인식한 뒤에 key를 확인하고 lanchain_google_genai혹은 lanchain_openai를 거쳐 googlegenerativeaiembeddinngs, openaiembeddings 생성
__name~ 직접 실행할경우 print안내문구 출력
# -*- coding: utf-8 -*-
"""
common.py — 모든 실습 공통 파일
목적:
- .env 의 API 키를 한 곳에서 로드한다.
- Gemini(주력) / OpenAI(보조) 모델 객체를 일관되게 생성한다.
- 실습 데이터(data/) 경로를 쉽게 찾는다.
각 강의 실습 코드 맨 위에서 다음처럼 불러 씁니다.
from common import get_chat, get_genai_client, DATA
"""
import os
import pathlib
from dotenv import load_dotenv
# 프로젝트 루트
ROOT = pathlib.Path(__file__).resolve().parent.parent
DATA = ROOT / "data"
DOCS = DATA / "docs"
# .env 로드 (루트의 .env 를 읽음)
load_dotenv(ROOT / ".env")
GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-flash")
GEMINI_EMBED_MODEL = os.getenv("GEMINI_EMBED_MODEL", "models/gemini-embedding-001")
def require_key(name: str) -> str:
"""환경변수 키가 없으면 종료."""
val = os.getenv(name)
if not val or val.startswith("여기에"):
raise SystemExit(
f"[설정 필요] {name} 가 .env 에 없습니다.\n"
f" 1) cp .env.example .env\n"
f" 2) .env 파일을 열어 {name} 값을 채우세요."
)
return val
# ---------- raw SDK (원리 학습용) ----------
def get_genai_client():
"""google-genai 클라이언트 (from google import genai)."""
from google import genai
return genai.Client(api_key=require_key("GOOGLE_API_KEY"))
# ---------- LangChain Chat 모델 (현업용) ----------
def get_chat(provider: str = "gemini", temperature: float = 0.0):
"""LangChain ChatModel 반환. provider: 'gemini'(기본) | 'openai'."""
if provider == "gemini":
require_key("GOOGLE_API_KEY")
from langchain_google_genai import ChatGoogleGenerativeAI
return ChatGoogleGenerativeAI(model=GEMINI_MODEL, temperature=temperature)
elif provider == "openai":
require_key("OPENAI_API_KEY")
from langchain_openai import ChatOpenAI
return ChatOpenAI(model="gpt-4o-mini", temperature=temperature)
raise ValueError(f"알 수 없는 provider: {provider}")
def get_embeddings(provider: str = "gemini"):
"""LangChain Embeddings 반환."""
if provider == "gemini":
require_key("GOOGLE_API_KEY")
from langchain_google_genai import GoogleGenerativeAIEmbeddings
# gemini-embedding-001 기본 출력은 3072차원. 실습 저장·속도를 위해 768로 고정.
# (768/1536/3072 중 선택 가능)
return GoogleGenerativeAIEmbeddings(
model=GEMINI_EMBED_MODEL, output_dimensionality=768)
elif provider == "openai":
require_key("OPENAI_API_KEY")
from langchain_openai import OpenAIEmbeddings
return OpenAIEmbeddings(model="text-embedding-3-small")
raise ValueError(f"알 수 없는 provider: {provider}")
if __name__ == "__main__":
print("ROOT :", ROOT)
print("DATA :", DATA, "(존재:", DATA.exists(), ")")
print("GEMINI_MODEL :", GEMINI_MODEL)
print("키 로드 상태 — GOOGLE_API_KEY:", bool(os.getenv("GOOGLE_API_KEY")),
"/ OPENAI_API_KEY:", bool(os.getenv("OPENAI_API_KEY")))
data_tools.py
import
__future__ import annotations, pandas, common
각 변수에 read.csv()를해 id, 상품명, 카테고리, 가격정보를 담는다.
get_price()
상품명 products에서 문자열이 포함된 행을 찾아 원화문자열을 합해 문자열로 return한다.
get_stock()
재고데이터에서도 마찬가지로. 포함하는 행을 찾아 문자열로 return.
get_order_status 주문번호, search_product 카테고리나 키워드로 상품을 검색해반환 , get_product_info_clear() id가 정확히 일치하는 것을 찾아반환, search_product_clear() 상품명에 키워드가 포함된 상품 후보 찾기
GEMINI_TOOLS SDK 에게 전달할 기본 도구목록 정의,
이름으로 실제파이썬 함수를 찾기 위한 매핑 TOOL_MAP 정의
# -*- coding: utf-8 -*-
"""제6강 Tool System 실습용 도구 함수와 데이터 로드 모듈입니다."""
from __future__ import annotations
# pandas는 CSV 파일을 DataFrame으로 읽고 조건 검색하는 데 사용합니다.
import pandas as pd
# DATA는 common.py에서 제공하는 공통 데이터 폴더 경로입니다.
from common import DATA
# products.csv는 상품 ID, 상품명, 카테고리, 가격 정보를 담습니다.
products = pd.read_csv(DATA / "products.csv", encoding="utf-8")
# inventory.csv는 상품명, 재고 수량, 창고 정보를 담습니다.
inventory = pd.read_csv(DATA / "inventory.csv", encoding="utf-8")
# orders.csv는 주문번호, 상품명, 수량, 배송 상태 정보를 담습니다.
orders = pd.read_csv(DATA / "orders.csv", encoding="utf-8")
def get_price(product_name: str) -> str:
"""상품명(일부만 입력해도 됨)을 받아 판매가(원)를 반환한다. 가격/얼마 질문에 사용."""
# 상품명 컬럼에서 입력 문자열이 포함된 행을 찾습니다.
row = products[products["product_name"].str.contains(product_name, na=False)]
# 검색 결과가 없으면 모델이 읽을 수 있는 안내 문자열을 반환합니다.
if row.empty:
return f"'{product_name}' 가격 정보를 찾지 못했습니다."
# 첫 번째 검색 결과를 선택합니다.
r = row.iloc[0]
# 가격을 쉼표가 포함된 원화 문자열로 반환합니다.
return f"{r['product_name']} 판매가 {int(r['price']):,}원"
def get_stock(product_name: str) -> str:
"""상품명(일부만 입력해도 됨)을 받아 현재 재고 수량과 창고를 반환한다. 재고/품절 질문에 사용."""
# 재고 데이터에서 상품명이 입력 문자열을 포함하는 행을 찾습니다.
row = inventory[inventory["product_name"].str.contains(product_name, na=False)]
# 검색 결과가 없으면 예외 대신 안내 문자열을 반환합니다.
if row.empty:
return f"'{product_name}' 재고 정보를 찾지 못했습니다."
# 첫 번째 검색 결과를 선택합니다.
r = row.iloc[0]
# 상품명, 재고 수량, 창고명을 사람이 읽는 문자열로 반환합니다.
return f"{r['product_name']} 재고 {int(r['stock'])}개 ({r['warehouse']})"
def get_order_status(order_id: str) -> str:
"""주문번호(예: O000106)를 받아 배송 상태를 반환한다. 주문/배송 추적에 사용."""
# 주문번호가 정확히 일치하는 행을 찾습니다.
row = orders[orders["order_id"] == order_id]
# 주문번호가 없으면 안내 문자열을 반환합니다.
if row.empty:
return f"주문번호 {order_id}를 찾지 못했습니다."
# 첫 번째 검색 결과를 선택합니다.
r = row.iloc[0]
# 주문 상태를 사람이 읽는 문자열로 반환합니다.
return f"주문 {order_id}: {r['product_name']} {int(r['quantity'])}개, 상태={r['status']}"
def search_product(keyword: str) -> str:
"""카테고리나 키워드로 상품을 검색해 이름 목록을 반환한다. '어떤 상품 있어?' 류에 사용."""
# 상품명 또는 카테고리에 키워드가 포함된 상품을 찾습니다.
hit = products[
products["product_name"].str.contains(keyword, na=False)
| products["category"].str.contains(keyword, na=False)
]
# 검색 결과가 없으면 안내 문자열을 반환합니다.
if hit.empty:
return f"'{keyword}' 관련 상품이 없습니다."
# 최대 5개 상품명을 목록으로 묶어 반환합니다.
return "검색 결과: " + ", ".join(hit["product_name"].head(5).tolist())
def get_product_info_clear(product_id: str) -> str:
"""[정확검색] 상품ID(예: P0001)로 정확히 1건의 상세정보(이름/카테고리/가격)를 반환한다. 상품ID를 이미 알 때만 사용. 상품명·키워드로는 사용하지 말 것."""
# 상품 ID가 정확히 일치하는 행을 찾습니다.
row = products[products["product_id"] == product_id]
# 상품 ID가 없으면 안내 문자열을 반환합니다.
if row.empty:
return f"상품ID {product_id} 상세정보를 찾지 못했습니다."
# 첫 번째 검색 결과를 선택합니다.
r = row.iloc[0]
# 상세 정보를 문자열로 반환합니다.
return f"{r['product_id']} {r['product_name']} / 카테고리={r['category']} / 가격={int(r['price']):,}원"
def search_product_clear(keyword: str) -> str:
"""[키워드검색] 상품명 일부(예: '이어버드', '청바지')로 여러 후보를 찾아 이름 목록을 반환한다. 상품ID를 모르고 이름·키워드만 알 때 사용."""
# 상품명에 키워드가 포함된 상품 후보를 찾습니다.
hit = products[products["product_name"].str.contains(keyword, na=False)]
# 후보가 없으면 안내 문자열을 반환합니다.
if hit.empty:
return f"'{keyword}' 키워드 상품 후보가 없습니다."
# 후보 목록을 문자열로 반환합니다.
return "상품 후보: " + ", ".join(hit["product_name"].tolist())
# Gemini SDK에 전달할 기본 도구 목록입니다.
GEMINI_TOOLS = [get_price, get_stock, get_order_status, search_product]
# 이름으로 실제 파이썬 함수를 찾기 위한 매핑입니다.
TOOL_MAP = {
"get_price": get_price,
"get_stock": get_stock,
"get_order_status": get_order_status,
"search_product": search_product,
"get_product_info_clear": get_product_info_clear,
"search_product_clear": search_product_clear,
}
torch_demo.py
import __future__ annotation 아직 정식으로 도입되지않았거나 다음버전에서 바뀔 예정인 편리한 타입 힌트 문법을 현재 버전에서 미리쓸수있게한다. 자기자신을 타입으로 지정한다.
타입 힌트 때문에 발생하는 자잘한 순환 참조 에러를 막고, 최신 문법을 안전하게 쓰고 싶을 때 사용함.
정의한 torch, data_tools에 inventory, products. data_tools 불러오기
run_torch_summary()
상품 가격 컬럼을 torch tensor로 변환, stock 재고수량도.
torch.mean() 평균 가격 계산, torch.max() 최고가격 계산, torch.mean() 평균 계산, max를 10으로 잡고 이하인 상품은 부족재고로 판단하도록 하며 이 조건 low_stock_mask에 맞는 list들만 flattern().tolist()해 가져온다 .
for문으로 부족 재고 상품명을 출력한다.
# -*- coding: utf-8 -*-
"""PyTorch로 상품/재고 데이터를 간단히 수치 분석하는 실습 모듈입니다."""
from __future__ import annotations
# torch는 텐서 계산을 위해 사용합니다.
import torch
# data_tools는 CSV를 읽어 둔 DataFrame을 제공합니다.
from data_tools import inventory, products
def run_torch_summary() -> None:
"""상품 가격과 재고를 텐서로 변환하여 평균, 최댓값, 부족 재고를 계산합니다."""
# 상품 가격 컬럼을 float32 텐서로 변환합니다.
price_tensor = torch.tensor(products["price"].tolist(), dtype=torch.float32)
# 재고 수량 컬럼을 float32 텐서로 변환합니다.
stock_tensor = torch.tensor(inventory["stock"].tolist(), dtype=torch.float32)
# 평균 가격을 계산합니다.
avg_price = torch.mean(price_tensor)
# 최고 가격을 계산합니다.
max_price = torch.max(price_tensor)
# 평균 재고를 계산합니다.
avg_stock = torch.mean(stock_tensor)
# 재고가 10개 이하인 상품을 부족 재고로 판단합니다.
low_stock_mask = stock_tensor <= 10
# True인 위치의 인덱스를 가져옵니다.
low_stock_indices = torch.nonzero(low_stock_mask, as_tuple=False).flatten().tolist()
# 계산 결과를 출력합니다.
print("\n[PyTorch 상품/재고 텐서 분석]")
print(f"평균 가격: {avg_price.item():,.0f}원")
print(f"최고 가격: {max_price.item():,.0f}원")
print(f"평균 재고: {avg_stock.item():.1f}개")
print("재고 10개 이하 상품:")
# 부족 재고 상품명을 출력합니다.
for index in low_stock_indices:
product_name = inventory.iloc[index]["product_name"]
stock = int(inventory.iloc[index]["stock"])
print(f"- {product_name}: {stock}개")
gemini_tools.py
__future__ annotations,
common에서 정의한것 불러오기.
google.genai types 불러오기 google.genai.types는 Function callling 걸정 객체를 제공합니다.
gemini는 기본적으로 실시간 날씨나 내 데이터베이스에 있는 재고수량을 모름으로 파이썬으로 get_weather를 가져오게끄음 만들어두고 gemini에게 필요할 경우 함수를 가져오라고 허락해주는 기능 generatiecontenconfig 전체요청 설정 상자로 프롬프트, 함수는 어떤것인지 담아두는것, toolconfig 큰 상자안에 들어가는 작은 상자로 방금 준 함수를 어떻게 다룰지 세부규칙을 정하는것, functioncallingconfig 함수작동 방식 상자 gemini에게 권한을 넘길지 auto, 아니면 any 둘중 하나를 실행시킬지 none 함수를 쓰지않고 대답하게 할지등을 작동 mode를 설정해주는 상자로 구성되어있다.
정의한 data_tools. 불러오기
which_tool()
gemini_tools가 none이면 기존 설정한 tools 4개만 사용하자.
get_genai_client() 클라이언트 생성
client_models_generate)content() 모델을 호출하되 자동함수를 비활성화해 선택 결과만 받는다.
function_calls가 없으면 모델이 도구를 선택하지않았다는 뜻이니 안내. for문으로 선택한 도구 이름과 인자를 출력한다 .
ask_with_gemini() client() 만들어 client.models.generate_content() 자동 function calling으로 도구실행 결과가 반영되도로 ㄱ한다.
코드는 자동이 아닐때 즉 수동일때 코드가 더 길다. 프로그래밍 세계에서는 사람이 편하려면 코드가 짧고 일일이 하려고 하면 코드가 길어진다는 법칙이있다. default가 automatic=true, 앞서 자율모드를 끄기위해 disable=True명시적인 명령을 했고 이 명령서를 옵션상자 types.AutomaticFunctionCallingConfig에 담아 전달했다 .
마찬가지로 안내문구 출력 return.
ask_with_gemini() get_genai_client() 개체를 만들어 client)models.generate_content() 실행. 결과 출력.
run_gemini_tool_choice_demo() 질문목록을 준비. 각 질문에 대해 선택 도구를 출력.
run_similar_tool_demo() tools 상품 도구 별도 목록으로 준비한다.
상품 id질문은 정확검색 도구가 선택되어야하며 키워드 질문은 키워드 검색 도구가 선택되어야한다.
run_gemini_answer_demo() 가격 조회 질문을 실행한다. ask_width_gemini(), 재고조회 질문을 실행한다. ask_width_gemini()
# -*- coding: utf-8 -*-
"""Gemini API로 도구 선택과 자동 Function Calling을 실험하는 모듈입니다."""
from __future__ import annotations
# common.py에서 Gemini 클라이언트 생성 함수와 모델명을 가져옵니다.
from common import GEMINI_MODEL, get_genai_client
# google.genai.types는 Function Calling 설정 객체를 제공합니다.
from google.genai import types
# data_tools는 실습용 도구 함수 목록을 제공합니다.
from data_tools import (
GEMINI_TOOLS,
get_product_info_clear,
get_stock,
search_product_clear,
)
def which_tool(question: str, tools: list | None = None) -> None:
"""자동 실행을 끄고 Gemini가 어떤 도구를 선택하는지 관찰합니다."""
# tools가 None이면 기본 도구 4개를 사용합니다.
selected_tools = GEMINI_TOOLS if tools is None else tools
# Gemini 클라이언트를 생성합니다.
client = get_genai_client()
# 모델을 호출하되 자동 함수 실행을 비활성화하여 선택 결과만 받습니다.
resp = client.models.generate_content(
model=GEMINI_MODEL,
contents=question,
config=types.GenerateContentConfig(
tools=selected_tools,
automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True),
temperature=0,
),
)
# function_calls가 없으면 모델이 도구를 선택하지 않았다는 뜻입니다.
calls = resp.function_calls or []
# 선택된 도구가 없으면 안내를 출력합니다.
if not calls:
print(f"'{question}' → 도구 미선택")
return
# 선택된 도구 이름과 인자를 출력합니다.
for fc in calls:
print(f"'{question}' → {fc.name}({dict(fc.args)})")
def ask_with_gemini(question: str) -> None:
"""Gemini 자동 Function Calling으로 도구 실행 결과가 반영된 최종 답변을 출력합니다."""
# Gemini 클라이언트를 생성합니다.
client = get_genai_client()
# tools에 파이썬 함수를 넘기면 SDK가 자동으로 함수 호출과 결과 전달을 처리합니다.
resp = client.models.generate_content(
model=GEMINI_MODEL,
contents=question,
config=types.GenerateContentConfig(tools=GEMINI_TOOLS, temperature=0),
)
# 최종 답변 텍스트를 출력합니다.
print(resp.text)
def run_gemini_tool_choice_demo() -> None:
"""제6강 도구 4개 선택 관찰 예제를 실행합니다."""
# 실습 질문 목록을 준비합니다.
questions = [
"슬림핏 청바지 얼마야?",
"이어버드 재고 있어?",
"주문 O000106 배송 어디까지 왔어?",
"패션의류 쪽에 어떤 상품들 있어?",
]
# 각 질문에 대해 선택 도구를 출력합니다.
for question in questions:
which_tool(question)
def run_similar_tool_demo() -> None:
"""비슷한 도구를 명확한 독스트링으로 구분하는 예제를 실행합니다."""
# 비슷한 상품 조회 도구 두 개를 별도 목록으로 준비합니다.
tools = [get_product_info_clear, search_product_clear]
# 상품 ID 질문은 정확검색 도구가 선택되어야 합니다.
which_tool("상품ID P0001 상세정보 알려줘.", tools)
# 키워드 질문은 키워드검색 도구가 선택되어야 합니다.
which_tool("이어버드 들어간 상품들 다 찾아줘.", tools)
def run_gemini_answer_demo() -> None:
"""Gemini 자동 도구 실행 결과를 확인합니다."""
# 가격 조회 질문을 실행합니다.
print("\nQ: 슬림핏 청바지 얼마야?")
ask_with_gemini("슬림핏 청바지 얼마야?")
# 재고 조회 질문을 실행합니다.
print("\nQ: 이어버드 재고 있어?")
ask_with_gemini("이어버드 재고 있어?")
openai_tools.py
import __future_ annotations, ljson, openai,
common에서 require_key 함수가져오기
data_tools에서 tool_map가져오기
openai_tools에 도구스키마 목록 정의하기
get_openai_client() requirekey함수를 실행해 발급받은 api_key return.
run_openai_tool_choice_demo() client() 클라이언트를 생성하고 questions를 준비해 fir문으로 질문마다 모델의 tools_calls를 확인한다.
run_openai_tool_execution_demo() client 를 만들고 question세팅, client.chat.completionscreate() 결과 대화기록에 할당, message,
도구호출이 없으면 일반 답변을출려기하고 for문으로 각 도구호출을 실제 파이썬 함수로 실행한다. 함수이름을 가져오고, args json.loads() 로 dict로 변환해 실제파이썬 함수를 찾고 func(**args) 인자를 풀어서 함수를 실행한다.
실행결과를 message.append()
final변수에 client.chat.jcompletions.create() message를 넣어 재 생성 출력.
# -*- coding: utf-8 -*-
"""OpenAI API로 도구 선택과 Tool Calling을 실험하는 모듈입니다."""
from __future__ import annotations
# json은 OpenAI가 반환한 arguments 문자열을 dict로 변환하는 데 사용합니다.
import json
# OpenAI 공식 SDK 클라이언트입니다.
from openai import OpenAI
# common.py의 require_key는 .env에 API 키가 있는지 확인합니다.
from common import require_key
# 실제 실행할 파이썬 도구 함수 매핑을 가져옵니다.
from data_tools import TOOL_MAP
# OpenAI에 전달할 도구 스키마 목록입니다.
OPENAI_TOOLS = [
{
"type": "function",
"function": {
"name": "get_price",
"description": "상품명 일부를 받아 판매가 원화 가격을 반환한다. 가격/얼마 질문에 사용한다.",
"parameters": {
"type": "object",
"properties": {"product_name": {"type": "string", "description": "상품명 또는 상품명 일부"}},
"required": ["product_name"],
},
},
},
{
"type": "function",
"function": {
"name": "get_stock",
"description": "상품명 일부를 받아 현재 재고 수량과 창고를 반환한다. 재고/품절 질문에 사용한다.",
"parameters": {
"type": "object",
"properties": {"product_name": {"type": "string", "description": "상품명 또는 상품명 일부"}},
"required": ["product_name"],
},
},
},
{
"type": "function",
"function": {
"name": "get_order_status",
"description": "주문번호를 받아 주문 상품과 배송 상태를 반환한다. 주문/배송 추적 질문에 사용한다.",
"parameters": {
"type": "object",
"properties": {"order_id": {"type": "string", "description": "주문번호 예: O000106"}},
"required": ["order_id"],
},
},
},
{
"type": "function",
"function": {
"name": "search_product",
"description": "카테고리 또는 키워드로 상품 후보 목록을 반환한다. 어떤 상품이 있는지 묻는 질문에 사용한다.",
"parameters": {
"type": "object",
"properties": {"keyword": {"type": "string", "description": "상품명 일부 또는 카테고리 키워드"}},
"required": ["keyword"],
},
},
},
]
def get_openai_client() -> OpenAI:
"""OPENAI_API_KEY를 확인한 뒤 OpenAI 클라이언트를 생성합니다."""
# require_key는 키가 없으면 사용자가 .env를 설정하도록 안내하고 종료합니다.
api_key = require_key("OPENAI_API_KEY")
# OpenAI 클라이언트를 생성하여 반환합니다.
return OpenAI(api_key=api_key)
def run_openai_tool_choice_demo() -> None:
"""OpenAI 모델이 어떤 도구를 고르는지 확인합니다."""
# OpenAI 클라이언트를 생성합니다.
client = get_openai_client()
# 실습 질문 목록을 준비합니다.
questions = [
"슬림핏 청바지 얼마야?",
"이어버드 재고 있어?",
"주문 O000106 배송 어디까지 왔어?",
"패션의류 쪽에 어떤 상품들 있어?",
]
# 질문마다 모델의 tool_calls를 확인합니다.
for question in questions:
# chat.completions.create는 OpenAI Chat Completions API 호출입니다.
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": question}],
tools=OPENAI_TOOLS,
tool_choice="auto",
temperature=0,
)
# 첫 번째 응답 메시지를 가져옵니다.
message = response.choices[0].message
# tool_calls가 없으면 도구 미선택으로 출력합니다.
if not message.tool_calls:
print(f"'{question}' → 도구 미선택")
continue
# 선택된 도구 이름과 인자를 출력합니다.
for call in message.tool_calls:
print(f"'{question}' → {call.function.name}({call.function.arguments})")
def run_openai_tool_execution_demo() -> None:
"""OpenAI Tool Calling 요청을 실제 파이썬 함수로 실행하고 최종 답변을 생성합니다."""
# OpenAI 클라이언트를 생성합니다.
client = get_openai_client()
# 사용자 질문을 준비합니다.
question = "이어버드 재고와 슬림핏 청바지 가격을 알려줘."
# 첫 번째 호출에서 모델이 필요한 도구를 고르게 합니다.
first = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": question}],
tools=OPENAI_TOOLS,
tool_choice="auto",
temperature=0,
)
# 모델 응답 메시지를 가져옵니다.
assistant_message = first.choices[0].message
# 대화 기록에 사용자 질문과 모델의 도구 호출 결정을 넣습니다.
messages = [
{"role": "user", "content": question},
assistant_message,
]
# 도구 호출이 없으면 일반 답변을 출력합니다.
if not assistant_message.tool_calls:
print(assistant_message.content)
return
# 각 도구 호출을 실제 파이썬 함수로 실행합니다.
for call in assistant_message.tool_calls:
# 함수 이름을 가져옵니다.
name = call.function.name
# arguments JSON 문자열을 dict로 변환합니다.
args = json.loads(call.function.arguments)
# TOOL_MAP에서 실제 파이썬 함수를 찾습니다.
func = TOOL_MAP[name]
# 인자를 풀어서 실제 함수를 실행합니다.
result = func(**args)
# 실행 결과를 OpenAI가 이해하는 tool 메시지로 대화 기록에 추가합니다.
messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
# 도구 실행 결과를 포함한 대화 기록으로 최종 답변을 요청합니다.
final = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=OPENAI_TOOLS,
temperature=0,
)
# 최종 답변을 출력합니다.
print(final.choices[0].message.content)
main.py
import sys
작성한 torch_demo, gemini_tools , openai_tools가져오기
print_menu() 메뉴 출력하기
read_int() 사용자의 입력을 정수로 변환하고 입력이 없고 기본값이 있으면 기본값을 반환한다.
main() while문으로 반복출력. printmenu() choice가 0 일때 종료한다 .
1이면 summary
2 명 tools 성택관찰, 3이면 자동도구실행, 4이면 도구구분 실험, 5번이면 openai 선택, 6번이면 도구실행 후 최종답변
그저 리모콘과 같은 기능의 main()
# -*- coding: utf-8 -*-
"""PyCharm에서 바로 실행하는 제6강 Tool System 콘솔 앱입니다."""
from __future__ import annotations
# sys는 프로그램 종료 처리에 사용합니다.
import sys
# torch_demo는 PyTorch 텐서 분석 예제를 실행합니다.
from torch_demo import run_torch_summary
# gemini_tools는 Gemini API 기반 도구 선택/실행 예제를 제공합니다.
from gemini_tools import (
run_gemini_answer_demo,
run_gemini_tool_choice_demo,
run_similar_tool_demo,
)
# openai_tools는 OpenAI API 기반 도구 선택/실행 예제를 제공합니다.
from openai_tools import run_openai_tool_choice_demo, run_openai_tool_execution_demo
def print_menu() -> None:
"""콘솔 메뉴를 출력합니다."""
# 메뉴 제목을 출력합니다.
print("\n" + "=" * 80)
print("제6강 Tool System 콘솔 실습 앱")
print("=" * 80)
# 각 실행 메뉴를 출력합니다.
print("1. PyTorch 상품/재고 텐서 분석")
print("2. Gemini 도구 선택 관찰")
print("3. Gemini 자동 도구 실행")
print("4. Gemini 비슷한 도구 구분 실험")
print("5. OpenAI 도구 선택 관찰")
print("6. OpenAI 도구 실행 후 최종 답변")
print("0. 종료")
def read_int(prompt: str, default: int | None = None) -> int | None:
"""사용자 입력을 정수로 변환하고 실패하면 기본값을 반환합니다."""
# 사용자에게 입력을 받습니다.
value = input(prompt).strip()
# 아무 입력이 없고 기본값이 있으면 기본값을 반환합니다.
if not value and default is not None:
return default
# 정수 변환을 시도합니다.
try:
return int(value)
# 변환 실패 시 안내 후 기본값을 반환합니다.
except ValueError:
print("숫자를 입력해야 합니다.")
return default
def main() -> None:
"""사용자 메뉴 입력에 따라 실습 기능을 실행합니다."""
# 무한 반복으로 메뉴를 계속 보여 줍니다.
while True:
# 메뉴를 출력합니다.
print_menu()
# 메뉴 번호를 입력받습니다.
choice = input("메뉴 선택: ").strip()
# 0번은 프로그램 종료입니다.
if choice == "0":
print("프로그램을 종료합니다.")
sys.exit(0)
# 1번은 HTML 파일 목록 출력입니다.
if choice == "1":
run_torch_summary()
# 2번은 Gemini 도구 선택 관찰입니다.
elif choice == "2":
run_gemini_tool_choice_demo()
# 3번은 Gemini 자동 도구 실행입니다.
elif choice == "3":
run_gemini_answer_demo()
# 4번은 비슷한 도구 구분 실험입니다.
elif choice == "4":
run_similar_tool_demo()
# 5번은 OpenAI 도구 선택 관찰입니다.
elif choice == "5":
run_openai_tool_choice_demo()
# 6번은 OpenAI 도구 실행 후 최종 답변입니다.
elif choice == "6":
run_openai_tool_execution_demo()
# 정의되지 않은 메뉴는 안내합니다.
else:
print("메뉴 번호를 다시 선택하세요.")
# 이 파일을 직접 실행할 때만 main 함수를 호출합니다.
if __name__ == "__main__":
main()
또다른 프로젝트 react_tools_agent_fastapi_project를 분석해보자.
common.py
import os, path, dotenv
각 변수설정, get_env(), has_key() 환경변수를 읽고 값이 없고 예시 문자열이라면 false를 반홚나다.
require_key() 환경변수값을 읽어 값을 반환한다.
get_openai_client() openai를 가져와 key를 확인한뒤 client 객체를 생성해서 return
get_genai_client() gemai를 가져와 key확인한뒤 client 객체를 생성해서 return.
get_chat() provider를 확인한 다음 key가 있는지 확인하고 langchain의 chatopenai를 import해 객체를 만들어 return.
print_env_status() 각 상태 확인 및 출력
# -*- coding: utf-8 -*-
"""
common.py — 프로젝트 전체 공통 모듈
이 파일은 이전 실습에서 사용한 common.py 구조를 FastAPI 프로젝트용으로 확장한 공통 모듈입니다.
목적:
1) .env 파일의 API 키와 모델명을 한 곳에서 로드합니다.
2) OpenAI, Gemini, LangChain ChatModel 생성 코드를 한 곳에 모읍니다.
3) data/, vector_store/ 같은 공통 경로를 전역 상수로 제공합니다.
"""
# 표준 라이브러리 os는 환경변수를 읽기 위해 사용합니다.
import os
# pathlib.Path는 운영체제별 경로 구분자를 안전하게 처리하기 위해 사용합니다.
from pathlib import Path
# python-dotenv의 load_dotenv는 .env 파일을 읽어 환경변수로 등록합니다.
from dotenv import load_dotenv
# ROOT는 현재 common.py 파일이 있는 프로젝트 루트 경로입니다.
ROOT = Path(__file__).resolve().parent
# DATA는 CSV와 문서 파일을 보관하는 data 폴더 경로입니다.
DATA = ROOT / "data"
# DOCS는 Vector DB에 넣을 텍스트 문서 폴더 경로입니다.
DOCS = DATA / "docs"
# VECTOR_STORE는 로컬 Vector DB 저장 파일을 보관하는 폴더 경로입니다.
VECTOR_STORE = ROOT / "vector_store"
# 루트의 .env 파일을 읽어 OPENAI_API_KEY, GOOGLE_API_KEY 등을 환경변수에 올립니다.
load_dotenv(ROOT / ".env")
# OpenAI 채팅 모델명은 .env에서 읽고, 없으면 gpt-4o-mini를 기본값으로 사용합니다.
OPENAI_CHAT_MODEL = os.getenv("OPENAI_CHAT_MODEL", "gpt-4o-mini")
# OpenAI 임베딩 모델명은 .env에서 읽고, 없으면 text-embedding-3-small을 기본값으로 사용합니다.
OPENAI_EMBEDDING_MODEL = os.getenv("OPENAI_EMBEDDING_MODEL", "text-embedding-3-small")
# Gemini 모델명은 .env에서 읽고, 없으면 gemini-2.5-flash를 기본값으로 사용합니다.
GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-flash")
def get_env(name: str, default: str = "") -> str:
"""환경변수를 안전하게 읽어오는 함수입니다."""
# os.getenv은 환경변수가 없을 때 None을 반환할 수 있으므로 default를 지정합니다.
return os.getenv(name, default)
def has_key(name: str) -> bool:
"""API 키가 .env에 실제로 설정되어 있는지 True/False로 확인합니다."""
# 환경변수 값을 읽습니다.
value = os.getenv(name)
# 값이 없거나 예시 문자열이면 실제 키가 아니므로 False를 반환합니다.
return bool(value and not value.startswith("여기에") and value.strip())
def require_key(name: str) -> str:
"""필수 API 키가 없으면 친절한 오류를 발생시킵니다."""
# 환경변수 값을 읽습니다.
value = os.getenv(name)
# 키가 없거나 예시 문자열이면 설정 방법을 안내하는 오류를 발생시킵니다.
if not value or value.startswith("여기에"):
raise RuntimeError(
f"{name} 값이 .env에 없습니다. .env.example을 .env로 복사한 뒤 {name} 값을 입력하세요."
)
# 정상 키 값을 반환합니다.
return value
def get_openai_client():
"""OpenAI 공식 SDK 클라이언트를 생성합니다."""
# OpenAI 패키지는 API 실행 시점에만 import하여, 키 없이도 로컬 메뉴가 실행되게 합니다.
from openai import OpenAI
# OPENAI_API_KEY가 없으면 명확한 오류를 냅니다.
api_key = require_key("OPENAI_API_KEY")
# OpenAI 클라이언트 객체를 생성해서 반환합니다.
return OpenAI(api_key=api_key)
def get_genai_client():
"""Google Gemini 공식 SDK 클라이언트를 생성합니다."""
# google-genai 패키지는 API 실행 시점에만 import합니다.
from google import genai
# GOOGLE_API_KEY가 없으면 명확한 오류를 냅니다.
api_key = require_key("GOOGLE_API_KEY")
# Gemini 클라이언트 객체를 생성해서 반환합니다.
return genai.Client(api_key=api_key)
def get_chat(provider: str = "openai", temperature: float = 0.0):
"""LangChain ChatModel을 provider 이름에 따라 생성합니다."""
# provider가 openai이면 LangChain OpenAI ChatModel을 생성합니다.
if provider == "openai":
# OPENAI_API_KEY가 있는지 먼저 확인합니다.
require_key("OPENAI_API_KEY")
# LangChain용 OpenAI ChatModel 클래스를 import합니다.
from langchain_openai import ChatOpenAI
# 지정한 모델명과 temperature로 ChatOpenAI 객체를 반환합니다.
return ChatOpenAI(model=OPENAI_CHAT_MODEL, temperature=temperature)
# provider가 gemini이면 LangChain Gemini ChatModel을 생성합니다.
if provider == "gemini":
# GOOGLE_API_KEY가 있는지 먼저 확인합니다.
require_key("GOOGLE_API_KEY")
# LangChain용 Gemini ChatModel 클래스를 import합니다.
from langchain_google_genai import ChatGoogleGenerativeAI
# 지정한 Gemini 모델명과 temperature로 객체를 반환합니다.
return ChatGoogleGenerativeAI(model=GEMINI_MODEL, temperature=temperature)
# 지원하지 않는 provider이면 오류를 발생시킵니다.
raise ValueError(f"지원하지 않는 provider입니다: {provider}")
def print_env_status() -> None:
"""콘솔에서 API 키와 경로 상태를 확인할 때 사용하는 디버깅 함수입니다."""
# 프로젝트 루트 경로를 출력합니다.
print("ROOT:", ROOT)
# data 폴더 경로와 존재 여부를 출력합니다.
print("DATA:", DATA, "존재:", DATA.exists())
# docs 폴더 경로와 존재 여부를 출력합니다.
print("DOCS:", DOCS, "존재:", DOCS.exists())
# OpenAI 키 설정 여부를 출력합니다.
print("OPENAI_API_KEY 설정:", has_key("OPENAI_API_KEY"))
# Gemini 키 설정 여부를 출력합니다.
print("GOOGLE_API_KEY 설정:", has_key("GOOGLE_API_KEY"))
# OpenAI 채팅 모델명을 출력합니다.
print("OPENAI_CHAT_MODEL:", OPENAI_CHAT_MODEL)
# Gemini 모델명을 출력합니다.
print("GEMINI_MODEL:", GEMINI_MODEL)
# 이 파일을 직접 실행하면 환경 설정 상태를 확인합니다.
if __name__ == "__main__":
print_env_status()
config.py
import
pydantic basemodel
작성한 common불러오기
Settings() app_name data_dir 등의 각각 정보를 설정해 객체를 생성한다.
# -*- coding: utf-8 -*-
"""
FastAPI 앱 설정 모듈입니다.
.env 값을 직접 흩뿌리지 않고 Settings 객체 하나로 관리합니다.
"""
# pydantic의 BaseModel은 설정 값을 타입이 있는 객체로 관리하기 위해 사용합니다.
from pydantic import BaseModel
# common 모듈의 공통 경로와 모델명을 가져옵니다.
from common import DATA, DOCS, VECTOR_STORE, OPENAI_CHAT_MODEL, OPENAI_EMBEDDING_MODEL, GEMINI_MODEL, has_key
class Settings(BaseModel):
"""프로젝트 설정 값을 담는 Pydantic 모델입니다."""
# 서비스 이름은 Swagger 문서와 health 응답에 표시됩니다.
app_name: str = "Tools 기반 ReAct Agent FastAPI 프로젝트"
# data 폴더 경로를 문자열로 보관합니다.
data_dir: str = str(DATA)
# docs 폴더 경로를 문자열로 보관합니다.
docs_dir: str = str(DOCS)
# Vector DB 저장 폴더 경로를 문자열로 보관합니다.
vector_store_dir: str = str(VECTOR_STORE)
# OpenAI 채팅 모델명을 보관합니다.
openai_chat_model: str = OPENAI_CHAT_MODEL
# OpenAI 임베딩 모델명을 보관합니다.
openai_embedding_model: str = OPENAI_EMBEDDING_MODEL
# Gemini 모델명을 보관합니다.
gemini_model: str = GEMINI_MODEL
# OpenAI API 키 설정 여부를 보관합니다.
openai_ready: bool = has_key("OPENAI_API_KEY")
# Gemini API 키 설정 여부를 보관합니다.
gemini_ready: bool = has_key("GOOGLE_API_KEY")
# FastAPI 전체에서 공유할 설정 객체를 생성합니다.
settings = Settings()
schemas.py
import typing any, list, optional
파이썬에서 타입을 꼼꼼하게 명시하여 코드의 가독성을 높이고 버그 예방하간다.
any어떤 타입의 데이터가 들어와도 모두 허용, list 특정 타입의 아이템만 모으기, optional 지정한 타입의 값이 들어올수도있지만 값이 없어서 none이 들어올 수도 있다.
from typing import Any
# 어떤 타입의 인자(data)든 받아서 출력만 하는 함수
def log_data(data: Any) -> None:
print(f"로그 기록: {data}")
log_data(10) # 숫자 가능
log_data("Hello") # 문자열 가능
log_data([1, 2, 3]) # 리스트 가능
from typing import List
# 이 함수의 인자는 '정수(int)들로만 구성된 리스트'여야 합니다.
def calculate_total(prices: List[int]) -> int:
return sum(prices)
calculate_total([1000, 2500, 4300]) # 올바른 사용
from typing import Optional
# 이름은 문자열(str)일 수도 있지만, 아직 입력 안 했을 수도 있음(None)
def greet_user(name: Optional[str] = None) -> str:
if name is None:
return "안녕하세요, 게스트님!"
return f"안녕하세요, {name}님!"
greet_user("철수") # "안녕하세요, 철수님!"
greet_user() # "안녕하세요, 게스트님!"
pydantic의 basemodel과 field import
QuestionRequest() question field 생성 사용자가 reAct 에이전트에게 입력하는 질문.
maxsteps 설정
AgentStep()
몇번 반복할것인지, 모델이 어떤 판단을 했는지 요약한 설면 thouth, action 호출한 도구 이름 action_iput 도구에 전달된 인자로 dict[], observation 도구실행 결과 변수 설정
AgentResponse() answer 최종 사용자 답변 정의 , steop list형 단계별 기록, stopped_by final_ansert 종료 이유 정의
VectorSearchRequest()
검색할 자연어 질문과 top k 검색 결과 개수
VectorSearchItem()
doc 문서 고유 id및 source 원본 파일명 출처, score 코사인 유사도, text 검색된 문서 내용 일부 정의
VectorSearchResponse()
items 검색결과목록
StatusResponse()
서버가 정상인지 나타내며 message 상태메세지 deatil은 optional dict 형으로 설정상태나 부가정보를 담도록 한다.
# -*- coding: utf-8 -*-
"""
API 요청/응답 스키마 모듈입니다.
FastAPI는 Pydantic 스키마를 이용해 요청 JSON을 검증하고 Swagger 문서를 자동 생성합니다.
"""
# typing.List는 문자열 리스트 같은 타입 주석에 사용합니다.
from typing import Any, List, Optional
# pydantic의 BaseModel과 Field는 데이터 스키마를 정의하기 위해 사용합니다.
from pydantic import BaseModel, Field
class QuestionRequest(BaseModel):
"""사용자 질문 요청 스키마입니다."""
# question은 사용자가 ReAct 에이전트에게 입력하는 질문입니다.
question: str = Field(..., description="사용자 질문")
# max_steps는 에이전트 루프의 최대 반복 횟수입니다.
max_steps: int = Field(6, description="ReAct 루프 최대 반복 횟수")
class AgentStep(BaseModel):
"""ReAct 루프의 한 단계 실행 기록 스키마입니다."""
# step은 현재 몇 번째 반복인지 나타냅니다.
step: int
# thought는 모델이 어떤 판단을 했는지 요약한 설명입니다.
thought: str
# action은 호출한 도구 이름입니다.
action: str
# action_input은 도구에 전달된 인자입니다.
action_input: dict[str, Any]
# observation은 도구 실행 결과입니다.
observation: str
class AgentResponse(BaseModel):
"""ReAct 에이전트 응답 스키마입니다."""
# answer는 최종 사용자 답변입니다.
answer: str
# steps는 ReAct 루프의 단계별 기록입니다.
steps: List[AgentStep] = []
# stopped_by는 종료 이유입니다.
stopped_by: str = "final_answer"
class VectorSearchRequest(BaseModel):
"""Vector DB 검색 요청 스키마입니다."""
# query는 검색할 자연어 질문입니다.
query: str = Field(..., description="Vector DB 검색 질의")
# top_k는 검색 결과 개수입니다.
top_k: int = Field(3, description="상위 검색 결과 개수")
class VectorSearchItem(BaseModel):
"""Vector DB 검색 결과 1건 스키마입니다."""
# doc_id는 문서 고유 ID입니다.
doc_id: str
# source는 원본 파일명 또는 출처입니다.
source: str
# score는 코사인 유사도 점수입니다.
score: float
# text는 검색된 문서 내용 일부입니다.
text: str
class VectorSearchResponse(BaseModel):
"""Vector DB 검색 응답 스키마입니다."""
# items는 검색 결과 목록입니다.
items: List[VectorSearchItem]
class StatusResponse(BaseModel):
"""서버 상태 응답 스키마입니다."""
# ok는 서버가 정상인지 나타냅니다.
ok: bool
# message는 상태 메시지입니다.
message: str
# details는 설정 상태나 부가 정보를 담습니다.
details: Optional[dict[str, Any]] = None
vector_db.py
import json, hashlib 문자열을 고정 길이 벡터로 바꾸기 위해 사용한다.
path, typing의 any, torch, torch.nn.functional F코사인 시밀러 함수를 사용하기 위한 import
comon에서 함수 가져오기
vector.json 파일 가져오기
_ensure_dirs() 폴더 생성
_read_documents() for문을 순ㄹ회해 docs폴더의 모든 txt파일을 이 름순으로 반복하되 빈파일은 건너뛴다. docs.append 로 각 문서 id 출처 텍스트를 dict로 저장해 return
_local_hash_embedding() torch.zero() 0으로 채운 벡터텐서를 생성하고, lower().split() 소문자, 공백기준으로 나누어서 간단한 토큰 목록을 만든다. 각 토큰을 for문으로 순회하여 hashlib.md5() 해시값으로 바꾼 뒤에 벡터 위치에 누적한다. 각 해시 앞 8자리를 정수로 바꾼 뒤에 dim범위로 나눈다. 해당 위치값을 1증가시키고 l2정규화로 벡터크기가 커지지않도록 한다.
md5 알고리즘은 "동일한 단어는 언제나 완벽하게 똑같은 해시 문자열"을 뱉기 때문에, 단어의 고유한 주민등록번호를 발급하는 것과 같다.
32자리 해시 문자열이 너무 길기 때문에 앞의 8자리만 뚝 자른 뒤(digest[:8]), 이를 10진수 정수로 변환한다. 나머지를 구하면 결과값인 idx는 무조건 0부터 dim - 1 사이의 숫자가 됩니다. 아무리 다양하고 많은 단어가 들어와도 우리가 정한 벡터의 방 크기(dim)를 벗어나지 않게 가두어야함으로 변환된 큰 정수를 우리가 미리 정해둔 벡터의 크기(dim)로 나눈 나머지 값(% dim)을구한다. 방금 계산한 인덱스(idx) 위치의 벡터 값을 1.0 증가시켜 문장 안에 같은 단어가 여러 번 나오거나, 서로 다른 단어인데 우연히 같은 idx로 배정(이를 해시 충돌이라고 합니다)되면 해당 위치의 숫자가 계속 커지게되고 출석푸 투표와 다름없다. 벡터의 모든 요소를 제곱해서 더한 뒤 루트를 씌운 값(L2 Norm)이 정확히 1이 되도록 전체 값의 비율을 깎아주는 것입니다. 이렇게 하면 문장의 길이에 상관없이 모든 벡터가 동일한 스케일(크기 1)을 갖게 되어 머신러닝 모델이 학습하기 아주 좋은 상태가 된다.
_openai_embedding() get_openai_client() 클라이언트 생성, 임베딩벡터 호출해서 첫번째 임베팅 백터를 반환한다.
embed_text() key가 있으면 ㄱeturn키가 없으면 로컬해시 임베딩을 사용한다.
rebuild_vector_db() 원본 문서를 읽고 for문으로 돌려 text 벡터로 변환하며 문서정보와 벡터를 하나의 레코드로 저장한다. json구조를 만들어 벡터파일을 json.dumps() 저장 및 처리결과리턴.
load_vector_db() 저장된 vector db를 읽고 없으면 자동으로 생성한다.
search_vector_db() 저장된 벡터 레코드를 읽어 검색 질문을 torch.tensor로 변환한뒤for문으로 records를 순회하며 저장된 문서도 torch 텐서로 변환한다. 코사인 유사도를 계산해 결과후보를 저장하고 점수가 높은 순서로 정렬해 top k 개만큼만 반환한다 .
# -*- coding: utf-8 -*-
"""
로컬 Vector DB 서비스입니다.
이 프로젝트는 외부 DB 서버 없이도 PyCharm에서 바로 실행되도록,
문서 임베딩을 JSON 파일에 저장하는 '미니 Vector DB'를 구현합니다.
특징:
1) OPENAI_API_KEY가 있으면 OpenAI 임베딩을 사용합니다.
2) 키가 없으면 Torch 기반 해시 임베딩을 사용하여 로컬 실습이 가능합니다.
3) 검색은 torch.nn.functional.cosine_similarity로 수행합니다.
"""
# json은 벡터 저장 파일을 읽고 쓰기 위해 사용합니다.
import json
# hashlib는 API 키가 없을 때 문자열을 고정 길이 벡터로 바꾸기 위해 사용합니다.
import hashlib
# pathlib.Path는 문서 경로와 저장 경로를 안전하게 다루기 위해 사용합니다.
from pathlib import Path
# typing.Any는 검색 결과 dict 타입을 표현하기 위해 사용합니다.
from typing import Any
# torch는 벡터 계산과 코사인 유사도 계산에 사용합니다.
import torch
# torch.nn.functional은 cosine_similarity 함수를 사용하기 위해 import합니다.
import torch.nn.functional as F
# common 모듈에서 공통 경로와 API 설정을 가져옵니다.
from common import DOCS, VECTOR_STORE, OPENAI_EMBEDDING_MODEL, get_openai_client, has_key
# VECTOR_FILE은 Vector DB가 저장될 JSON 파일 경로입니다.
VECTOR_FILE = VECTOR_STORE / "vectors.json"
# LOCAL_DIM은 API 키가 없을 때 사용할 로컬 해시 임베딩 차원입니다.
LOCAL_DIM = 128
def _ensure_dirs() -> None:
"""Vector DB 저장 폴더와 문서 폴더가 없으면 생성합니다."""
# vector_store 폴더를 생성합니다.
VECTOR_STORE.mkdir(parents=True, exist_ok=True)
# docs 폴더를 생성합니다.
DOCS.mkdir(parents=True, exist_ok=True)
def _read_documents() -> list[dict[str, str]]:
"""data/docs 폴더의 txt 문서를 읽어 Vector DB 입력 문서 목록으로 변환합니다."""
# 폴더가 반드시 존재하도록 보장합니다.
_ensure_dirs()
# 문서 목록을 담을 리스트를 생성합니다.
docs: list[dict[str, str]] = []
# docs 폴더의 모든 txt 파일을 이름순으로 반복합니다.
for path in sorted(DOCS.glob("*.txt")):
# 각 파일의 텍스트 내용을 UTF-8로 읽습니다.
text = path.read_text(encoding="utf-8").strip()
# 빈 파일은 검색 품질을 떨어뜨리므로 건너뜁니다.
if not text:
continue
# 문서 ID, 출처, 텍스트를 dict로 저장합니다.
docs.append({"doc_id": path.stem, "source": path.name, "text": text})
# 읽어온 문서 목록을 반환합니다.
return docs
def _local_hash_embedding(text: str, dim: int = LOCAL_DIM) -> list[float]:
"""OpenAI 키가 없을 때 사용하는 Torch 실습용 해시 임베딩입니다."""
# 0으로 채운 벡터 텐서를 생성합니다.
vec = torch.zeros(dim, dtype=torch.float32)
# 텍스트를 공백 기준으로 나누어 간단한 토큰 목록을 만듭니다.
tokens = text.lower().split()
# 토큰이 하나도 없으면 0벡터를 그대로 반환합니다.
if not tokens:
return vec.tolist()
# 각 토큰을 해시값으로 바꾼 뒤 벡터 위치에 누적합니다.
for token in tokens:
# md5는 동일한 문자열에 항상 같은 해시를 만듭니다.
digest = hashlib.md5(token.encode("utf-8")).hexdigest()
# 해시 앞 8자리를 정수로 바꾼 뒤 dim 범위로 나눕니다.
idx = int(digest[:8], 16) % dim
# 해당 위치 값을 1 증가시킵니다.
vec[idx] += 1.0
# 벡터 크기가 커지는 것을 막기 위해 L2 정규화합니다.
vec = F.normalize(vec, dim=0)
# JSON 저장이 가능하도록 리스트로 변환합니다.
return vec.tolist()
def _openai_embedding(text: str) -> list[float]:
"""OpenAI Embedding API로 텍스트 임베딩을 생성합니다."""
# OpenAI 클라이언트를 생성합니다.
client = get_openai_client()
# 임베딩 API를 호출합니다.
response = client.embeddings.create(model=OPENAI_EMBEDDING_MODEL, input=text)
# 첫 번째 임베딩 벡터를 반환합니다.
return response.data[0].embedding
def embed_text(text: str) -> list[float]:
"""환경에 따라 OpenAI 또는 로컬 해시 임베딩을 선택합니다."""
# OpenAI 키가 있으면 실제 OpenAI 임베딩을 사용합니다.
if has_key("OPENAI_API_KEY"):
return _openai_embedding(text)
# 키가 없으면 로컬 해시 임베딩을 사용합니다.
return _local_hash_embedding(text)
def rebuild_vector_db() -> dict[str, Any]:
"""문서 폴더를 다시 읽어 Vector DB JSON 파일을 재생성합니다."""
# 저장 폴더와 문서 폴더를 보장합니다.
_ensure_dirs()
# 원본 문서를 읽습니다.
docs = _read_documents()
# 저장할 레코드 리스트를 생성합니다.
records: list[dict[str, Any]] = []
# 각 문서를 하나씩 벡터화합니다.
for doc in docs:
# 문서 텍스트를 임베딩 벡터로 변환합니다.
vector = embed_text(doc["text"])
# 문서 정보와 벡터를 하나의 레코드로 저장합니다.
records.append({**doc, "vector": vector})
# JSON 파일에 저장할 전체 구조를 만듭니다.
payload = {"count": len(records), "records": records}
# 벡터 파일을 UTF-8 JSON으로 저장합니다.
VECTOR_FILE.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
# 처리 결과를 반환합니다.
return {"saved": str(VECTOR_FILE), "count": len(records), "using_openai_embedding": has_key("OPENAI_API_KEY")}
def load_vector_db() -> list[dict[str, Any]]:
"""저장된 Vector DB를 읽고, 없으면 자동으로 생성합니다."""
# 벡터 파일이 없으면 rebuild_vector_db를 실행합니다.
if not VECTOR_FILE.exists():
rebuild_vector_db()
# JSON 파일 내용을 읽습니다.
payload = json.loads(VECTOR_FILE.read_text(encoding="utf-8"))
# records 목록을 반환합니다.
return payload.get("records", [])
def search_vector_db(query: str, top_k: int = 3) -> list[dict[str, Any]]:
"""질문과 가장 유사한 문서를 Vector DB에서 검색합니다."""
# 저장된 벡터 레코드를 읽습니다.
records = load_vector_db()
# 레코드가 없으면 빈 리스트를 반환합니다.
if not records:
return []
# 검색 질문을 임베딩 벡터로 변환합니다.
query_vector = torch.tensor(embed_text(query), dtype=torch.float32)
# 검색 결과 후보를 담을 리스트입니다.
scored: list[dict[str, Any]] = []
# 모든 문서 레코드를 순회합니다.
for record in records:
# 저장된 문서 벡터를 torch 텐서로 변환합니다.
doc_vector = torch.tensor(record["vector"], dtype=torch.float32)
# 두 벡터 차원이 다른 경우는 임베딩 방식이 바뀐 것이므로 DB 재생성이 필요합니다.
if query_vector.numel() != doc_vector.numel():
raise RuntimeError("임베딩 차원이 다릅니다. /api/vector/rebuild 를 먼저 실행하세요.")
# 코사인 유사도를 계산합니다.
score = F.cosine_similarity(query_vector.unsqueeze(0), doc_vector.unsqueeze(0)).item()
# 결과 후보를 저장합니다.
scored.append(
{
"doc_id": record["doc_id"],
"source": record["source"],
"score": round(float(score), 4),
"text": record["text"],
}
)
# 점수가 높은 순서로 정렬한 뒤 top_k개만 반환합니다.
return sorted(scored, key=lambda x: x["score"], reverse=True)[:top_k]
gemini_service.py
common 불러오기 app의 service vector db 불러오기 db 검색함수
import common, app.service의 vector_db 가져오기
ask_gemini_with_context() key확인한뒤 search_vector_db() 질문과 관련된 문서 검색 결과를 가져와 프롬프트에 넣기 좋은 문자열로 변환하고 client를 생성해 질문한다.응답 텍스트 반환.
# -*- coding: utf-8 -*-
"""
Gemini API 기반 보조 응답 서비스입니다.
이 프로젝트의 핵심 ReAct 루프는 OpenAI + LangChain Tools로 구현합니다.
Gemini API는 같은 질문에 대해 도구 결과 요약 또는 개념 설명을 확인하는 보조 메뉴로 제공합니다.
"""
# common 모듈에서 Gemini 클라이언트와 모델명, 키 확인 함수를 가져옵니다.
from common import GEMINI_MODEL, get_genai_client, has_key
# Vector DB 검색 함수를 가져옵니다.
from app.services.vector_db import search_vector_db
def ask_gemini_with_context(question: str) -> str:
"""Vector DB 검색 결과를 함께 넣어 Gemini에게 답변을 요청합니다."""
# Gemini API 키가 없으면 안내 문자열을 반환합니다.
if not has_key("GOOGLE_API_KEY"):
return "GOOGLE_API_KEY가 설정되어 있지 않아 Gemini API를 실행하지 않았습니다."
# 질문과 관련된 문서 검색 결과를 가져옵니다.
docs = search_vector_db(question, top_k=3)
# 검색 결과를 프롬프트에 넣기 좋은 문자열로 변환합니다.
context = "\n\n".join([f"[{d['source']}]\n{d['text']}" for d in docs])
# Gemini 클라이언트를 생성합니다.
client = get_genai_client()
# Gemini에 전달할 프롬프트를 구성합니다.
prompt = (
"아래 참고 문서를 바탕으로 사용자의 질문에 한국어로 답하라.\n\n"
f"[참고 문서]\n{context}\n\n"
f"[질문]\n{question}"
)
# Gemini 모델을 호출합니다.
response = client.models.generate_content(model=GEMINI_MODEL, contents=prompt)
# 응답 텍스트를 반환합니다.
return response.text
tools_service.py
import pd, lanchain_core.tools에서 tool를 가져와 데이터를 분석하는 파이썬 함수를 짜놓고 위에 @tool 마크를 붙여주면, AI가 이 함수의 이름과 설명을 읽고 "아, 내가 데이터 분석이 필요할 때 이 파이썬 함수를 호출(Function Calling)하면 되겠구나!"라고 인지할 수 있게 한다.
common 모듈에서 data경로 가져오기
app.service.vector_db 가져오기
@tool - get_price() 잘 불러다 쓸 수 있도록 """ 설명을 잘 작성했다. 입력 문자열이 포함된 행을 찾아 iloc[0] 첫번째 검색 결과 행을 가져와 return
@tool - get_stock() inventory에 무자열이 포함된 재고 행을 찾아 마찬가지로 첫번째 경과행을 가져와 return.
@tool - get_reorder_level() 마찬가지, 재주문 기준 수량을 찾아 문자열로 반환
@tool - get_order_status() 주문번호가 일치하는 행을 찾아 return
@tool - search_knowledge_base() db에서 query와 유사한 문서 top k 3개를 검색해 읽기 쉬운 여러줄 문자열로 변환해 return
local_stock_summary() 주요컬럼을 dict 리스트로 변환해 return.
langchain 에 등록할 도구 목록 저으이 ,
도구 tools를 저장한 딕셔너리 정의.
# -*- coding: utf-8 -*-
"""
ReAct 에이전트가 사용할 도구 함수 모듈입니다.
도구 설계 원칙:
1) 한 도구는 한 가지 일만 수행합니다.
2) 함수 이름은 기능을 바로 알 수 있게 작성합니다.
3) 인자 이름은 모델이 채우기 쉽게 명확하게 작성합니다.
4) 독스트링에는 '무엇을 하는가'와 '언제 쓰는가'를 적습니다.
5) 반환값은 항상 문자열로 통일합니다.
"""
# pandas는 CSV 데이터를 읽고 필터링하기 위해 사용합니다.
import pandas as pd
# langchain_core.tools의 tool 데코레이터는 함수를 LangChain 도구로 등록합니다.
from langchain_core.tools import tool
# common 모듈에서 data 폴더 경로를 가져옵니다.
from common import DATA
# Vector DB 검색 함수를 가져옵니다.
from app.services.vector_db import search_vector_db
# products.csv 파일을 DataFrame으로 읽습니다.
products = pd.read_csv(DATA / "products.csv")
# inventory.csv 파일을 DataFrame으로 읽습니다.
inventory = pd.read_csv(DATA / "inventory.csv")
# orders.csv 파일을 DataFrame으로 읽습니다.
orders = pd.read_csv(DATA / "orders.csv")
@tool
def get_price(product_name: str) -> str:
"""상품명 일부를 받아 판매가(원)를 반환한다. 가격/얼마/판매가 질문에 사용한다."""
# 상품명에 입력 문자열이 포함된 행을 찾습니다.
row = products[products["product_name"].str.contains(product_name, na=False)]
# 검색 결과가 없으면 안내 문자열을 반환합니다.
if row.empty:
return f"'{product_name}' 가격 정보를 찾지 못했습니다."
# 첫 번째 검색 결과 행을 가져옵니다.
r = row.iloc[0]
# 상품명과 가격을 사람이 읽기 쉬운 문자열로 반환합니다.
return f"{r['product_name']} 판매가는 {int(r['price']):,}원입니다."
@tool
def get_stock(product_name: str) -> str:
"""상품명 일부를 받아 현재 재고 수량과 창고를 반환한다. 재고/품절/남은 수량 질문에 사용한다."""
# 상품명에 입력 문자열이 포함된 재고 행을 찾습니다.
row = inventory[inventory["product_name"].str.contains(product_name, na=False)]
# 검색 결과가 없으면 안내 문자열을 반환합니다.
if row.empty:
return f"'{product_name}' 재고 정보를 찾지 못했습니다."
# 첫 번째 검색 결과 행을 가져옵니다.
r = row.iloc[0]
# 재고 수량과 창고 정보를 문자열로 반환합니다.
return f"{r['product_name']} 현재 재고는 {int(r['stock'])}개이며, 보관 창고는 {r['warehouse']}입니다."
@tool
def get_reorder_level(product_name: str) -> str:
"""상품명 일부를 받아 재주문 기준 수량을 반환한다. 재주문 필요 여부 판단에 사용한다."""
# 상품명에 입력 문자열이 포함된 행을 찾습니다.
row = inventory[inventory["product_name"].str.contains(product_name, na=False)]
# 검색 결과가 없으면 안내 문자열을 반환합니다.
if row.empty:
return f"'{product_name}' 재주문 기준 정보를 찾지 못했습니다."
# 첫 번째 검색 결과 행을 가져옵니다.
r = row.iloc[0]
# 재주문 기준 수량을 문자열로 반환합니다.
return f"{r['product_name']} 재주문 기준은 {int(r['reorder_level'])}개입니다."
@tool
def get_order_status(order_id: str) -> str:
"""주문번호(예: O000106)를 받아 주문 상품, 수량, 배송 상태를 반환한다. 주문/배송 추적 질문에 사용한다."""
# 주문번호가 정확히 일치하는 행을 찾습니다.
row = orders[orders["order_id"] == order_id]
# 주문번호가 없으면 안내 문자열을 반환합니다.
if row.empty:
return f"주문번호 {order_id}를 찾지 못했습니다."
# 첫 번째 주문 행을 가져옵니다.
r = row.iloc[0]
# 주문 상태를 문자열로 반환합니다.
return f"주문 {order_id}: {r['product_name']} {int(r['quantity'])}개, 현재 상태는 {r['status']}입니다."
@tool
def search_knowledge_base(query: str) -> str:
"""ReAct, 도구 설계, Agent Loop, 무한루프 방지 같은 강의 지식을 Vector DB에서 검색한다."""
# Vector DB에서 query와 유사한 문서 3개를 검색합니다.
results = search_vector_db(query=query, top_k=3)
# 검색 결과가 없으면 안내 문자열을 반환합니다.
if not results:
return "관련 지식 문서를 찾지 못했습니다."
# 결과를 사람이 읽기 쉬운 여러 줄 문자열로 변환합니다.
lines = [f"[{item['source']} | score={item['score']}] {item['text'][:300]}" for item in results]
# 줄바꿈으로 결합하여 반환합니다.
return "\n\n".join(lines)
def local_stock_summary() -> dict:
"""Torch 실습용 재고 통계를 계산하기 위해 재고 DataFrame을 dict 형태로 반환합니다."""
# DataFrame의 주요 컬럼을 dict 리스트로 변환합니다.
return inventory[["product_name", "stock", "reorder_level", "warehouse"]].to_dict(orient="records")
# LangChain에 등록할 도구 목록입니다.
TOOLS = [get_price, get_stock, get_reorder_level, get_order_status, search_knowledge_base]
# 이름으로 도구 객체를 찾기 위한 딕셔너리입니다.
TOOL_MAP = {tool.name: tool for tool in TOOLS}
react_agent.py
import json,
typing의 any, langchain_core.message 에서의 HumanMessage, systemMessage, toolmessage
ANy는 어떤 타입이든 무조건 패스라는 타입힌트, 랭체인 메세지 삼총사로 humanmessage, systemmessage, toolmessage는 각각 역할이 지정된 메세지 객체로 움직이는데 시스템메세지는 ai에게너는 어떤 존재이고 어떤 규칙을 지켜야한다고 정체성을 부여하는 메세지이고 humanmessage는사용자의 말, 툴메세지는 어떤함수가 일한 결과인지 알려주는 세트등을 보내준다.
common에서 불러오기,
app.schemas에서 불러오기
app.services.tools_service 도구 목록과 이름 매핑을 가져오기
_tool_signature() json.dumps 도구명과 인자를 하나 문자열 서명으로 만들어 return한다.
_trim_messages() len 보존 기준이하면 그대로 반환하고 첫번째 시스템메세지는 규칙이니까 보존하되 최근 메세지도 tail이라는 변수에 할당해 head와 tail만 합쳐 return 한다.
run_openai_react_agent() key가 있으면 Agentresponse() 해 응답을 반환한다.
get_chat() 랭케인 chat모델을 생성해 도구몰곡을 바인딩하고 system메세지를 정의한다.
사용자 질문을 humanmessage로 생성한다.
이미 실행한 도구호출을 중복없이 저장하기 위해 set으로 저장한다.
max_steps횟수만큼 ReAct 루프를 반복하되 메세지가 길어지면 토큰 관리를 위해 _trim_message() 한다.
model_width_tools.invoke() 현재 대화목록을 모델에 전달하며 다음행동을 결정하게 하며
message.append() 모델 응답을 history에 추가한다.
getattr했을때 tool_calls가 없으면 agentresponse를 바로 return하고 모델이 요청한 도구 호출을 순서대로 실행해 같은 도구와같은 인자를 만난다면 루프 중단, 실행목록에 저장해 응답용 리스트에 추가해, message에도 도구실행 결과를 만들어 append한다 .
max_steps를 모두 사용하면 안정장치에 의해 종료한다. agentresponse() return.
# -*- coding: utf-8 -*-
"""
OpenAI + LangChain Tools 기반 ReAct 에이전트 서비스입니다.
이 모듈은 LangChain의 tool 객체와 OpenAI ChatModel의 tool calling 기능을 이용해
다음 루프를 직접 구현합니다.
1) 사용자 질문을 history에 넣습니다.
2) LLM이 사용할 도구를 결정합니다.
3) 파이썬 코드가 도구를 실제 실행합니다.
4) 실행 결과를 ToolMessage로 다시 history에 넣습니다.
5) 모델이 최종 답을 낼 때까지 반복합니다.
안전장치:
- max_steps로 최대 반복 횟수를 제한합니다.
- 같은 도구와 같은 인자의 중복 호출을 차단합니다.
"""
# json은 tool args를 안정적인 문자열로 직렬화하기 위해 사용합니다.
import json
# typing.Any는 도구 인자 dict 타입을 표현하기 위해 사용합니다.
from typing import Any
# LangChain 메시지 객체들을 가져옵니다.
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage
# 공통 설정과 키 확인 함수를 가져옵니다.
from common import get_chat, has_key
# API 응답 스키마를 가져옵니다.
from app.schemas import AgentResponse, AgentStep
# 도구 목록과 이름 매핑을 가져옵니다.
from app.services.tools_service import TOOLS, TOOL_MAP
def _tool_signature(name: str, args: dict[str, Any]) -> str:
"""중복 호출 방지를 위해 도구명과 인자를 하나의 문자열 서명으로 만듭니다."""
# dict의 키 순서가 달라도 같은 인자는 같은 문자열이 되도록 sort_keys=True를 사용합니다.
return f"{name}:{json.dumps(args, ensure_ascii=False, sort_keys=True)}"
def _trim_messages(messages: list, keep: int = 8) -> list:
"""대화 기록이 너무 길어지는 것을 막기 위해 최근 메시지만 남깁니다."""
# 메시지가 보존 기준 이하이면 그대로 반환합니다.
if len(messages) <= keep + 2:
return messages
# 첫 번째 SystemMessage는 에이전트 규칙이므로 보존합니다.
head = messages[:2]
# 최근 keep개 메시지만 꼬리로 남깁니다.
tail = messages[-keep:]
# 앞부분과 최근 기록을 합쳐 반환합니다.
return head + tail
def run_openai_react_agent(question: str, max_steps: int = 6) -> AgentResponse:
"""OpenAI 모델과 LangChain Tools로 ReAct 에이전트를 실행합니다."""
# OpenAI API 키가 없으면 로컬 실행 안내 응답을 반환합니다.
if not has_key("OPENAI_API_KEY"):
return AgentResponse(
answer=(
"OPENAI_API_KEY가 설정되어 있지 않아 OpenAI ReAct 루프를 실행하지 않았습니다. "
".env 파일에 OPENAI_API_KEY를 설정한 뒤 다시 실행하세요. "
"키가 없어도 /api/torch/stock-summary 와 /api/vector/search 는 로컬로 확인할 수 있습니다."
),
steps=[],
stopped_by="missing_openai_key",
)
# LangChain OpenAI ChatModel을 생성합니다.
llm = get_chat(provider="openai", temperature=0)
# 모델에 도구 목록을 바인딩합니다.
model_with_tools = llm.bind_tools(TOOLS)
# 시스템 메시지는 ReAct 에이전트의 역할과 규칙을 정의합니다.
system_message = SystemMessage(
content=(
"너는 승승장구몰의 ReAct 에이전트다. "
"필요하면 도구를 호출해 가격, 재고, 주문상태, 강의 지식을 확인한다. "
"도구 결과를 관찰한 뒤 다음 행동을 결정한다. "
"재고가 재주문 기준 이하이면 재주문 필요라고 판단한다. "
"최종 답변은 한국어로 간결하게 작성한다."
)
)
# 사용자 질문을 HumanMessage로 생성합니다.
user_message = HumanMessage(content=question)
# 대화 기록은 시스템 메시지와 사용자 질문으로 시작합니다.
messages = [system_message, user_message]
# 실행 단계 기록을 저장할 리스트입니다.
steps: list[AgentStep] = []
# 이미 실행한 도구 호출을 저장하는 set입니다.
seen_calls: set[str] = set()
# max_steps 횟수만큼 ReAct 루프를 반복합니다.
for step_no in range(1, max_steps + 1):
# 메시지가 길어지면 토큰 관리를 위해 trimming합니다.
messages = _trim_messages(messages)
# 현재 대화 기록을 모델에 전달하여 다음 행동을 결정하게 합니다.
ai_message = model_with_tools.invoke(messages)
# 모델 응답을 history에 추가합니다.
messages.append(ai_message)
# tool_calls가 없으면 모델이 최종 답을 낸 것으로 판단합니다.
if not getattr(ai_message, "tool_calls", None):
return AgentResponse(answer=ai_message.content, steps=steps, stopped_by="final_answer")
# 모델이 요청한 도구 호출들을 순서대로 실행합니다.
for tool_call in ai_message.tool_calls:
# 도구 이름을 읽습니다.
tool_name = tool_call["name"]
# 도구 인자를 읽습니다.
tool_args = tool_call.get("args", {})
# 중복 호출 감지를 위한 서명을 만듭니다.
signature = _tool_signature(tool_name, tool_args)
# 같은 도구와 같은 인자를 이미 실행했다면 루프를 중단합니다.
if signature in seen_calls:
return AgentResponse(
answer="동일한 도구와 인자가 반복 호출되어 무한루프 방지 장치가 작동했습니다.",
steps=steps,
stopped_by="duplicate_tool_call",
)
# 처음 보는 호출이면 실행 기록에 저장합니다.
seen_calls.add(signature)
# 도구 이름이 등록되어 있지 않으면 오류 관찰을 만듭니다.
if tool_name not in TOOL_MAP:
observation = f"등록되지 않은 도구입니다: {tool_name}"
else:
# LangChain Tool 객체를 실제 실행합니다.
observation = TOOL_MAP[tool_name].invoke(tool_args)
# 단계 기록을 응답용 리스트에 추가합니다.
steps.append(
AgentStep(
step=step_no,
thought="모델이 질문과 관찰 결과를 바탕으로 도구 호출을 결정했습니다.",
action=tool_name,
action_input=tool_args,
observation=str(observation),
)
)
# 도구 실행 결과를 ToolMessage로 만들어 모델에게 되돌려줍니다.
messages.append(
ToolMessage(
content=str(observation),
tool_call_id=tool_call["id"],
)
)
# max_steps를 모두 사용하면 안전장치에 의해 종료합니다.
return AgentResponse(
answer="최대 반복 횟수에 도달하여 ReAct 루프를 중단했습니다.",
steps=steps,
stopped_by="max_steps",
)
torch_service.py
import torch, app.service.tools_service 가져오기
analyze_inventory_with_torch()
csv에서 읽은 재고 레코드를 가져와 상품명 리스트를 만들고 torch.tensor로 변환한다.
현재 재고가 기준 이하인지 확인하고
total_count 계산하기, 상품수,계산하기.
for문으로 돌려 if flag 재주문이 필요한 상품 목록을 만들기
float.maen().ietm() 평균 재고 수량을 계산한다. 결과를 dict로 반환
# -*- coding: utf-8 -*-
"""
Torch 기반 분석 서비스입니다.
이 프로젝트에서 torch는 두 가지 역할을 합니다.
1) 재고와 재주문 기준을 텐서로 변환해 재주문 필요 상품을 계산합니다.
2) Vector DB 검색에서 코사인 유사도 계산을 수행합니다.
"""
# torch는 텐서 계산을 위해 사용합니다.
import torch
# 도구 서비스에서 재고 데이터를 dict 형태로 가져옵니다.
from app.services.tools_service import local_stock_summary
def analyze_inventory_with_torch() -> dict:
"""재고 데이터에서 재주문 필요 상품을 Torch 텐서 연산으로 계산합니다."""
# CSV에서 읽은 재고 레코드를 가져옵니다.
rows = local_stock_summary()
# 상품명 리스트를 만듭니다.
names = [row["product_name"] for row in rows]
# 현재 재고 수량을 float32 텐서로 변환합니다.
stock_tensor = torch.tensor([row["stock"] for row in rows], dtype=torch.float32)
# 재주문 기준 수량을 float32 텐서로 변환합니다.
reorder_tensor = torch.tensor([row["reorder_level"] for row in rows], dtype=torch.float32)
# 현재 재고가 재주문 기준 이하인지 Boolean 텐서로 계산합니다.
need_reorder_mask = stock_tensor <= reorder_tensor
# 전체 상품 수를 계산합니다.
total_count = len(rows)
# 재주문 필요 상품 수를 계산합니다.
need_count = int(need_reorder_mask.sum().item())
# 재주문 필요 상품 목록을 만듭니다.
need_items = [names[i] for i, flag in enumerate(need_reorder_mask.tolist()) if flag]
# 평균 재고 수량을 계산합니다.
avg_stock = float(stock_tensor.mean().item())
# 분석 결과를 dict로 반환합니다.
return {
"total_products": total_count,
"need_reorder_count": need_count,
"need_reorder_items": need_items,
"average_stock": round(avg_stock, 2),
}
app/api.py
import fastapi에 apirouter 여러 api 경로를 모듈 단위로 묶기 위해 사용한다.
app.core.config의 settings가져오기
app.schemas 가져오기
app.service 가져오기
라우터 객체 생성
@router.get("/health") statusresponse 객체를 만들어 return. 서버 상태와 api 키 준비 여부를 반환한다.
@router.post("/vector/rebuild") rebuild_vector_db() 실행
@router.post("/vector/search") iems 결과를 가져와 vectorearchresponse() 스키마로 반환
@router.get("/torch/stock-summary") 분석결과 반환
@router.post("/react/openai") run_api 반환
@router.post("/gemini/ask") ask_gemini이후 answer 반환
@router.get("/tools/local-demo") 도구들을 실행해 결과반환.
# -*- coding: utf-8 -*-
"""
FastAPI API 라우터입니다.
화면과 Swagger에서 호출할 수 있는 API 엔드포인트를 정의합니다.
"""
# APIRouter는 여러 API 경로를 모듈 단위로 묶기 위해 사용합니다.
from fastapi import APIRouter
# 설정 객체를 가져옵니다.
from app.core.config import settings
# 요청/응답 스키마를 가져옵니다.
from app.schemas import (
AgentResponse,
QuestionRequest,
StatusResponse,
VectorSearchRequest,
VectorSearchResponse,
)
# Torch 분석 서비스를 가져옵니다.
from app.services.torch_service import analyze_inventory_with_torch
# Vector DB 서비스를 가져옵니다.
from app.services.vector_db import rebuild_vector_db, search_vector_db
# ReAct 에이전트 서비스를 가져옵니다.
from app.services.react_agent import run_openai_react_agent
# Gemini 보조 응답 서비스를 가져옵니다.
from app.services.gemini_service import ask_gemini_with_context
# 도구 함수를 가져옵니다.
from app.services.tools_service import get_price, get_stock, get_reorder_level, get_order_status, search_knowledge_base
# API 라우터 객체를 생성합니다.
router = APIRouter(prefix="/api", tags=["api"])
@router.get("/health", response_model=StatusResponse)
def health() -> StatusResponse:
"""서버와 환경 설정 상태를 확인합니다."""
# 서버 상태와 API 키 준비 여부를 반환합니다.
return StatusResponse(
ok=True,
message="FastAPI 서버가 정상 실행 중입니다.",
details=settings.model_dump(),
)
@router.post("/vector/rebuild")
def rebuild_vector_store() -> dict:
"""data/docs 문서를 다시 임베딩하여 Vector DB를 재생성합니다."""
# Vector DB 재생성 결과를 반환합니다.
return rebuild_vector_db()
@router.post("/vector/search", response_model=VectorSearchResponse)
def vector_search(request: VectorSearchRequest) -> VectorSearchResponse:
"""Vector DB에서 질문과 관련된 문서를 검색합니다."""
# 검색 결과를 가져옵니다.
items = search_vector_db(request.query, request.top_k)
# 응답 스키마로 반환합니다.
return VectorSearchResponse(items=items)
@router.get("/torch/stock-summary")
def stock_summary() -> dict:
"""Torch 텐서 연산으로 재고/재주문 요약을 계산합니다."""
# Torch 분석 결과를 반환합니다.
return analyze_inventory_with_torch()
@router.post("/react/openai", response_model=AgentResponse)
def react_openai(request: QuestionRequest) -> AgentResponse:
"""OpenAI + LangChain Tools 기반 ReAct 에이전트를 실행합니다."""
# ReAct 에이전트를 실행하고 결과를 반환합니다.
return run_openai_react_agent(request.question, request.max_steps)
@router.post("/gemini/ask")
def gemini_ask(request: QuestionRequest) -> dict:
"""Vector DB 검색 결과를 참고 문맥으로 넣어 Gemini API에 질문합니다."""
# Gemini 응답을 생성합니다.
answer = ask_gemini_with_context(request.question)
# 응답을 dict로 반환합니다.
return {"answer": answer}
@router.get("/tools/local-demo")
def local_tools_demo() -> dict:
"""API 키 없이도 도구 함수 동작을 확인하는 로컬 데모입니다."""
# 여러 도구를 직접 실행해 결과를 확인합니다.
return {
"price": get_price.invoke({"product_name": "이어버드"}),
"stock": get_stock.invoke({"product_name": "스마트워치"}),
"reorder_level": get_reorder_level.invoke({"product_name": "스마트워치"}),
"order": get_order_status.invoke({"order_id": "O000106"}),
"knowledge": search_knowledge_base.invoke({"query": "ReAct Agent Loop 안전장치"}),
}
main.py
import path, fastapi, htmlresponse, staticfiles,
app.routers,api에서의 router, app.core.config의 settings가져오기
app의 fastapi 설정 앱 객체를 생성한다.
static 파일경로 계산.
app.mount() static 경로로 정적 파일을 제공한다. /api 하위 경로에 api 라우터를 등록한다.
@app.get("/")
index를 표시한다.
__name ~ 실행시 import uvicorn 하여 uvicorn.run() fastapi 웹서버프로그램을 내 컴퓨터에서 실제로 작동시켜 인터넷 브라우저로 접속할 수 있도록 구동하는 코드. uvicorn, 즉 fastapi는 스스로 돌아갈 수 없고 반드시 asgi서버라는 구동엔진이 필요하다. uvicorn은 대중적인 서버 프로세스. app.main파일을 :app main.py에서 fasiapi의 실제객체 변수이름 선언. host에는 나 자신을 가리키는 127.0.0.1 설정, port 설정. reload=true 코드가 바뀌면 서버를 알아서 껐다 키라는 개발자용 옵션.
# -*- coding: utf-8 -*-
"""
FastAPI 애플리케이션 진입점입니다.
실행 명령:
uvicorn app.main:app --reload
PyCharm에서는 app/main.py를 직접 실행하거나,
터미널에서 위 명령을 실행하면 됩니다.
"""
# pathlib.Path는 정적 파일 경로를 안전하게 지정하기 위해 사용합니다.
from pathlib import Path
# FastAPI는 웹 API 서버를 만들기 위한 프레임워크입니다.
from fastapi import FastAPI
# HTMLResponse는 index.html을 직접 반환할 때 사용합니다.
from fastapi.responses import HTMLResponse
# StaticFiles는 CSS/JS 같은 정적 파일을 서빙할 때 사용합니다.
from fastapi.staticfiles import StaticFiles
# API 라우터를 가져옵니다.
from app.routers.api import router as api_router
# 설정 객체를 가져옵니다.
from app.core.config import settings
# FastAPI 앱 객체를 생성합니다.
app = FastAPI(
title=settings.app_name,
description="Tools 기반 ReAct 에이전트를 FastAPI, Torch, OpenAI, LangChain, Vector DB로 구현한 PyCharm 프로젝트입니다.",
version="1.0.0",
)
# 현재 파일 기준 static 폴더 경로를 계산합니다.
STATIC_DIR = Path(__file__).resolve().parent / "static"
# /static 경로로 정적 파일을 제공합니다.
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
# /api 하위 경로에 API 라우터를 등록합니다.
app.include_router(api_router)
@app.get("/", response_class=HTMLResponse)
def index() -> HTMLResponse:
"""브라우저에서 사용할 기본 실습 화면을 반환합니다."""
# index.html 파일 경로를 지정합니다.
html_path = STATIC_DIR / "index.html"
# HTML 파일 내용을 읽어 응답합니다.
return HTMLResponse(html_path.read_text(encoding="utf-8"))
# 이 파일을 직접 실행할 때 uvicorn 서버를 띄웁니다.
if __name__ == "__main__":
# uvicorn은 FastAPI 앱을 실행하는 ASGI 서버입니다.
import uvicorn
# 개발 모드로 서버를 실행합니다.
uvicorn.run("app.main:app", host="127.0.0.1", port=8000, reload=True)
'Personal > SK 네트웍스 AI 캠프' 카테고리의 다른 글
| [SK네트웍스 Family AI 캠프] 32기 11주차 회고: Day40 ~ Day43 (0) | 2026.07.08 |
|---|---|
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day42_Vector DB (0) | 2026.07.08 |
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day41_LLM의 주요 기능과 활용 사례 (0) | 2026.07.08 |
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day40_LLM 텍스트 생성 제어 파라미터 (0) | 2026.07.07 |
| [SK네트웍스 Family AI 캠프] 32기 10주차 회고: Day35 ~ Day39 + 6월 종합 (0) | 2026.07.05 |
