Day 1

ADK를 시작해본다

https://aistudio.google.com/u/1/prompts/new_chat?pli=1

#.env
GOOGLE_API_KEY=AIz...
GOOGLE_GENAI_USE_VERTEXAI=FALSE

pip install google-adk==1.9.0

#agent.py
from google.adk.agents import Agent

def greet_user() -> str:
    return "안녕"

root_agent = Agent(
    name='skn_agent',
    model="gemini-2.5-flash",
    description='테스트',
    instruction='사용자에게 친한 친구처럼 인사해',
    tools = [greet_user]
)

(base) C:\skn-25\wk13\adk_test>adk web

• .env에서 일일이 설정 귀찮으면 시스템 환경 변수에서 설정도 가능하다.

C:\skn-25\wk13>echo %GOOGLE_API_KEY%

import os

os.getenv("GOOGLE_API_KEY")

위치 정보

pip install geopy

from geopy.geocoders import Nominatim
geo = Nominatim(user_agent='weather_app')

geo.geocode('분당')

url = "https://api.open-meteo.com/v1/forecast?latitude=37.3947237&longitude=127.1115312&current_weather=true"

import httpx
r = httpx.get(url)
r.json()

날씨 합치기

내 코드

from google.adk.agents import Agent
from geopy.geocoders import Nominatim
import httpx

def greet_user() -> str:
    return "안녕"

def get_weather(city_name : str) -> dict:
    geo = Nominatim(user_agent='weather_app')
    url = f"https://api.open-meteo.com/v1/forecast?latitude={geo.geocode(city_name).latitude}&longitude={geo.geocode(city_name).longitude}&current_weather=true"
    r = httpx.get(url)
    output=r.json()
    return output['current_weather']

root_agent = Agent(
    name='skn_agent',
    model="gemini-2.5-flash",
    description='테스트',
    instruction='사용자에게 친한 친구처럼 인사해',
    tools = [greet_user,get_weather]
)

from google.adk.agents import Agent
from geopy.geocoders import Nominatim
import httpx

def get_weather(city_name : str) -> dict:
    """
    도시 이름을 사용하여 위도, 경도 위치를 알아내고
    그 값을 통해서 현재 날씨 정보를 반환합니다.
    """
    geo = Nominatim(user_agent='weather_app')
    location = geo.geocode(city_name)
    if location:
        lat, lon  = location.latitude, location.longitude
    else:
        raise ValueError(f'{city_name} 위치의 데이터가 없다 ')
   
    url = "https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&current_weather=true"
    r = httpx.get(url)
    r.raise_for_status()
    return r.json()

def greet_user() -> str:
    return "안녕"

root_agent = Agent(
    name='skn_agent',
    model="gemini-2.5-flash",
    description='테스트',
    instruction='사용자에게 반말해줘',
    tools = [greet_user]
)

강사님 코드

from google.adk.agents import Agent
from geopy.geocoders import Nominatim
import httpx

def get_weather(city_name : str) -> dict:
    """
    도시 이름을 사용하여 위도, 경도 위치를 알아내고
    그 값을 통해서 현재 날씨 정보를 반환합니다.
    """
    geo = Nominatim(user_agent='weather_app')
    location = geo.geocode(city_name)
    if location:
        lat, lon  = location.latitude, location.longitude
    else:
        raise ValueError(f'{city_name} 위치의 데이터가 없다 ')
   
    url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&current_weather=true"
    r = httpx.get(url)
    r.raise_for_status()
    return r.json()

def greet_user() -> str:
    return "안녕"

root_agent = Agent(
    name='skn_agent',
    model="gemini-2.5-flash",
    description='테스트',
    instruction='사용자에게 반말해줘',
    tools = [greet_user, get_weather]
)

ADK

에이전트

• 워크플로 에이전트
  • 다른 에이전트의 실행 흐름을 체계적으로 제어
  • SequentialAgent - 에이전트를 순차적으로 실행
  • ParallelAgent - 여러 에이전트를 동시에 실행
  • LoopAgent - 특정 조건이 만족될 때 까지 반복하여 실행
• 커스텀 에이전트 ⇒ BaseAgent를 직접 확장하여 맞춤화하여 제공
• LLM 에이전트 특징
  • 자연어 이해와 추론
  • 동적 의사결정
  • 유연성
  • 언어 중심 작업 ⇒ 문서 생성, 번역 , 요약

에이전트 전용 환경 세팅

pip install uv

폴더 myagent 

uv venv 

uv venv --python 3.12

.venv\Scripts\activate

uv pip list

requirements.txt
--
google-adk==1.9.0
geopy==2.4.1
httpx
fastmcp==3.1.1

uv pip install -r requirements.txt 

where adk

(myagent) C:\skn-25\wk13\myagent>where adk
C:\skn-25\wk13\myagent\.venv\Scripts\adk.exe
C:\Users\Ming9\miniconda3\Scripts\adk.exe

adk web 

기업정보 가져오는 법 탐구

판다스 모듈 추가

uv pip install pandas

회사 정보 추가

from google.adk.agents import Agent
from geopy.geocoders import Nominatim
import httpx
import pandas as pd
from dotenv import load_dotenv
load_dotenv()

def get_code(company_name: str) -> dict:
    """
    사용자가 회사 이름을 전달하면 해당 회사의 종목 코드를 dict형태로 반환한다.
    """
    df = pd.read_csv("C:/skn-25/wk13/myagent/skn_agent/data_2058_20260323.csv", encoding='cp949')
    return df[df['한글 종목명'].apply(lambda x : x.find(company_name) > -1)].to_json()

def get_company_info(company_code:str) -> dict:
    """
    종목 코드를 사용하여 회사 정보를 반환하는 코드
    회사명에서 종목코드 추출은 get_code가 담당
    company_code예시 : 005930

    """
    url = f"https://wts-info-api.tossinvest.com/api/v2/stock-infos/A{company_code}/overview"
    rt = httpx.get(url).json()
    return rt['result']['company']

def get_weather(city_name : str) -> dict:
    """
    도시 이름을 사용하여 위도, 경도 위치를 알아내고
    그 값을 통해서 현재 날씨 정보를 반환합니다.
    """
    geo = Nominatim(user_agent='weather_app')
    location = geo.geocode(city_name)
    if location:
        lat, lon  = location.latitude, location.longitude
    else:
        raise ValueError(f'{city_name} 위치의 데이터가 없다 ')
   
    url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&current_weather=true"
    r = httpx.get(url)

    r.raise_for_status()
    return r.json()

def greet_user() -> str:
    return "안녕"

root_agent = Agent(
    name='skn_agent',
    model="gemini-2.5-flash",
    description='자기소개서 완성하는 에이전트',
    instruction='사용자 요청한 내용을 바탕으로 회사 자기소개서를 완성하는 에이전트임',
    tools = [greet_user,get_weather, get_code, get_company_info]
)

adk web #myagent 폴더에서

Day 2

resume agent 추가

from google.adk.agents import Agent
from geopy.geocoders import Nominatim
import httpx
import pandas as pd
from dotenv import load_dotenv
load_dotenv()

INSTRUCTIONS = """
당신은 채용 전문가 에이전트입니다.
사용자가 입력한 자기소개서를 바탕으로 아래 기준에 따라서 자기소개서를 수정할 것

##기본 평가항목
- 질문 항목에 대한 답변으로 적절한가?
- 지원자의 지원동기 등 입사에 대한 열의가 있는가?
- 핵심가치 및 인재상에 부합하는 인재인가?
- 지원 직무에 대한 역량은 갖추었는가?
- 예비 사회인으로서 다양한 경험을 하였는가?
- 팀워크, 리더십, 갈등해결 등 대인관계능력을 갖추었는가?
- 지원 기업에 대한 이해도를 갖추고자 노력하였는가?
- 다양한 변화에 대응할 인문학적 소양 등을 갖추었는가?

##가점요인
- 역량 중심으로 고민을 담아 성의껏 작성하였는가?
- 경험 위주로 근거를 갖고 구체적으로 기술하고 있는가?
- 표현능력(논리력, 문장력 등)이 우수한가?

##감점요인
- 동일 내용을 반복하고 있는가?
- 제시된 분량에 턱없이 모자라는가? (70% 이하)
- 원론적이고 추상적인 내용으로 기술하고 있는가?
- 질문과 무관한 내용이 담겨 있는가?
- 오타 및 부정확한 표현은 없는가?
- 타사 지원서를 복사하여 지원 회사명이 잘못 기재하였는가?

"""

root_agent = Agent(  
    name='resume_agent',
    model="gemini-2.5-flash",
    description='사용자가 입력한 자기소개서를 주어진 지시에 따라 수정하여 출력하는 에이전트',
    instruction=INSTRUCTIONS
)

news 에이전트 설계

from google.adk.agents import Agent
from google.adk.tools import google_search

from dotenv import load_dotenv
load_dotenv()

root_agent = Agent(
    name='news',
    model="gemini-2.5-flash",
    instruction='제공된 기업의 주요 뉴스를 요약할 것. 앞으로 전망도 이야기할 것',
    tools = [google_search]
)

최종 자기소개서 작성 agent

from google.adk.tools import google_search
from google.adk.agents import Agent, ParallelAgent, SequentialAgent
from geopy.geocoders import Nominatim
import httpx
import pandas as pd

INSTRUCTIONS = """
당신은 채용 전문가 에이전트입니다.
사용자가 입력한 자기소개서를 바탕으로 아래 기준에 따라서 자기소개서를 수정할 것

##기본 평가항목
- 질문 항목에 대한 답변으로 적절한가?
- 지원자의 지원동기 등 입사에 대한 열의가 있는가?
- 핵심가치 및 인재상에 부합하는 인재인가?
- 지원 직무에 대한 역량은 갖추었는가?
- 예비 사회인으로서 다양한 경험을 하였는가?
- 팀워크, 리더십, 갈등해결 등 대인관계능력을 갖추었는가?
- 지원 기업에 대한 이해도를 갖추고자 노력하였는가?
- 다양한 변화에 대응할 인문학적 소양 등을 갖추었는가?

##가점요인
- 역량 중심으로 고민을 담아 성의껏 작성하였는가?
- 경험 위주로 근거를 갖고 구체적으로 기술하고 있는가?
- 표현능력(논리력, 문장력 등)이 우수한가?

##감점요인
- 동일 내용을 반복하고 있는가?
- 제시된 분량에 턱없이 모자라는가? (70% 이하)
- 원론적이고 추상적인 내용으로 기술하고 있는가?
- 질문과 무관한 내용이 담겨 있는가?
- 오타 및 부정확한 표현은 없는가?
- 타사 지원서를 복사하여 지원 회사명이 잘못 기재하였는가?

"""

def get_code(company_name: str) -> dict:
    """
    사용자가 회사 이름을 전달하면 해당 회사의 종목 코드를 dict형태로 반환한다.
    """
    df = pd.read_csv("C:/Users/playdata2/Downloads/myagent/skn25/data_2058_20260323.csv", encoding='cp949')
    return df[df['한글 종목명'].apply(lambda x : x.find(company_name) > -1)].to_json()

def get_company_info(company_code:str) -> dict:
    """
    종목 코드를 사용하여 회사 정보를 반환하는 코드
    회사명에서 종목코드 추출은 get_code가 담당
    company_code예시 : 005930

    """
    url = f"https://wts-info-api.tossinvest.com/api/v2/stock-infos/A{company_code}/overview"
    rt = httpx.get(url).json()
    return rt['result']['company']

company_agent = Agent(
    name='company_agent',
    model="gemini-2.5-flash",
    description='기업 정보를 사용자한테 전달하는 에이전트',
    instruction='사용자가 작성 자기소개서에서 기업에 대한 정보를 도구를 사용하여 출력하는 에이전트',
    output_key='company_info',
    tools = [get_code, get_company_info]
)

news_agent = Agent(
    name="news_agent",
    model="gemini-2.5-flash",
    instruction="사용자가 작성 자기소개서를 바탕으로 해당 기업의 주요 뉴스를 요약할 것. 앞으로 전망도 이야기할 것",
    tools=[google_search],
    output_key='news_info'
)

resume_agent = Agent(  
    name='resume_agent',
    model="gemini-2.5-flash",
    description='사용자가 작성 자기소개서를 주어진 지시에 따라 수정하여 출력하는 에이전트',
    instruction=INSTRUCTIONS,
    output_key='resume_info'
)

parallel_fet = ParallelAgent(
    name="multi_info_fetcher",
    sub_agents=[company_agent, news_agent, resume_agent],
    description="여러 정보를 동시에 요청한 내용대로 수집 및 출력"
)

summarizer = Agent(
    name='final_agent',
    model="gemini-2.5-flash",
    instruction="""
    에이전트들이 수집한 자료를 바탕으로 최종 자기 소개서를 작성하세요.
    - 기업 정보 : {company_info}
    - 기업 뉴스 : {news_info}
    - 자기소개서 수정안 : {resume_info}

    위의 정보를 바탕으로 최종 자기 소개서를 작성해서 출력하세요
    """
)

root_agent = SequentialAgent(
    name="ai_resume_system",
    sub_agents=[parallel_fet, summarizer],
    description="최종 정보를 바탕으로 자기소개서를 완성하는 에이전트"
)

claude 연계

{
  "preferences": {
    "coworkWebSearchEnabled": true,
    "ccdScheduledTasksEnabled": false,
    "coworkScheduledTasksEnabled": false,
    "sidebarMode": "chat"
  },
  "mcpServers": {
    "resume-evaluator": {
      "command": "C:/skn-25/wk13/myagent/.venv/Scripts/python.exe",
      "args": [
        "C:/skn-25/wk13/myagent/ai_agent/agent.py"
      ],
      "env": {
        "GOOGLE_API_KEY": ㅇㅇㅇㅇㅇㅇ
      }
    }
  }
}

MCP로 쓸 코드 등록

from mcp.server.fastmcp import FastMCP
mcp = FastMCP("ADK-resume")

@mcp.tool()
async def modify_resume(resume: str) -> str:
    """
    지원자의 자기소개서와 해당 지원하는 회사 이름과 직무를 입력하면
    에이전트 내용 수정하여 최종 수정된 자기소개서를 출력한다.
    """
    response = root_agent.run_async(resume)
    return str(response)

if __name__=="__main__":
    mcp.run()

from google.adk.tools import google_search
from google.adk.agents import Agent, ParallelAgent, SequentialAgent
from geopy.geocoders import Nominatim
import httpx
import pandas as pd

INSTRUCTIONS = """
당신은 채용 전문가 에이전트입니다.
사용자가 입력한 자기소개서를 바탕으로 아래 기준에 따라서 자기소개서를 수정할 것

##기본 평가항목
- 질문 항목에 대한 답변으로 적절한가?
- 지원자의 지원동기 등 입사에 대한 열의가 있는가?
- 핵심가치 및 인재상에 부합하는 인재인가?
- 지원 직무에 대한 역량은 갖추었는가?
- 예비 사회인으로서 다양한 경험을 하였는가?
- 팀워크, 리더십, 갈등해결 등 대인관계능력을 갖추었는가?
- 지원 기업에 대한 이해도를 갖추고자 노력하였는가?
- 다양한 변화에 대응할 인문학적 소양 등을 갖추었는가?

##가점요인
- 역량 중심으로 고민을 담아 성의껏 작성하였는가?
- 경험 위주로 근거를 갖고 구체적으로 기술하고 있는가?
- 표현능력(논리력, 문장력 등)이 우수한가?

##감점요인
- 동일 내용을 반복하고 있는가?
- 제시된 분량에 턱없이 모자라는가? (70% 이하)
- 원론적이고 추상적인 내용으로 기술하고 있는가?
- 질문과 무관한 내용이 담겨 있는가?
- 오타 및 부정확한 표현은 없는가?
- 타사 지원서를 복사하여 지원 회사명이 잘못 기재하였는가?

"""

def get_code(company_name: str) -> dict:
    """
    사용자가 회사 이름을 전달하면 해당 회사의 종목 코드를 dict형태로 반환한다.
    """
    df = pd.read_csv(r"C:\skn-25\wk13\myagent\ai_agent\data_2058_20260323.csv", encoding='cp949')
    return df[df['한글 종목명'].apply(lambda x : x.find(company_name) > -1)].to_json()

def get_company_info(company_code:str) -> dict:
    """
    종목 코드를 사용하여 회사 정보를 반환하는 코드
    회사명에서 종목코드 추출은 get_code가 담당
    company_code예시 : 005930

    """
    url = f"https://wts-info-api.tossinvest.com/api/v2/stock-infos/A{company_code}/overview"
    rt = httpx.get(url).json()
    return rt['result']['company']

company_agent = Agent(
    name='company_agent',
    model="gemini-2.5-flash",
    description='기업 정보를 사용자한테 전달하는 에이전트',
    instruction='사용자가 작성 자기소개서에서 기업에 대한 정보를 도구를 사용하여 출력하는 에이전트',
    output_key='company_info',
    tools = [get_code, get_company_info]
)

news_agent = Agent(
    name="news_agent",
    model="gemini-2.5-flash",
    instruction="사용자가 작성 자기소개서를 바탕으로 해당 기업의 주요 뉴스를 요약할 것. 앞으로 전망도 이야기할 것",
    tools=[google_search],
    output_key='news_info'
)

resume_agent = Agent(  
    name='resume_agent',
    model="gemini-2.5-flash",
    description='사용자가 작성 자기소개서를 주어진 지시에 따라 수정하여 출력하는 에이전트',
    instruction=INSTRUCTIONS,
    output_key='resume_info'
)

parallel_fet = ParallelAgent(
    name="multi_info_fetcher",
    sub_agents=[company_agent, news_agent, resume_agent],
    description="여러 정보를 동시에 요청한 내용대로 수집 및 출력"
)

summarizer = Agent(
    name='final_agent',
    model="gemini-2.5-flash",
    instruction="""
    에이전트들이 수집한 자료를 바탕으로 최종 자기 소개서를 작성하세요.
    - 기업 정보 : {company_info}
    - 기업 뉴스 : {news_info}
    - 자기소개서 수정안 : {resume_info}

    위의 정보를 바탕으로 최종 자기 소개서를 작성해서 출력하세요
    """
)

root_agent = SequentialAgent(
    name="ai_resume_system",
    sub_agents=[parallel_fet, summarizer],
    description="최종 정보를 바탕으로 자기소개서를 완성하는 에이전트"
)

from mcp.server.fastmcp import FastMCP
mcp = FastMCP("ADK-resume")

@mcp.tool()
async def modify_resume(resume: str) -> str:
    """
    지원자의 자기소개서와 해당 지원하는 회사 이름과 직무를 입력하면
    에이전트 내용 수정하여 최종 수정된 자기소개서를 출력한다.
    """
    response = root_agent.run_async(resume)
    return str(response)

if __name__=="__main__":
    mcp.run()

소개타임… - crewai

• uv pip install crewai

멀티 에이전트를 위한 패키지

검색엔진 발전시키기

https://serpapi.com/dashboard

from crewai import Agent, Task, Crew, Process, LLM
from crewai_tools import SerperDevTool

from dotenv import load_dotenv
load_dotenv()
import os
os.environ["OPENAI_API_KEY"] = "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"

os.environ["SERPER_API_KEY"] ="2497f33b9d1dc6fc8804c0bae6088833d0783959140895b7e7630674e6956e8e"

search_tool = SerperDevTool()

llm = LLM(
    model='gpt-4.1-2025-04-14',
    temperature=0.4
)

Casanova_agent = Agent(
    role="연애 전문가",
    goal="이성이 좋아할 만한 매력적인 글을 작성한다.",
    backstory="""
    당신은 연애 편지를 잘 쓰는 작가입니다.
    다른 사람이 초안을 작성한 내용을 당신의 탁월한 감각으로 편집합니다.
    """,
    verbose=True,
    allow_delegation=False,
    llm=llm
)

Writer_agent = Agent(
    role='편지 작성자',
    goal="이성이 좋아할 만한 편지의 초안을 작성한다.",
    backstory="""
    이성에게 편지를 자주 작성하며, 상대방의 취향을 잘 파악하는 사람입니다.
    """,
    verbose=True,
    allow_delegation=False,
    llm=llm,
    tools=[search_tool]
)

write_task = Task(
    description="""
    상대방이 좋아하는 것들 : 샤넬, 플스5, 맥북
    MBTI : ISTP
    위 정보를 바탕으로 편지의 초안을 작성할 것.
    필요하다면 위의 좋아하는 것들을 검색 도구를 사용하여 가격이나 후기등을 찾아 편지에 반영할 것
    """,
    expected_output="상대방의 취향과 성향을 반영한 편지 초안.",
    agent=Writer_agent
)

modify_task = Task(
    description="""
    제시된 편지 초안을 카사노바 입장에서 더욱 매력적으로 수정할 것.
    """,
    # 사용자의 이모지 사용 금지 규칙을 반영하여 관련 지시어 제거
    expected_output="달콤하고 매력적인 말투의 편지 내용 (약 300~500 단어).",
    agent=Casanova_agent
)

crew = Crew(
    # 논리적 흐름에 따라 초안 작성자(Writer)를 먼저 배치하는 것이 일반적이나,
    # Task의 순서가 Process.sequential에 의해 결정되므로 agents 배열의 순서는 실행에 영향을 주지 않습니다.
    agents=[Writer_agent, Casanova_agent],
    tasks=[write_task, modify_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff()

# ===========================================
# 5. 결과 출력
# ===========================================
print("\n\n=== 최종 결과 ===\n")
# 최신 버전의 CrewOutput 객체에서 원시 텍스트(raw text)를 추출하여 출력합니다.
print(result.raw)

스파크, 엘라스틱 서치 해봐야하려나

Day 3

Claude 로컬 MCP로 연결해보기

generator는 co-routine 활용

한 꺼번에 주는지, 나눠서 주는지 다름

동기& 비동기

동기= 김밥가게

• 요리사가 김밥을 하나하나 썰어야만 함. 주문이 순차적으로 처리됨.

비동기 = 맥도날드

• 여러 개의 키오스크로 동시에 주문 받기 가능. 여러 태스크를 병렬적으로 처리할 수 있음
• 불고기 버거와 쿼터파운드 치즈 버거는 조리 시간이 달라서 주문에 따른 결과 순서를 보장할 수 없음

agent.py 수정

uv pip install --upgrade opentelemetry-api opentelemetry-sdk opentelemetry-semantic-conventions
uv pip install --upgrade google-genai

from google.adk.tools import google_search
from google.adk.agents import Agent, ParallelAgent, SequentialAgent
from geopy.geocoders import Nominatim
import httpx
import pandas as pd
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types

def get_code(company_name: str) -> dict:
    """
    사용자가 회사 이름을 전달하면 해당 회사의 종목 코드를 dict형태로 반환한다.
    """
    df = pd.read_csv(r"C:\skn-25\wk13\myagent\ai_agent\data_2058_20260323.csv", encoding='cp949')
    return df[df['한글 종목명'].apply(lambda x : x.find(company_name) > -1)].to_json()

def get_company_info(company_code:str) -> dict:
    """
    종목 코드를 사용하여 회사 정보를 반환하는 코드
    회사명에서 종목코드 추출은 get_code가 담당
    company_code예시 : 005930

    """
    url = f"https://wts-info-api.tossinvest.com/api/v2/stock-infos/A{company_code}/overview"
    rt = httpx.get(url).json()
    return rt['result']['company']

company_agent = Agent(
    name='company_agent',
    model="gemini-2.5-flash",
    description='기업 정보를 사용자한테 전달하는 에이전트',
    instruction='사용자가 작성 자기소개서에서 기업에 대한 정보를 도구를 사용하여 출력하는 에이전트',
    output_key='company_info',
    tools = [get_code, get_company_info]
)

news_agent = Agent(
    name="news_agent",
    model="gemini-2.5-flash",
    instruction="사용자가 작성 자기소개서를 바탕으로 해당 기업의 주요 뉴스를 요약할 것. 앞으로 전망도 이야기할 것",
    tools=[google_search],
    output_key='news_info'
)

resume_agent = Agent(  
    name='resume_agent',
    model="gemini-2.5-flash",
    description='사용자가 작성 자기소개서를 주어진 지시에 따라 수정하여 출력하는 에이전트',
    instruction=INSTRUCTIONS,
    output_key='resume_info'
)

parallel_fet = ParallelAgent(
    name="multi_info_fetcher",
    sub_agents=[company_agent, news_agent, resume_agent],
    description="여러 정보를 동시에 요청한 내용대로 수집 및 출력"
)

summarizer = Agent(
    name='final_agent',
    model="gemini-2.5-flash",
    instruction="""
    에이전트들이 수집한 자료를 바탕으로 최종 자기 소개서를 작성하세요.
    - 기업 정보 : {company_info}
    - 기업 뉴스 : {news_info}
    - 자기소개서 수정안 : {resume_info}

    위의 정보를 바탕으로 최종 자기 소개서를 작성해서 출력하세요
    """
)

root_agent = SequentialAgent(
    name="ai_resume_system",
    sub_agents=[parallel_fet, summarizer],
    description="최종 정보를 바탕으로 자기소개서를 완성하는 에이전트"
)

from mcp.server.fastmcp import FastMCP
mcp = FastMCP("ADK-resume")

session_service = InMemorySessionService()
runner = Runner(
    agent=root_agent,
    app_name="resume_app",
    session_service=session_service
)

@mcp.tool()
async def modify_resume(resume: str) -> str:
    """
    지원자의 자기소개서와 해당 지원하는 회사 이름과 직무를 입력하면
    에이전트 내용 수정하여 최종 수정된 자기소개서를 출력한다.
    """
    try:
            session = await session_service.create_session(
                app_name="resume_app",
                user_id="user"
            )
            content = types.Content(role='user', parts=[types.Part(text=resume)])
            final_response = ""
            for event in runner.run(
                session_id=session.id,
                user_id="user",
                new_message=content
            ):
                if event.is_final_response():
                    if event.content and event.content.parts:
                        final_response = event.content.parts[0].text
            return final_response
    except Exception as e:
        raise

if __name__=="__main__":
    mcp.run()

Langgraph로 설계하기

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_classic.agents import (
    AgentExecutor,
    create_tool_calling_agent,
    tool,
)
from langgraph.graph import StateGraph, START, END

그 전에 정보를 어떻게 뽑아올건지 생각해보자

from selenium import webdriver
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.chrome.service import Service as ChromeService
from selenium.webdriver.common.by import By
driver = webdriver.Chrome(service=ChromeService(ChromeDriverManager().install()))

driver.get("https://navercomp.wisereport.co.kr/v2/company/c1010001.aspx?cmp_cd=001510")

from bs4 import BeautifulSoup
report = BeautifulSoup(driver.page_source).find_all('table', class_='gHead01 all-width')

import pandas as pd
import io

report_text = "\n".join([pd.read_html(io.StringIO(str(x)))[0].to_json() for x in report])

from dotenv import load_dotenv

load_dotenv()
llm  = ChatOpenAI(model='gpt-4o', temperature=0.2)
response = llm.invoke(f"제공된 데이터 분석해봐 {report_text}")
print(response.content)

from selenium import webdriver
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.chrome.service import Service as ChromeService
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
options = Options()
options.add_argument('--headless=new')
driver = webdriver.Chrome(service=ChromeService(ChromeDriverManager().install()), options=options)

재무제표 보는 agent 설계

#@tool
def finance_report(company_code : str) -> str:
    """
    company_code : 회사 종목 코드
    종목코드 회사의 연간, 분기 재무제표를 값을 리턴하는 함수

    return : 연간, 분기 재무제표 값
    """
    options = Options()
    options.add_argument('--headless=new')
    driver = webdriver.Chrome(service=ChromeService(ChromeDriverManager().install()), options=options)
    driver.get(f"https://navercomp.wisereport.co.kr/v2/company/c1010001.aspx?cmp_cd={company_code}")
    report = BeautifulSoup(driver.page_source).find_all('table', class_='gHead01 all-width')
    report_text = "\n".join([pd.read_html(io.StringIO(str(x)))[0].to_json() for x in report])
    return report_text

@tool
def get_code(company_name: str) -> dict:
    """
    company_name : 회사명
    회사명으로 뉴스를 검색하여 해당 내용을 반환한다.

    return : 회사명으로 검색된 뉴스의 총 텍스트값을 리턴
    """
    df = pd.read_csv(r"C:\skn-25\wk13\day3\data_2058_20260323.csv", encoding='cp949')
    return df[df['한글 종목명'].apply(lambda x : x.find(company_name) > -1)].to_json()

tools = [finance_report, get_code ]

prompt = ChatPromptTemplate([
    ( "system", """당신은 재무 분석가입니다. CFO 수준의 재무제표 전문가입니다.
     사용자가 제공한 재무제표 데이터를 기반으로 재무 건전성, 수익성, 성장성, 안정성을 종합 분석할 것
     그리고 재무 분석가의 의견으로 이야기할 것 """ ),
     ("human", "{input}"),
     ('placeholder', "{agent_scratchpad}")]
)
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = agent_executor.invoke({"input" : "광전자"})

뉴스 찾는 Agent 설계

from langchain_community.document_loaders import WebBaseLoader
from bs4 import SoupStrainer
import httpx

def get_news(company_name : str) -> str:
    global total_naver_url
    client_id = "YXw4dmid0O2qTQNfhcY2"
    client_secret = "BksteLLT_U"
    url = "https://openapi.naver.com/v1/search/news.json"
    headers = {
        "X-Naver-Client-Id": client_id,
        "X-Naver-Client-Secret": client_secret
    }
    params = {'query' : company_name , 'display' : 100, 'start': 1, "sort" : "date"}
    response = httpx.get(url, headers=headers, params=params)
    total_naver_url = [url['link'] for url in response.json()['items'] if 'naver.com' in url['link']]
    bs4_kwargs = {
        'parse_only' : SoupStrainer("div", id="newsct_article")
    }
    rt = WebBaseLoader(total_naver_url[0],  bs_kwargs=bs4_kwargs)

    return " ".join([WebBaseLoader(x,bs_kwargs=bs4_kwargs).load()[0].page_content.strip() for x in total_naver_url])

@tool
def get_news(company_name : str) -> str:
    """  
    company_name : 회사명
    회사명으로 뉴스를 검색하여 해당 내용을 반환한다.

    return : 회사명으로 검색된 뉴스의 총 텍스트값을 리턴
    """
    global total_naver_url
    client_id = "YXw4dmid0O2qTQNfhcY2"
    client_secret = "BksteLLT_U"
    url = "https://openapi.naver.com/v1/search/news.json"
    headers = {
        "X-Naver-Client-Id": client_id,
        "X-Naver-Client-Secret": client_secret
    }
    params = {'query' : company_name , 'display' : 100, 'start': 1, "sort" : "date"}
    response = httpx.get(url, headers=headers, params=params)
    total_naver_url = [url['link'] for url in response.json()['items'] if 'naver.com' in url['link']]
    bs4_kwargs = {
        'parse_only' : SoupStrainer("div", id="newsct_article")
    }
    rt = WebBaseLoader(total_naver_url[0],  bs_kwargs=bs4_kwargs)

    return " ".join([WebBaseLoader(x,bs_kwargs=bs4_kwargs).load()[0].page_content.strip() for x in total_naver_url])

prompt_news = ChatPromptTemplate([
    ( "system", """당신은 기업 분석가입니다. 제공된 뉴스를 바탕으로 현재 기업의 상태, 미래 전망,
     사업의 건전성등을 기업에 대한 모든 정보를 출력합니다.""" ),
     ("human", "{input}"),
     ('placeholder', "{agent_scratchpad}")]
)
news_tools = [get_news]
news_agent = create_tool_calling_agent(llm, news_tools, prompt_news)
news_agent_executor = AgentExecutor(agent=news_agent, tools=news_tools, verbose=True)
news_result = news_agent_executor.invoke({"input" : "광전자"})

구조를 지정해볼까요…?

• pydantic

from pydantic import BaseModel, Field

from pydantic import BaseModel, Field
class CompanyAnalysisNews(BaseModel):
    current_status : str = Field(description="현재 기업의 상태 요약")
    future_outlook : str = Field(description="미래 전망 및 성장 가능성")
    business_health : str = Field(description="사업의 건정성 및 리스크 요인")
    core_keyword: list[str] = Field(description="기업과 관련된 핵심 뉴스 키워드")

structured_llm = llm.with_structured_output(CompanyAnalysisNews)
final_result = structured_llm.invoke(
    f"다음 기업 분석 텍스트를 분석하여 구조화된 데이터로 추출할 것\n\n {news_result['output']}"
)

final_result.model_dump_json()

시가저가거래량

stock_url = "https://m.stock.naver.com/front-api/external/chart/domestic/info?symbol=017900&requestType=1&startTime=20230614&endTime=20240513&timeframe=day"

data = eval(httpx.get(stock_url).text.strip())

import json
stock_json = json.dumps(data)

from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
load_dotenv()
llm = ChatOpenAI(model="gpt-4o",   temperature=0)

rt = llm.invoke(f"데이터 분석 : {stock_json}")
print(rt.content)

@tool
def get_data(company_code : str, sdate : str, edate : str) -> str:
    """
    company_code  : 종목코드 (예: 005930)
    sdate : 데이터 시작날짜 (예: 20260102)
    edate : 데이터 종료날짜 (예: 20260324)
    주식 데이터를 가져오는 함수
    return 종목의 시가, 고가, 저가,종가, 거래량 등의 데이터
    """
    stock_url = f"https://m.stock.naver.com/front-api/external/chart/domestic/info?symbol={company_code}&requestType=1&startTime={sdate}&endTime={edate}&timeframe=day"
    data = eval(httpx.get(stock_url).text.strip())
    return json.dumps(data)

프롬프트 설계

stock_prompt = ChatPromptTemplate([
    ( "system", """당신은 20년 경력의 전업 투자가입니다. 주가의 움직임 속에 모든 정보와 투자자의 심리가 있다고 판단하고 있습니다. 주어진 데이터를 통해서 기술적 분석, 통계등을 이용해서 상태를 진단하고, 냉철하게 분석할 것
     사용자가 종목명을 입력하면 도구를 활용해서 종목 코드를 가져오고, 도구를 활용해서
     데이터를 가져올때 제시된 날짜부터 약 300일치 데이터를 가져올 것""" ),
     ("human", "{input}"),
     ('placeholder', "{agent_scratchpad}")]
)
stock_tools = [get_data, get_code ]
stock_agent = create_tool_calling_agent(llm, stock_tools, prompt)
stock_agent_executor = AgentExecutor(agent=stock_agent, tools=stock_tools, verbose=True)
result = stock_agent_executor.invoke({"input" : "2026년 3월 25일 기준 광전자 주가 정보를 분석해봐"})

랭그래프로 연결하기

• ainvoke( ) : 비동기 실행

async def stock_node(state : CompanyState):
    """
    사용자 입력한 회사 주가정보를 분석하여 리턴하는 노드
    """
    stock_prompt = ChatPromptTemplate([
    ( "system", """당신은 20년 경력의 전업 투자가입니다. 주가의 움직임 속에 모든 정보와 투자자의 심리가 있다고 판단하고 있습니다. 주어진 데이터를 통해서 기술적 분석, 통계등을 이용해서 상태를 진단하고, 냉철하게 분석할 것
     사용자가 종목명을 입력하면 도구를 활용해서 종목 코드를 가져오고, 도구를 활용해서
     데이터를 가져올때 제시된 날짜부터 약 300일치 데이터를 가져올 것""" ),
     ("human", "{input}"),
     ('placeholder', "{agent_scratchpad}")]
)
    stock_tools = [get_data, get_code ]
    stock_agent = create_tool_calling_agent(llm, stock_tools, stock_prompt)
    stock_agent_executor = AgentExecutor(agent=stock_agent, tools=stock_tools, verbose=True)
    result = stock_agent_executor.invoke({"input" : state['question']})
    return {'company_stock' : result['output']}

async def summarize_node(state : CompanyState):
    """에이전트 전달한 자료를 기반으로 최종 종목에서 판단하는 노드"""
    prompt = ChatPromptTemplate.from_messages([
        ('system', """
        에이전트들이 정리한 자료를 바탕으로 최종 종목에 대해서 매수의견을 작성하세요
         - 기업에 대한 재무제표 정리: {company_finance}
         - 기업에 대한 뉴스정리 : {company_news}
         - 기업에 대한 주가정보 정리 : {company_stock}
       
         위의 제공된 정보를 바탕으로 최종 의견을 제시할 것
        """),
        ('human', "{question}")
    ])
    chain = prompt | llm
    result = await chain.ainvoke( {
        'company_finance' : state.get('company_finance', ""),
        'company_news' : state.get('company_news', ""),
        'company_stock' : state.get('company_stock', ""),
        'question' : state['question']}
    )

    return {'final_report' : result.content}

workflow = StateGraph(CompanyState)
workflow.add_node('finance_node', finance_node)
workflow.add_node('news_node', news_node)
workflow.add_node('stock_node', stock_node)
workflow.add_node('summarize_node', summarize_node)

workflow.add_edge(START, 'finance_node')
workflow.add_edge(START, 'news_node')
workflow.add_edge(START, 'stock_node')

workflow.add_edge('finance_node', 'summarize_node')
workflow.add_edge('news_node', 'summarize_node')
workflow.add_edge('stock_node', 'summarize_node')

workflow.add_edge('summarize_node', END)

app= workflow.compile()

png = app.get_graph().draw_mermaid_png()

with open("./stock_graph.png", "wb") as f:
    f.write(png)

await app.ainvoke({'question':'광전자에 대해서 2026년 3월 25일 기준으로 분석해서 알려줘'})

mcp 설계

uv init stock_agent #폴더 만들기

cd stock_agent 

uv python install 3.12

C:\skn-25\wk13\stock_agent>uv pip list
Using Python 3.12.12 environment at: C:\Users\Ming9\AppData\Roaming\uv\python\cpython-3.12.12-windows-x86_64-none
Package Version
------- -------
pip     26.0.1

C:\skn-25\wk13\stock_agent>

uv venv
.venv\Scripts\activate

uv pip install -r langchain_requirements.txt
uv pip install fastmcp

평소처럼 mcp 등록 후

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_classic.agents import ( AgentExecutor, create_tool_calling_agent, tool)
from langgraph.graph import StateGraph, START, END
import requests 
from selenium import webdriver
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.chrome.service import Service as ChromeService
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from langchain_community.document_loaders import WebBaseLoader
from bs4 import SoupStrainer
import json 
from typing import TypedDict 
import asyncio 
from typing import TypedDict 
import asyncio 
from pydantic import BaseModel, Field 
from mcp.server.fastmcp import FastMCP
from bs4 import BeautifulSoup
import pandas as pd
import io 
from dotenv import load_dotenv
load_dotenv()

llm = ChatOpenAI(model='gpt-5.4-mini-2026-03-17')

@tool
def finance_report(company_code : str) -> str:
    """ 
    company_code : 회사 종목 코드
    종목코드 회사의 연간, 분기 재무제표를 값을 리턴하는 함수 

    return : 연간, 분기 재무제표 값 
    """
    options = Options()
    options.add_argument('--headless=new')
    options.binary_location = r"C:\Program Files\Google\Chrome\Application\chrome.exe"
    driver = webdriver.Chrome(service=ChromeService(ChromeDriverManager().install()), options=options)
    driver.get(f"https://navercomp.wisereport.co.kr/v2/company/c1010001.aspx?cmp_cd={company_code}")
    report = BeautifulSoup(driver.page_source).find_all('table', class_='gHead01 all-width')
    report_text = "\n".join([pd.read_html(io.StringIO(str(x)))[0].to_json() for x in report])
    driver.close()
    return report_text

@tool 
def get_code(company_name: str) -> dict: 
    """
    사용자가 회사 이름을 전달하면 해당 회사의 종목 코드를 dict형태로 반환한다. 
    """
    df = pd.read_csv(r"C:\skn-25\wk13\stock_agent\data_2058_20260323.csv", encoding='cp949')
    return df[df['한글 종목명'].apply(lambda x : x.find(company_name) > -1)].to_json()

@tool
def get_news(company_name : str) -> str: 
    """  
    company_name : 회사명
    회사명으로 뉴스를 검색하여 해당 내용을 반환한다. 

    return : 회사명으로 검색된 뉴스의 총 텍스트값을 리턴
    """
    global total_naver_url
    client_id = "YXw4dmid0O2qTQNfhcY2"
    client_secret = "BksteLLT_U"
    url = "https://openapi.naver.com/v1/search/news.json"
    headers = {
        "X-Naver-Client-Id": client_id,
        "X-Naver-Client-Secret": client_secret
    }
    params = {'query' : company_name , 'display' : 100, 'start': 1, "sort" : "date"}
    response = requests.get(url, headers=headers, params=params)
    total_naver_url = [url['link'] for url in response.json()['items'] if 'naver.com' in url['link']]
    bs4_kwargs = {
        'parse_only' : SoupStrainer("div", id="newsct_article")
    }
    rt = WebBaseLoader(total_naver_url[0],  bs_kwargs=bs4_kwargs)

    return " ".join([WebBaseLoader(x,bs_kwargs=bs4_kwargs).load()[0].page_content.strip() for x in total_naver_url])

@tool
def get_data(company_code : str, sdate : str, edate : str) -> str:
    """
    company_code  : 종목코드 (예: 005930)
    sdate : 데이터 시작날짜 (예: 20260102)
    edate : 데이터 종료날짜 (예: 20260324)
    주식 데이터를 가져오는 함수 
    return 종목의 시가, 고가, 저가,종가, 거래량 등의 데이터 
    """
    stock_url = f"https://m.stock.naver.com/front-api/external/chart/domestic/info?symbol={company_code}&requestType=1&startTime={sdate}&endTime={edate}&timeframe=day"
    data = eval(requests.get(stock_url).text.strip())
    return json.dumps(data)

class CompanyState(TypedDict):
    question : str 
    company_finance : str 
    company_news : str 
    company_stock : str 
    final_report : str 

async def finance_node(state : CompanyState):
    """ 
    사용자 입력한 회사의 재무제표 정보를 가져와서 분석하여 리턴하는 노드
    """
    prompt = ChatPromptTemplate([
                ( "system", """당신은 금융 감독위원회의 재무 분석가입니다. 검찰 수준으로 검토할 것 
                사용자가 제공한 재무제표 데이터를 기반으로 재무 건전성, 수익성, 성장성, 안정성을 종합 분석할 것
                연간, 분기별로 별도로 분석해 
                그리고 재무 분석가의 의견으로 이야기할 것 """ ),
                ("human", "{input}"),
                ('placeholder', "{agent_scratchpad}")]
            )

    tools = [finance_report, get_code ]

    agent = create_tool_calling_agent(llm, tools, prompt)
    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=False, handle_parsing_errors=True)

    result = await agent_executor.ainvoke({"input" : state['question'] })
    return {'company_finance' : result['output']}

async def news_node(state : CompanyState):
    """ 
    사용자 입력한 회사의 뉴스 정보를 분석하여 리턴하는 노드
    """
    prompt_news = ChatPromptTemplate([
    ( "system", """당신은 기업 분석가입니다. 제공된 뉴스를 바탕으로 현재 기업의 상태, 미래 전망, 
     사업의 건전성등을 기업에 대한 모든 정보를 출력합니다.""" ),
     ("human", "{input}"),
     ('placeholder', "{agent_scratchpad}")]
)
    news_tools = [get_news]
    news_agent = create_tool_calling_agent(llm, news_tools, prompt_news)
    news_agent_executor = AgentExecutor(agent=news_agent, tools=news_tools, verbose=False, handle_parsing_errors=True)
    result =  await news_agent_executor.ainvoke({"input" : state['question'] })
    return {'company_news' : result['output']}

async def stock_node(state : CompanyState):
    """ 
    사용자 입력한 회사 주가정보를 분석하여 리턴하는 노드
    """
    stock_prompt = ChatPromptTemplate([
    ( "system", """당신은 20년 경력의 전업 투자가입니다. 주가의 움직임 속에 모든 정보와 투자자의 심리가 있다고 판단하고 있습니다. 주어진 데이터를 통해서 기술적 분석, 통계등을 이용해서 상태를 진단하고, 냉철하게 분석할 것 
     사용자가 종목명을 입력하면 도구를 활용해서 종목 코드를 가져오고, 도구를 활용해서 
     데이터를 가져올때 제시된 날짜부터 약 300일치 데이터를 가져올 것""" ),
     ("human", "{input}"),
     ('placeholder', "{agent_scratchpad}")]
)
    stock_tools = [get_data, get_code ]
    stock_agent = create_tool_calling_agent(llm, stock_tools, stock_prompt)
    stock_agent_executor = AgentExecutor(agent=stock_agent, tools=stock_tools, verbose=False, handle_parsing_errors=True)
    result = await stock_agent_executor.ainvoke({"input" : state['question']})
    return {'company_stock' : result['output']}

async def summarize_node(state : CompanyState):
    """에이전트 전달한 자료를 기반으로 최종 종목에서 판단하는 노드"""
    prompt = ChatPromptTemplate.from_messages([
        ('system', """ 
        에이전트들이 정리한 자료를 바탕으로 최종 종목에 대해서 매수의견을 작성하세요 
         - 기업에 대한 재무제표 정리: {company_finance}
         - 기업에 대한 뉴스정리 : {company_news}
         - 기업에 대한 주가정보 정리 : {company_stock}
        
         위의 제공된 정보를 바탕으로 최종 의견을 제시할 것
        """),
        ('human', "{question}")
    ])
    chain = prompt | llm 
    result = await chain.ainvoke( {
        'company_finance' : state.get('company_finance', ""),
        'company_news' : state.get('company_news', ""),
        'company_stock' : state.get('company_stock', ""), 
        'question' : state['question']}
    )

    return {'final_report' : result.content}

workflow = StateGraph(CompanyState)
workflow.add_node('finance_node', finance_node)
workflow.add_node('news_node', news_node)
workflow.add_node('stock_node', stock_node)
workflow.add_node('summarize_node', summarize_node)
workflow.add_edge(START, 'finance_node')
workflow.add_edge(START, 'news_node')
workflow.add_edge(START, 'stock_node')
workflow.add_edge('finance_node', 'summarize_node')
workflow.add_edge('news_node', 'summarize_node')
workflow.add_edge('stock_node', 'summarize_node')
workflow.add_edge('summarize_node', END)
app = workflow.compile()

mcp = FastMCP("stock analysis")

@mcp.tool()
async def stock_anlysis(question : str) -> str:
    """ 
    사용자가 주식 종목을 이야기하면 해당 종목에 대해서 분석해서 매수 의견을 반환하여 출력 
    :question : 2026년 3월 25일 기준으로 삼성전자 종목을 분석해봐 
    """
    try:
        result = await app.ainvoke({'question' : question})
        return result['final_report']
    except Exception as e:
        raise e  
    
if __name__=="__main__":
    mcp.run()

Day 4

기술 블로그 리스트

• 우아한 기술 블로그

• 쿠팡 기술 블로그

• 네이버 기술 블로그

• 인스타그램 기술 블로그

OPEN ai client로 영상에다 자막 달아주기

import yt_dlp
import static_ffmpeg
static_ffmpeg.add_paths()

url = "https://www.youtube.com/watch?v=H26xnpL-ei0"

setting = {
    'format' : 'bestvideo+bestaudio/best',
    'outtmpl' : '%(title)s.%(ext)s',
    'keepvideo' : True ,
    'postprocessors' : [
        {
            'key' : 'FFmpegExtractAudio',
            'preferredcodec' : 'mp3',
            'preferredquality' : '96'
        }
    ]
}

with yt_dlp.YoutubeDL(setting) as ydl:
    ydl.download(url)

from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()

client = OpenAI()

with open("NVIDIA GTC 2026 Open Models Panel Highlights with Jensen Huang.mp3", 'rb') as f:
    transcript = client.audio.transcriptions.create(
        model='whisper-1',
        response_format='srt',
        file=f
    )

with open("NVIDIA GTC 2026 Open Models Panel Highlights with Jensen Huang.srt", 'w') as f:
    f.write(transcript)

response = client.chat.completions.create(
    model='gpt-5.4-2026-03-05',
    messages=[
        {'role' : 'system', 'content' : '''당신은 번역가입니다. 주여진 영어를 한글로 번역할 것
         제공 형식은 SRT 형식이므로 형태 그대로 반듯이 유지할 것'''},
         {'role' : 'user', 'content' : transcript}
    ],
    temperature=0.2
)

VLC 미디어 재생기로 틀면 자막으로 쓸 수 있게 나옴.

요약 진행

response2 = client.chat.completions.create(
    model='gpt-5.4-2026-03-05',
    messages=[
        {'role' : 'system', 'content' : '''당신은 요약전문가입니다. 제공된 스크립트에서 중요 내용을
         요약해서 출력하세요. 그리고 해당 도메인을 이해할수 용어나 지식들이 있다면 설명도 같이 할 것'''},
         {'role' : 'user', 'content' : transcript}
    ],
    temperature=0.2
)

summary = response2.choices[0].message.content.strip()

요약 읽어줌

audio = client.audio.speech.create(
    model='gpt-4o-mini-tts-2025-12-15',
    input=summary[:3000],
    voice='alloy'
)

audio_file = audio.content

with open("./요약.mp3", 'wb') as f:
    f.write(audio_file)

OPEN ai client로 영상에다 자막 달아주기

openai는 자극적인 프롬프트 금지

from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()

client = OpenAI()

prompt = """
보르조이 10마리
"""
img = client.images.generate(
    model='dall-e-3',
    prompt=prompt,
    size='1024x1024',
    quality='standard',
    n=1,
    response_format='b64_json')

import base64
img_data=base64.b64decode(img.data[0].b64_json)

from PIL import Image
import io
Image.open(io.BytesIO(img_data))

책 훑기

• 파라미터 크기 클수록 초반에는 모델의 성능이 오른다는 논문이 존재한다.
• RLHF
• 정렬 : open ai가 사용자의 답변에 대해 제재하는 것과 연관되어있음.

lora

• 이미지, 로라….

https://velog.io/@lse0912/RunPod-%EC%84%9C%EB%B2%84-%EB%8C%80%EC%97%AC-%EB%B0%8F-VS-CODE-%EC%97%B0%EB%8F%99

neo4j

https://bosungtea9416.tistory.com/entry/VScode%EB%A1%9C-%EC%84%9C%EB%B2%84%EC%97%90-SSH-%EC%9B%90%EA%B2%A9-%EC%A0%91%EC%86%8D%ED%95%98%EA%B8%B0

• RLHF : 보상해킹 - 강화학습 과정에서 평가 점수만 높게 받으려는

⇒ ppo로 해결할 수 있음.

강화학습 사용하지 않는 학습법도 있음

sllm에서 많이 사용된다.

1. 기각 샘플링

Lora

PEFT : Parameter- Efficient-Fine-tuning

• 원래 가중치에 그냥 위에 더해주는 것이기 때문에, 학습 효과를 보일 수 있다.

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

!pip install transformers
!pip install trl

import torch
import transformers
import trl

print(f"PyTorch version: {torch.__version__}")
print(f"Transformers version: {transformers.__version__}")
print(f"TRL version: {trl.__version__}")
print(f"CUDA available: {torch.cuda.is_available()}")
if torch.cuda.is_available():
    print(f"CUDA version: {torch.version.cuda}")

hug_api ="내가만든토큰"

from huggingface_hub import login
login(token=hug_api)

model_id = "google/gemma-2-9b-it" 

from transformers import (AutoTokenizer, 
                          AutoModelForCausalLM, 
                          TrainingArguments, 
                          BitsAndBytesConfig, 
                          pipeline, 
                          StoppingCriteria)
model = AutoModelForCausalLM.from_pretrained(
    model_id,
    device_map="auto",
    torch_dtype=torch.bfloat16,
    attn_implementation='eager')
    

터미널 용량 확인

watch -n 1 nvidia-smi

cd .cache
du -h --max-depth=1

지도 미세 조정 trl 패키지로 진행 & 추가 패키지 설치

from trl import (     
                 SFTTrainer,
                 SFTConfig)

tokenizer = AutoTokenizer.from_pretrained(model_id)
tokenizer.pad_token = tokenizer.eos_token
tokenizer.padding_side = 'right'

• 지도 미세 조정(Supervised Fine-tuning)

터미널

wget https://raw.githubusercontent.com/MrBananaHuman/CounselGPT/main/total_kor_multiturn_counsel_bot.jsonl

ls -alh

root@7ff7d93fc021:/workspace# head total_kor_multiturn_counsel_bot.jsonl 

root@7ff7d93fc021:/workspace# tail -50 total_kor_multiturn_counsel_bot.jsonl 

import json
with open('./total_kor_multiturn_counsel_bot.jsonl',  'r', 
          encoding='utf-8') as file:
    original_jsonl_data = [json.loads(line) for line in file]

original_jsonl_data[0]

converted_messages = []
for m in original_jsonl_data:
    m
    break
[{'role' : x['speaker'], 'content' : x['utterance']} for x in m]

cat total_kor_multiturn_counsel_bot.jsonl
cat total_kor_multiturn_counsel_bot.jsonl | grep user

• grep :있는 거 잡아내기

tmp = [{'role' : x['speaker'], 'content' : x['utterance']} for x in m]
tmp[0]['role']

'상담사’

최종 전처리 수정

import json
with open('./total_kor_multiturn_counsel_bot.jsonl',  'r', 
          encoding='utf-8') as file:
    original_jsonl_data = [json.loads(line) for line in file]

original_jsonl_data[0]

speaker_dict = {'내담자': 'user', '상담사': 'assistant'}

def merge_consecutive_messages(messages):
    if not messages:
        return []
    
    merged = []
    current_role = messages[0]['role']
    current_content = messages[0]['content']
    
    for message in messages[1:]:
        if message['role'] == current_role:
            current_content += " " + message['content']
        else:
            merged.append({'role': current_role, 'content': current_content})
            current_role = message['role']
            current_content = message['content']
    
    merged.append({'role': current_role, 'content': current_content})
    return merged
    
def preprocess_conversation(messages):
    converted_messages = [{'role': speaker_dict[m['speaker']], 'content': m['utterance']} for m in messages]
    converted_messages = merge_consecutive_messages(converted_messages)
    if not converted_messages or len(converted_messages) % 2 != 0:
        return []
    return converted_messages
def transform_to_new_format(original_data):
    transformed_data = []
    for conversation in original_data:
        processed_conversation = preprocess_conversation(conversation)
        if processed_conversation:
            transformed_data.append(processed_conversation)
    return transformed_data

result = transform_to_new_format(original_jsonl_data)

저장해두기

with open("./train_dataset.jsonl", "w", encoding="utf-8") as file:
    for conversation in result:
        json_obj = {"messages": conversation}
        json.dump(json_obj, file, ensure_ascii=False)
        file.write("\n") 

학습용 데이터로 불러오기

from datasets import Dataset, load_dataset

dataset = load_dataset("json", data_files="./train_dataset.jsonl",split='train')

dataset['messages'][1]

from peft import AutoPeftModelForCausalLM, LoraConfig, PeftConfig 

peft_config = LoraConfig(
        lora_alpha=128,
        lora_dropout=0.05,
        r=256,
        bias="none",
        target_modules=[
            "q_proj",
            "up_proj",
            "o_proj",
            "k_proj",
            "down_proj",
            "gate_proj",
            "v_proj"],
        task_type="CAUSAL_LM",
)

args = SFTConfig(
    output_dir="./model_output", 
    num_train_epochs=3,          
    per_device_train_batch_size=2,
    gradient_accumulation_steps=4,
    gradient_checkpointing=True,  
    optim="adamw_torch_fused",    
    logging_steps=100,            
    save_strategy="epoch",        
    learning_rate=2e-4,           
    bf16=True,                    
    tf32=True,                    
    max_grad_norm=0.3,            
    warmup_ratio=0.03,            
    lr_scheduler_type="constant", 
    packing=True,
)

!pip 에서 ! : 그 운영체제에 맞게 판단함

정상작동

import torch
import transformers
import trl

print(f"PyTorch version: {torch.__version__}")
print(f"Transformers version: {transformers.__version__}")
print(f"TRL version: {trl.__version__}")
print(f"CUDA available: {torch.cuda.is_available()}")
if torch.cuda.is_available():
    print(f"CUDA version: {torch.version.cuda}")

hug_api = "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"

from huggingface_hub import login
login(token=hug_api)

model_id = "google/gemma-2-9b-it" 

from transformers import (AutoTokenizer, 
                          AutoModelForCausalLM, 
                          TrainingArguments, 
                          BitsAndBytesConfig, 
                          pipeline, 
                          StoppingCriteria)
model = AutoModelForCausalLM.from_pretrained(
    model_id,
    device_map="cuda",
    torch_dtype=torch.bfloat16,
    attn_implementation='eager')
    

from trl import (SFTTrainer,
                 SFTConfig)

tokenizer = AutoTokenizer.from_pretrained(model_id)
tokenizer.pad_token = tokenizer.eos_token
tokenizer.padding_side = 'right'

import json
with open('./total_kor_multiturn_counsel_bot.jsonl',  'r', 
          encoding='utf-8') as file:
    original_jsonl_data = [json.loads(line) for line in file]

speaker_dict = {'내담자': 'user', '상담사': 'assistant'}

def merge_consecutive_messages(messages):
    if not messages:
        return []
    
    merged = []
    current_role = messages[0]['role']
    current_content = messages[0]['content']
    
    for message in messages[1:]:
        if message['role'] == current_role:
            current_content += " " + message['content']
        else:
            merged.append({'role': current_role, 'content': current_content})
            current_role = message['role']
            current_content = message['content']
    
    merged.append({'role': current_role, 'content': current_content})
    return merged
    
def preprocess_conversation(messages):
    converted_messages = [{'role': speaker_dict[m['speaker']], 'content': m['utterance']} for m in messages]
    converted_messages = merge_consecutive_messages(converted_messages)
    if converted_messages and converted_messages[0]['role'] == 'assistant':
        converted_messages = converted_messages[1:]
    while converted_messages and converted_messages[-1]['role'] == 'user':
        converted_messages = converted_messages[:-1]
        
    if not converted_messages or len(converted_messages) % 2 != 0:
        return []
    return converted_messages
def transform_to_new_format(original_data):
    transformed_data = []
    for conversation in original_data:
        processed_conversation = preprocess_conversation(conversation)
        if processed_conversation:
            transformed_data.append(processed_conversation)
    return transformed_data

result = transform_to_new_format(original_jsonl_data)

with open("./train_dataset.jsonl", "w", encoding="utf-8") as file:
    for conversation in result:
        json_obj = {"messages": conversation}
        json.dump(json_obj, file, ensure_ascii=False)
        file.write("\n") 

from datasets import Dataset, load_dataset

dataset = load_dataset("json", data_files="./train_dataset.jsonl", split='train')

from peft import AutoPeftModelForCausalLM, LoraConfig, PeftConfig 

peft_config = LoraConfig(
        lora_alpha=128,
        lora_dropout=0.05,
        r=128,
        bias="none",
        target_modules=[
            "q_proj",
            "o_proj",
            "k_proj",
            "v_proj"],
        task_type="CAUSAL_LM",
)

args = SFTConfig(
    output_dir="./model_output", 
    num_train_epochs=3,          
    per_device_train_batch_size=1,
    gradient_accumulation_steps=8,
    gradient_checkpointing=True,  
    optim="adamw_torch_fused",    
    # optim="paged_adamw_8bit",
    logging_steps=100,            
    save_strategy="epoch",      
    learning_rate=2e-4,                               
    max_grad_norm=0.3,            
    warmup_steps=0.03,            
    lr_scheduler_type="constant", 
    fp16=True
)

from peft import get_peft_model, LoraConfig, TaskType

trainer = SFTTrainer(
    model=model,
    args=args,
    train_dataset=dataset,
    peft_config=peft_config,
    processing_class=tokenizer)

trainer.train()

백그라운드에서 돌릴 때는 방법이 있음

→ train.py를 nohup에서 실행시키기

nohup python train.py &

.

Day 5

https://jaeyung1001.tistory.com/entry/bf16-fp16-fp32의-차이점

오늘은 QLora를 돌려보자!

import torch
import torch.nn as nn
import os

class SimpleModel(nn.Module):
    def __init__(self):
        super(SimpleModel, self).__init__()
        self.fc1 = nn.Linear(10, 50) 
        self.relu = nn.ReLU()
        self.fc2 = nn.Linear(50, 5)

    def forward(self, x):
        x = self.fc1(x)
        x = self.relu(x)
        x = self.fc2(x)
        return x
        
model_fp32 = SimpleModel()

model_fp32.state_dict()['fc1.weight'][0][0].dtype

torch.save(model_fp32.state_dict(), 'tmp.pth')
print(os.path.getsize("tmp.pth") ) #5557

• floating point : 32

int 8로 변환. 1바이트로 변환 중임

model_int8 = torch.quantization.quantize_dynamic(
    model_fp32, # 대상 
    {nn.Linear},
    dtype=torch.qint8
)

torch.save(model_int8.state_dict(), 'tmp2.pth')
print(os.path.getsize("tmp2.pth") ) #4643

5557 → 4643으로 모델의 크기가 변형된 것을 확인할 수 있음

https://docs.pytorch.org/docs/stable/quantization.html

양자화 전용 라이브러리

pip install bitsandbytes==0.41.3
pip install accelerate==0.24.1

ssh root@213.173.110.174 -p 13779 -i ~/.ssh/id_ed25519

scp -P 13779 -i ~/.ssh/id_ed25519 final_mode.tar root@213.173.110.174:/workspace/

tar xvf final_mode.tar

mv ./workspace/final_model/ ./

• workspace 안 폴더 상위폴더로 꺼내기.

vim requirements.txt
trl==0.29.1
torch==2.8.0+cu128
torchaudio==2.8.0+cu128
torchvision==0.23.0+cu128
nvidia-cublas-cu12==12.8.4.1
nvidia-cuda-cupti-cu12==12.8.90
nvidia-cuda-nvrtc-cu12==12.8.93
nvidia-cuda-runtime-cu12==12.8.90
nvidia-cudnn-cu12==9.10.2.21
nvidia-cufft-cu12==11.3.3.83
nvidia-cufile-cu12==1.13.1.3
nvidia-curand-cu12==10.3.9.90
nvidia-cusolver-cu12==11.7.3.90
nvidia-cusparse-cu12==12.5.8.93
nvidia-cusparselt-cu12==0.7.1
nvidia-nccl-cu12==2.27.3
nvidia-nvjitlink-cu12==12.8.93
nvidia-nvtx-cu12==12.8.90
transformers==5.3.0
peft==0.18.1
bitsandbytes==0.49.2

pip install -r requirements.txt 

from peft import get_peft_model, LoraConfig, TaskType
import torch
import transformers
import trl
from transformers import (AutoTokenizer, 
                          AutoModelForCausalLM, 
                          TrainingArguments, 
                          BitsAndBytesConfig, 
                          pipeline, 
                          StoppingCriteria)
from huggingface_hub import login
from trl import (SFTTrainer,
                 SFTConfig)
import json
from datasets import Dataset, load_dataset
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
from peft import PeftModel

print(f"PyTorch version: {torch.__version__}")
print(f"Transformers version: {transformers.__version__}")
print(f"TRL version: {trl.__version__}")
print(f"CUDA available: {torch.cuda.is_available()}")
if torch.cuda.is_available():
    print(f"CUDA version: {torch.version.cuda}")

hug_api = "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"

login(token=hug_api)

model_id = "google/gemma-2-9b-it" 

    

bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True
)

model = AutoModelForCausalLM.from_pretrained(
    model_id,
    quantization_config=bnb_config,
    device_map="auto")

model = PeftModel.from_pretrained(model, "./final_model")

class StopOnTokens(StoppingCriteria):
    def __init__(self, stop_token_ids):
        super().__init__()
        self.stop_token_ids = stop_token_ids

    def __call__(self, input_ids: torch.LongTensor, scores: torch.FloatTensor, **kwargs) -> bool:
        for stop_id in self.stop_token_ids:
            if input_ids[0][-1] == stop_id:
                return True
        return False

tokenizer = AutoTokenizer.from_pretrained( "./final_model")
user_token_id = tokenizer.encode("user", add_special_tokens=False)[0]

input_text = "요즘 겁나 힘드네"
input_ids = tokenizer.encode(input_text, return_tensor='pt')
input_ids = torch.tensor(input_ids).to('cuda')
input_ids  = input_ids.unsqueeze(0)

output = model.generate(
    input_ids, 
    max_new_tokens=400,
    do_sample=True,
    temperature=0.7,
    pad_token_id=tokenizer.eos_token_id
)

print(tokenizer.decode(output[0], skip_special_tokens=True))

def chat(input_text):
    input_ids = tokenizer.encode(input_text, return_tensor='pt')
    input_ids = torch.tensor(input_ids).to('cuda')
    input_ids = input_ids.unsqueeze(0)
    output = model.generate(
    input_ids, 
    max_new_tokens=400,
    do_sample=True,
    temperature=0.7,
    pad_token_id=tokenizer.eos_token_id
    )
    print(tokenizer.decode(output[0], skip_special_tokens=True))

def chat(input_text):
    input_ids = tokenizer.encode(input_text, return_tensors='pt').to('cuda')
    output = model.generate(
    input_ids, 
    max_new_tokens=400,
    do_sample=True,
    temperature=0.7,
    pad_token_id=tokenizer.eos_token_id
    )
    return tokenizer.decode(output[0], skip_special_tokens=True)

rt = chat("네. 요즘 저는 다른 사람들과 비교하게 되면, 왜 내가 저렇게 못하거나 따라갈 수 없는지 고민하게 됩니다.")

model_original = AutoModelForCausalLM.from_pretrained(
    model_id,
    quantization_config=bnb_config,
    device_map="auto")

from transformers import pipeline
pipe = pipeline(
    "text-generation",
    model=model_original,
    tokenizer=tokenizer,
)

messages = [
    {"role": "user", "content":  "요즘 겁나 힘드네"},
]

prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)

outputs = pipe(
    prompt,
    max_new_tokens=256,
    do_sample=True,
    temperature=0.7,
    top_k=50,
    top_p=0.95
)

print(outputs[0]['generated_text'])

빌런 대화로 학습시켜보자

import torch
from datasets import load_dataset
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from trl import SFTTrainer, SFTConfig

model_id = "Qwen/Qwen2.5-7B-Instruct"
dataset = load_dataset("json", data_files="fantasy_villain_dataset.jsonl", split="train")

tokenizer = AutoTokenizer.from_pretrained(model_id)
if tokenizer.pad_token is None:
    tokenizer.pad_token = tokenizer.eos_token
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
        bnb_4bit_use_double_quant=True
)

model = AutoModelForCausalLM.from_pretrained(
    model_id,
    quantization_config=bnb_config,
    device_map="auto"
)

메모리 안정화

model = prepare_model_for_kbit_training(model)

Lora 설정

lora_config = LoraConfig(
    r=128,
    lora_alpha=32,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"],
     lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM")

SFT 설정

sft_config = SFTConfig(
    output_dir="./qwen2.5_villain_output",
    per_device_train_batch_size=2,  
    gradient_accumulation_steps=4,  
    gradient_checkpointing=True, 
    optim="paged_adamw_8bit",
    learning_rate=2e-4,             
    num_train_epochs=3,  
    logging_steps=10   )

trainer = SFTTrainer(
    model=model,
    train_dataset=dataset,
    peft_config=lora_config,
    args=sft_config,
    processing_class=tokenizer)

trainer.train()

llama.cpp

올라마 내장 기본 기능이 cpp.

허깅페이스에서 받은 모델을 올라마에 얹어서 설계한다.

런팟 터미널에서 실행~

curl -LsSf https://hf.co/cli/install.sh | bash

hf auth login 

hf download LGAI-EXAONE/EXAONE-4.0-1.2B --local-dir /workspace/lg

git clone https://github.com/ggerganov/llama.cpp.git

허깅페이스에서 받은 모델을 gguf 형태로 변환하기.

• ollama에 올리기 위함

cd llama.cpp

pip install -r requirements.txt

 python convert_hf_to_gguf.py /workspace/lg --outfile /workspace/exaone

curl -fsSL https://ollama.com/install.sh | sh

ollama

vim Modelfile 

FROM /workspace/exaone.gguf

TEMPLATE """{{ if .System }}<|system|>
{{ .System }}<|thought|>
{{ end }}<|user|>
{{ .Prompt }}<|assistant|>
"""

PARAMETER stop "<|user|>"
PARAMETER stop "<|assistant|>"
PARAMETER stop "<|system|>"
PARAMETER stop "<|thought|>"
PARAMETER stop "</s>"

ollama create exaone-1.2b -f /workspace/Modelfile