1-1. 환경변수 설정
from dotenv import load_dotenv
from datetime import datetime, timedelta
from openai import OpenAI
import os
load_dotenv(override=True)
apikey = os.getenv("OPENAI_API_KEY")
default_model = os.getenv("OPENAI_DEFAULT_MODEL")
client = OpenAI(api_key=apikey)
1-2. 툴셋 파일
tools = [
{
"type":"function",
"function" :{
"name":"get_weather",
"description":"오늘의 날씨를 알려주는 함수",
"parameters":{
"type":"object",
"properties":{
"location":{
"type":"string",
"description":"날씨를 알고 싶은 지역"
},
},
"required":["location"]
}
}
},
{
"type":"function",
"function" : {
"name" : "get_time",
"description" : "현재 시간을 알려주는 함수",
"parameters" : {
"type" : "object",
"properties" : {},
"required" : []
}
},
}
]
1-3. 함수 선언
def get_weather(location):
'''
날씨를 가져오는 함수
'''
weather_data = {
"서울": "맑음, 기온 25도",
"부산": "맑음, 기온 30도",
"대구": "맑음, 기온 20도",
"인천": "맑음, 기온 21도"
}
return weather_data.get(location, f"{location}지역의 날씨 정보를 찾을 수 없습니다.")
def get_time():
'''
시간을 가져오는 함수
'''
time_data = {
"현재" : datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
return time_data
available_functions = {
"get_weather" : get_weather,
"get_time" : get_time
}
1-4. 실행파일
import json
tool_calls = response.choices[0].message.tool_calls
if tool_calls:
for tool_call in tool_calls: # 모든 tool_call을 반복 처리
func_name = tool_call.function.name
# print(f"처리 중인 함수: {func_name}")
# get_weather 처리
if func_name == "get_weather":
try:
args = json.loads(tool_call.function.arguments)
if func_name in available_functions:
result = available_functions[func_name](**args)
print(f"날씨 정보: {result}")
except json.JSONDecodeError:
print("인자 파싱 오류")
# get_time 처리
elif func_name == "get_time":
if func_name in available_functions:
result = available_functions[func_name]() # 파라미터 없음
print(f"현재 시각: {result['현재']}")
2-1. 그라디오 기본셋팅
import gradio as gr
def greet(name):
return f"안녕하세요. {name}님! 만나서 반갑습니다."
demo = gr.Interface(fn=greet, inputs="text", outputs="text")
demo.launch(share=True)
2-2. 그라디오 슬라이더
import gradio as gr
def greet(name, is_morning, age):
a = "아침" if is_morning else "저녁"
age = age-10
b = f"안녕하세요. {name}님! 좋은 {a} 입니다. {age}처럼 보입니다"
return b, age
demo = gr.Interface(fn=greet, inputs=["text", "checkbox", gr.Slider(0, 100, 50)],
outputs=["text", "number"])
demo.launch(share=True)
2-3. 그라디오 탭 생성
import gradio as gr
tab1 = gr.Interface(lambda x : x[::-1],
inputs="text",
outputs="text",
title = 'reverse')
tab2 = gr.Interface(lambda x : x.upper(),
inputs="text",
outputs="text",
title = 'Uppercase')
demo = gr.TabbedInterface([tab1, tab2], ["뒤집기", "대문자"])
demo.launch(share=True)
2-4. 그라디오 챗봇 화면
import time
import random
def response(msg, history):
result = random.choice(['안녕하세요', '반갑습니다','배가 고파요','집에 가고 싶어요'])
# 초기 응답은 빈 문자열로 시작
result_stream = ""
for char in result:
result_stream += char
time.sleep(0.1)
# 히스토리 마지막 요소를 업데이트하여 yield
yield history + [(msg, result_stream)], ""
with gr.Blocks() as demo:
chatbot = gr.Chatbot()
msg = gr.Textbox(placeholder="메시지를 입력하세요", label="메시지")
btn2 = gr.Button("초기화")
msg.submit(fn = response,
inputs=[msg, chatbot],
outputs=[chatbot, msg],
)
demo.launch(share=True)
2-5. 챗봇 1차 version
from dotenv import load_dotenv
from datetime import datetime
import os
import gradio as gr
from openai import OpenAI
# 환경 변수 로딩
load_dotenv(override=True)
apikey = os.getenv("OPENAI_API_KEY")
default_model = os.getenv("OPENAI_DEFAULT_MODEL")
client3 = OpenAI(api_key=apikey)
SAVE_PATH = "chat_history.json"
# 대화 불러오기 함수
def load_history():
if os.path.exists(SAVE_PATH):
with open(SAVE_PATH, "r", encoding="utf-8") as f:
return json.load(f)
return []
# 대화 저장 함수
def save_history(history):
with open(SAVE_PATH, "w", encoding="utf-8") as f:
json.dump(history, f, ensure_ascii=False, indent=2)
return "✅ 대화가 저장되었습니다!"
# 응답 함수
def response(msg, history):
if history is None:
history = load_history()
messages = [{"role": "system", "content": "너는 지금 금융 전문가 AI야. 보고서 형식으로 깔끔하게 대답해줘."}]
# Gradio가 전달한 history를 직접 사용
messages += history.copy()
messages.append({"role": "user", "content": msg})
result = client3.chat.completions.create(
model=default_model,
messages=messages,
stream=True,
max_tokens=500,
)
reply = ""
for chunk in result:
delta = chunk.choices[0].delta.content
if delta:
reply += delta
time.sleep(0.1)
# 실시간으로 답변 생성하면서 반환
yield history + [{"role": "user", "content": msg},
{"role": "assistant", "content": reply}
], ""
# 마지막에 저장
history.append({"role": "user", "content": msg})
history.append({"role": "assistant", "content": reply})
save_history(history)
with gr.Blocks() as demo:
chatbot = gr.Chatbot(label="AI 챗봇", type="messages") # 반드시 type='messages'
msg = gr.Textbox(placeholder="메시지를 입력하세요.", label="메시지")
btn2 = gr.Button("초기화")
msg.submit(response, inputs=[msg, chatbot], outputs=[chatbot, msg])
save_btn = gr.Button("💾 저장")
load_btn = gr.Button("📂 불러오기")
clear_btn = gr.Button("🧹 초기화")
status = gr.Markdown("") # ⬅️ 알림 메시지 출력용 컴포넌트
save_btn.click(fn=save_history, inputs=[chatbot], outputs=[status])
load_btn.click(fn=load_history, inputs=[], outputs=[chatbot])
clear_btn.click(lambda:[], inputs=None, outputs=chatbot)
demo.launch(share=True)
3-1. Lang chain : 기본 셋팅
from dotenv import load_dotenv
from datetime import datetime
import os
import gradio as gr
from openai import OpenAI
from langchain_openai import ChatOpenAI
from langchain_core.messages import AIMessage, SystemMessage, HumanMessage
# 환경 변수 로딩
load_dotenv(override=True)
apikey = os.getenv("OPENAI_API_KEY")
default_model = os.getenv("OPENAI_DEFAULT_MODEL")
default_model2 = os.getenv("OPENAI_DEFAULT_MODEL_4")
llm = ChatOpenAI(api_key=apikey, model=default_model)
llm
messages1 = [("system", "당신은 서울의 음식과 문화 전문가입니다."),
("human", "서울을 대표하는 음식을 맛볼 수 있는 레스토랑 5개를 추천해주세요")]
messages2 = [[("system", "당신은 서울의 음식 전문가입니다."),
("human", "서울 신도림역쪽 주변의 블루리본 맛집을 5군데 추천해주세요")],
[("system", "당신은 서울의 문화 전문가입니다."),
("human", "서울의 문화 데이트 코스 5군데를 짜주세요.")]]
response = llm.stream(messages1)
# response = llm.batch(messages)
for chunk in response:
print(chunk.content, end='')
chat = ChatOpenAI(api_key=apikey, model=default_model, temperature=0.9)
sys = SystemMessage(content="당신은 음악을 추천해주는 전문 AI입니다.")
user = HumanMessage(content="2000년대 팝송 5곡을 추천해줘.")
msg = chat([sys, user])
print(msg.content)
3-2. lang-chain 병렬처리
import asyncio
import time
async def invoke_async(llm, messages):
response = await llm.ainvoke(messages)
print(response.content)
async def invoke_parallel(llm, messages):
tasks = [invoke_async(llm, messages) for _ in range(10)]
await asyncio.gather(*tasks)
print("Async")
start = time.perf_counter()
await invoke_parallel(llm, messages)
end = time.perf_counter()
print(f"Elapsed time: {end - start:.2f} seconds")
3-3. lang-chain for문으로 여러 메시지 보내기(대답은 이전 질문을 기억해서 답변하도록)
from langchain.schema import AIMessage, SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI
chat = ChatOpenAI(api_key=apikey, model=default_model, temperature=0.9)
messages = [('system', '당신은 여행 전문가로 고객의 여행 일정에 도움을 줍니다.')]
msg = ["부산 여행에서 딱 한 곳만 가봐야 한다면 어떤 곳인지 추천해주세요.",
'부산역에서 그 곳에 가는 교통편을 알려주세요.',
'그 근처에서 먹을만한 음식점을 추천해주세요.']
for i in msg:
messages.append(HumanMessage(content=f"'user', {i}"))
response = llm.invoke(messages)
messages.append(response)
print(response.content)
3-4. lang-chain 에서 history를 기억함. 어떻게 사용할까?
from langchain.memory import chat_message_histories
from langchain.memory.chat_message_histories import ChatMessageHistory
history = ChatMessageHistory()
history.add_message(SystemMessage(content='당신은 여행 전문가로 고객이ㅡ 여행 일정에 도움을 줍니다.'))
msg = ["부산 여행에서 딱 한 곳만 가봐야 한다면 어떤 곳인지 추천해주세요.",
'부산역에서 그 곳에 가는 교통편을 알려주세요.',
'그 근처에서 먹을만한 음식점을 추천해주세요.']
for i in msg:
history.add_user_message(i)
response = llm.invoke(history.messages)
history.add_ai_message(response.content)
3-5. 최근 대화 n개만 저장하도록!
from langchain.memory import ConversationBufferWindowMemory
#같은 메모리인데 최근 대화 k개를 가지고 있는 메모리
memory = ConversationBufferWindowMemory(memory_key="chat_history",
return_messages=True,
k=3) #최근 k개의 대화만 저장
memory.clear()
memory.save_context({'human':'안녕하세요 챗봇님'},
{'bot':'안녕하세요! 반가워요, 호칭을 어떻게 하는게 좋을까요?'}) #context의 시작은 항상 사용자
memory.save_context({'human':'나는 홍길동입니다'},
{'bot':'네, 홍길동님 만나서 반갑습니다'})
memory.save_context({'human':'경복궁을 가고 싶은데 어떻게 가면 되는지 알려주세요'},
{'bot':'물론이죠. 경복궁은 지하철을 타고 가면 편리합니다'})
memory.save_context({'human':'몇 호선을 타고 가야하는지 아니?'},
{'bot':'3호선을 타고 가세요'})
result = memory.load_memory_variables({})
for msg in result['chat_history']:
print(type(msg), msg.content)
3-6. 대화에서 요약된 내용만 출력하세요.
from langchain.memory import ConversationSummaryBufferMemory
memory = ConversationSummaryBufferMemory(llm=llm,
return_messages=True,
max_token_limit=50)
memory.clear()
memory.save_context({'human':'안녕하세요 챗봇님'},
{'bot':'안녕하세요! 반가워요, 호칭을 어떻게 하는게 좋을까요?'}) #context의 시작은 항상 사용자
memory.save_context({'human':'나는 홍길동입니다'},
{'bot':'네, 홍길동님 만나서 반갑습니다'})
memory.save_context({'human':'경복궁을 가고 싶은데 어떻게 가면 되는지 알려주세요'},
{'bot':'물론이죠. 경복궁은 지하철을 타고 가면 편리합니다'})
memory.save_context({'human':'몇 호선을 타고 가야하는지 아니?'},
{'bot':'3호선을 타고 가세요'})
result= memory.load_memory_variables({})
print(result) # 요약된 대화 내용 출력
3-7. 대답을 하는데에 걸리는 시간, 비용, 토큰 수를 출력하고싶어.
from langchain.callbacks import get_openai_callback
with get_openai_callback() as callback:
message = [HumanMessage(content="서울 광장시장에서 가장 맛있는 길거리 음식은 무엇인가요?")]
response = llm.invoke(message)
print(response.content)
print(callback)
print(callback.total_tokens)
'키움 디지털 아카데미' 카테고리의 다른 글
| python _ 키움 아카데미 _ 19회차 _ (AI수업: RAG, VectorDB, pdf학습) (2) | 2025.06.17 |
|---|---|
| python _ 키움 아카데미 _ 18회차 _ (AI수업: Redis 연결) (4) | 2025.06.16 |
| python _ 키움 아카데미 _ 16회차 _ (AI수업: GPT API 연결) (8) | 2025.06.12 |
| python _ 키움 아카데미 _ 15회차 _ (AI수업: FastAPI) (4) | 2025.06.11 |
| python _ 키움 아카데미 _ 14회차 _ (금융데이터 추세분석2) (0) | 2025.05.14 |