14주차 필기록 통합본

김다피·2026년 4월 10일
SK 네트웍스 Family AI 캠프캠프 필기본

SKN-25 필기본

목록 보기
14/14
post-thumbnail

Day 1

런팟 킬게

!!!!!!!찌릿!!!!!!!!!

git clone https://github.com/onlybooks/llm.git

chapter 6장을 시작한다.

  • 라이브러리 버전은 출시 날짜보고 얼추 맞추기.
!pip install transformers==4.40.1 bitsandbytes==0.43.1 accelerate==0.29.3 datasets==2.19.0 tiktoken==0.6.0 huggingface_hub==0.22.2 autotrain-advanced==0.7.77 -qqq

!pip uninstall -y autotrain-advanced
!pip uninstall -y huggingface-hub
!pip install huggingface-hub==0.26.3

from huggingface_hub import login
login(token="여기 토큰 입력~")

모델 정보 확인

https://huggingface.co/spaces/hf-accelerate/model-memory-usage

갑자기 나에게 불쑥 다가온 wsl

Dataset → fine tuning → vllm 가져가보기

LLM Ops.. airflow로 구축해보기~!~

환경 변수에 airflow 등록하기


#wsl 
vim ~/.bashrc
ming9@DESKTOP-9B5C8HR:~$ vim ~/.bashrc
ming9@DESKTOP-9B5C8HR:~$ source ~/.bashrc
ming9@DESKTOP-9B5C8HR:~$ echo $AIRFLOW_HOME
**/home/ming9/airflow**

폴더 만들기

ming9@DESKTOP-9B5C8HR:~$ mkdir ~/airflow
ming9@DESKTOP-9B5C8HR:~$ ls
airflow  k8s  redis_data  workspace
ming9@DESKTOP-9B5C8HR:~$

airflow 회사에서는 2점대 많이 사용.

pip install apache-airflow --break-system-packages

실행파일 있는지 확인

ming9@DESKTOP-9B5C8HR:~$ cd ~/.local/bin
ming9@DESKTOP-9B5C8HR:~/.local/bin$ ls -al | grep airflow
-rwxr-xr-x 1 ming9 ming9  215 Mar 30 11:21 airflow
ming9@DESKTOP-9B5C8HR:~/.local/bin$

실행되는 지 확인

./airflow

bashrc 열어서 환경변수 등록

export PATH=$PATH:~/.local/bin

: 콜론 뒤에 붙이면 환경변수 연결이 된다. 

airflow standalone

이 뒤에 localhost:8080 접속

standalone으로 한 번 실행 시 설정파일이 쭉 생김

PS C:\Users\Ming9> wsl
ming9@DESKTOP-9B5C8HR:~$ cd ~/airflow
ming9@DESKTOP-9B5C8HR:~/airflow$ ll
total 5568
drwxr-xr-x  3 ming9 ming9    4096 Mar 30 11:31 ./
drwxr-x--- 13 ming9 ming9    4096 Mar 30 11:27 ../
-rw-------  1 ming9 ming9   81779 Mar 30 11:31 airflow.cfg
-rw-r--r--  1 ming9 ming9 1449984 Mar 30 11:33 airflow.db
-rw-r--r--  1 ming9 ming9   32768 Mar 30 11:34 airflow.db-shm
-rw-r--r--  1 ming9 ming9 4120032 Mar 30 11:34 airflow.db-wal
drwxr-xr-x  3 ming9 ming9    4096 Mar 30 11:31 logs/
-rw-r--r--  1 ming9 ming9      30 Mar 30 11:31 simple_auth_manager_passwords.json.generated
ming9@DESKTOP-9B5C8HR:~/airflow$ cat simple_auth_manager_passwords.json.generated
{"admin": "이거가 비번"}
ming9@DESKTOP-9B5C8HR:~/airflow$

→ admin 계정으로 로그인하기

  • Dag 기반 작동.

vim에서 소문자 gg → 맨 위

G → 맨 아래

리눅스에서 자동화 걸어보기 - crontab

crontab -e 
# 실행 결과와 에러를 모두 log.txt에 저장
* * * * * /home/ming9/script.sh >> /home/ming9/log.txt 2>&1
* * * * * date >> ~/cron_test.log

vscode 켜기

standalone으로 스타트하기 때문에 sqllite라는 db를 사용.

flask로 만들어져있고, was는 uvicorn

default_timezone = Asia/Seoul

변경하기.

dags 폴더 생성

# airflow 2.x 방식
from datetime import datetime, timezone
from airflow import DAG
from airflow.operators.python import PythonOperator

with DAG(
    dag_id = 'skn25_airflow_1',
    schedule="45 12 * * *",
    start_date=datetime(2026, 2, 10),
    catchup=False,
    tags=['skn25']
) as dag:

    def hello():
        print("hi")

    first_task  = PythonOperator(
        tack_id = 'hello',
        python_callable= hello
    )

    def hi():
        print("gggg")

    second_task  = PythonOperator(
        tack_id = 'hi',
        python_callable= hi
    )

    def goodgbye():
        print("goodbye")

    thrid_task =  PythonOperator(
        tack_id = 'goodbye',
        python_callable= goodbye
    )

vllm → paged attention

  • 자체적 추론을 시도할 수 있다.
  • 캐싱을 활용하여 속도가 빠르다고 볼 수 있다.

Day 2

모델 메모리에 대해서 잘 생각해봐야한다…

gradient checkpointing → 메모리 줄이는 방법

gradient accumulation

실습을 해봅시다

import httpx
from langchain_community.document_loaders import WebBaseLoader
from bs4 import BeautifulSoup
import os

url = "https://news.naver.com/main/ranking/popularDay.naver"

r = httpx.get(url)
bs = BeautifulSoup(r.text)
rt = bs.find_all("div", {'class' : 'list_content'})
len(rt)

from urllib import request
import re
p = re.compile("[0-9a-zA-Z가-힣]+")
from tqdm import tqdm
target_folder= "./img_src"
os.makedirs(target_folder, exist_ok=True)
for x in tqdm(rt):
    target_url = x.a['href']
    r2 = httpx.get(target_url)
    bs2 = BeautifulSoup(r2.text)
    tmp = bs2.find('article', id="dic_area").find_all("img")
    for x in tmp:
        filename = " ".join(p.findall(x['alt']))
        request.urlretrieve(x['data-src'], f"./{target_folder}/{filename}.jpg")

from chromadb.utils.embedding_functions import OpenCLIPEmbeddingFunction
from chromadb.utils.data_loaders import ImageLoader
from dotenv import load_dotenv
import chromadb

chroma_client = chromadb.PersistentClient(path="./img_vecdb3")
image_loader = ImageLoader()
CLIP = OpenCLIPEmbeddingFunction()
imgae_vdb =chroma_client.get_or_create_collection(
    name='news_image',
    embedding_function=CLIP,
    data_loader=image_loader
)

target_folder
ids = []
uris = []
for f in os.listdir(target_folder):
    # print(f)
   
    f_name = f.split(".")[0]
    if len(f_name) < 2:
        continue
    ids.append(f_name)
    uris.append(f"{target_folder}/{f}")
   

imgae_vdb.add(
    ids=ids,
    uris=uris
)

rt = imgae_vdb.query(
    query_texts=['photo of the stock indicators'],
    n_results=3,
    include=['uris', 'distances']
)

wsl로 airflow 구동하고, 비밀번호 켜서 확인하기.

맨 하단 json 에 admin 아이디의 비밀번호 있음

wsl 환경에서 버전 맞추기 위해 라이브러리 추가 설치

pip install lxml --break-system-packages
pip install bs4 --break-system-packages
pip install chromadb --break-system-packages

wsl 안에 airflow 폴더 안에 clip_news.py 설계

from airflow.decorators import dag, task
from pendulum import datetime
import httpx
from bs4 import BeautifulSoup
from urllib import request
import re
import os
from tqdm  import tqdm

@dag(dag_id="naver_news_image",
    start_date=datetime(2026,2,1),
    schedule= "45 14 * * *",
    catchup=False,
    tags=['naver', 'images', 'news']
)
def clip_news_pipeline():

    @task
    def get_news_img():
        p = re.compile("[0-9a-zA-Z가-힣]+")
        url = "https://news.naver.com/main/ranking/popularDay.naver"
        rt = BeautifulSoup(httpx.get(url).text).find_all("div", {'class' : 'list_content'})

        target_folder= "./img_src"
        os.makedirs(target_folder, exist_ok=True)
        for x in tqdm(rt):
            target_url = x.a['href']
            r2 = httpx.get(target_url)
            bs2 = BeautifulSoup(r2.text)
            tmp = bs2.find('article', id="dic_area").find_all("img")
            for x in tmp:
                filename = " ".join(p.findall(x['alt']))[:20]
                request.urlretrieve(x['data-src'], f"./{target_folder}/{filename}.jpg")

    get_news_img()

clip_news_pipeline()

  • 트리거 실행 시 자동으로 코드가 돌아간다

짜잔~

기술 하나 더 추가~

get_news_img()밑에 추가됨

chromaDB관련 코드 추가


from airflow.decorators import dag, task
from pendulum import datetime
import httpx
from bs4 import BeautifulSoup
from urllib import request
import re
import os
from tqdm  import tqdm
from chromadb.utils.embedding_functions import OpenCLIPEmbeddingFunction
from chromadb.utils.data_loaders import ImageLoader
import chromadb
from huggingface_hub import login

@dag(dag_id="naver_news_image",
    start_date=datetime(2026,2,1),
    schedule= "45 6 * * *",
    catchup=False,
    tags=['naver', 'images', 'news']
)
def clip_news_pipeline():

    @task
    def get_news_img():
        p = re.compile("[0-9a-zA-Z가-힣]+")
        url = "https://news.naver.com/main/ranking/popularDay.naver"
        rt = BeautifulSoup(httpx.get(url).text).find_all("div", {'class' : 'list_content'})

        target_folder= "./img_src"
        os.makedirs(target_folder, exist_ok=True)
        for x in tqdm(rt):
            target_url = x.a['href']
            r2 = httpx.get(target_url)
            bs2 = BeautifulSoup(r2.text)
            tmp = bs2.find('article', id="dic_area").find_all("img")
            for x in tmp:
                filename = " ".join(p.findall(x['alt']))[:20]
                request.urlretrieve(x['data-src'], f"./{target_folder}/{filename}.jpg")

    @task
    def store_vector_images():
        hf_token = "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"

        login(token=hf_token)

        chroma_client = chromadb.PersistentClient(path="./img_vecdb3")
        image_loader = ImageLoader()
        CLIP = OpenCLIPEmbeddingFunction()
        imgae_vdb =chroma_client.get_or_create_collection(
            name='news_image',
            embedding_function=CLIP,
            data_loader=image_loader
        )

        target_folder =  "./img_src"
        ids = []
        uris = []
        for f in os.listdir(target_folder):
            # print(f)
       
            f_name = f.split(".")[0]
            if len(f_name) < 2:
                continue
            ids.append(f_name)
            uris.append(f"{target_folder}/{f}")
       

        imgae_vdb.add(
            ids=ids,
            uris=uris
        )
       

    get_news_img()
    store_vector_images()

clip_news_pipeline()

근데? 여기서 멈추지 않고 병렬처리되지 않게 각 task에 순서를 주입해본다.


    first_task = get_news_img()
    second_task = store_vector_images()

    first_task >> second_task

clip_news_pipeline()

이렇게 되면 꾸준히 데이터가 수집되어서 벡터db에 쌓이게 되고, 질문을 넣으면 돌아가게 됨.

→ 이미지 기반 검색 챗봇

연결~

Day 3

오늘의 목표 : 이미지 기준으로 검색해서 기사 찾기

API를 설정해봅시다…

wsl에서 환경 세팅

-가상환경 폴더 만들기 
uv init vdb_serv

- 폴더 이동 
cd vdb_serv

- 가상환경 생성  
uv venv

- 아래 requirements.txt에 패키지 버전 넣기 
open_clip_torch==3.3.0
chromadb==1.5.5
fastapi
uvicorn
pydantic

- python 환경 변경 
source .venv/bin/activate

- 패키지 설치 
uv pip install -r requirements.txt

test.py

from fastapi import FastAPI
from pydantic import BaseModel

# FastAPI 인스턴스 생성
app = FastAPI()

# POST 요청으로 받을 데이터 구조
class UserQuery(BaseModel):
    query : str

@app.post("/query/")
async def create_user(query : UserQuery):
    query_dict = query.model_dump()
    query_dict.update({"status" : "성공"})
    return query_dict
- 터미널(환경 로그인 된 상태에서)
uvicorn test:app --reload

- 브라우저
http://127.0.0.1:8000/docs

POST 버튼 클릭하여 접속

‘잘 되는구리.

이제 chroma DB설계한거 넣기

  • 어제 벡터 db만든거 가지고 서비스 올리려고 설계한 것임.
  • 윈도우에서 streamlit 뜨는데, db는 리눅스 컴퓨터에 있으니 이 2개를 fastapi로 연결하고자 한다.
from fastapi import FastAPI
from pydantic import BaseModel
from chromadb.utils.embedding_functions import OpenCLIPEmbeddingFunction
from chromadb.utils.data_loaders import ImageLoader
import chromadb
from huggingface_hub import login

hf_token = "hf_z"

login(token=hf_token)
chroma_client = chromadb.PersistentClient(path="/home/playdata2/img_vecdb3")
image_loader = ImageLoader()
CLIP = OpenCLIPEmbeddingFunction()
imgae_vdb =chroma_client.get_or_create_collection(
    name='news_image',
    embedding_function=CLIP,
    data_loader=image_loader
)

# FastAPI 인스턴스 생성
app = FastAPI()

# POST 요청으로 받을 데이터 구조
class UserQuery(BaseModel):
    query : str

@app.post("/query/")
async def create_user(query : UserQuery):
    query_dict = query.model_dump()
    # query_dict.update({"status" : "성공"})
    # return query_dict

    print(query_dict['query'])
   
    rt = imgae_vdb.query(
            query_texts=[query_dict['query']],
            n_results=3,
            include=['uris', 'distances']
        )
    print(rt)
    query_dict.update({"return" : f"{rt}"})
    return rt

clip은 query를 영어로만 인식한다!

uv pip install celery
uv pip install redis

기능을 ~ 추가하자~

celery 기능 켜는 터미널 1

redis 환경 띄우기

history | grep redis
- 컨테이너 목록 에 있는 분들 
- 
docker ps -a 

docker start redis

task.py

  • celey 상견례….?
from celery import Celery  
import time  

app = Celery(
    'tasks',
    broker='redis://:123@localhost:6379/0',
    backend='redis://:123@localhost:6379/0'
)

@app.task
def add(x , y):
    time.sleep(3)
    return x + y

터미널에서 실행

celery -A task worker --loglevel=info

  • docker start redis를 했기 때문에, task를 실행했을 때 접속이 가능해짐.

celery_test.ipynb


from task import add
result = add.delay(10,5)
print(f"현재 작업 요청 상태 : {result.status}")
print(f"계산 결과 : {result.get()}")
  • 비동기이기 때문에 서버는 계속 켜져있으나, 요청할 때 값이 나옴

현재 작업 요청 상태 : PENDING
계산 결과 : 15

소프트웨어로 데이터 넘기는 거 확인하기~~!~!

  • celery-task-meta 있는지 확인

접속 정보 : broker로 접속하기~

연결도 했겠다…

import random
for x in range(100):
    result = add.delay(random.randint(1,100),random.randint(1,100))

랜덤으로 값을 큐로 삽입해본다.

celery를 통해 메시지 큐 & fast api 설정한다…

serving.py

  • fast api 설정하는 코드.
from fastapi import FastAPI
from pydantic import BaseModel
from chromadb.utils.embedding_functions import OpenCLIPEmbeddingFunction
from chromadb.utils.data_loaders import ImageLoader
import chromadb
from huggingface_hub import login
from task import get_query

hf_token = ""

login(token=hf_token)
chroma_client = chromadb.PersistentClient(path="/home/playdata2/img_vecdb3")
image_loader = ImageLoader()
CLIP = OpenCLIPEmbeddingFunction()
imgae_vdb =chroma_client.get_or_create_collection(
    name='news_image',
    embedding_function=CLIP,
    data_loader=image_loader
)

# FastAPI 인스턴스 생성
app = FastAPI()

# POST 요청으로 받을 데이터 구조
class UserQuery(BaseModel):
    query : str

@app.post("/query/")
async def create_user(query : UserQuery):
    query_dict = query.model_dump()
    # query_dict.update({"status" : "성공"})
    # return query_dict

    # print(query_dict['query'])
    task = get_query.delay(query_dict['query'])

    return {
        'task_id' : task.id, 'messages' : '메시지를 전달했습니다.'
    }

task.py

from celery import Celery  
import time  
from chromadb.utils.embedding_functions import OpenCLIPEmbeddingFunction
from chromadb.utils.data_loaders import ImageLoader
import chromadb

from huggingface_hub import login

hf_token = ""

login(token=hf_token)
chroma_client = chromadb.PersistentClient(path="/home/playdata2/img_vecdb3")
image_loader = ImageLoader()
CLIP = OpenCLIPEmbeddingFunction()
imgae_vdb =chroma_client.get_or_create_collection(
    name='news_image',
    embedding_function=CLIP,
    data_loader=image_loader
)

app = Celery(
    'tasks',
    broker='redis://:123@localhost:6379/0',
    backend='redis://:123@localhost:6379/0'
)

@app.task
def add(x , y):
    time.sleep(3)
    return x + y

@app.task
def get_query(query : str):
    rt = imgae_vdb.query(
    query_texts=[query],
        n_results=3,
        include=['uris', 'distances'])
   
    return rt
  • 이후 celery와 fastapi 각각의 터미널에 실행
  • fastapi 가 celery에게 데이터를 넘김.
uvicorn serving:app --reload
celery -A task worker --loglevel=info

→ 이 코드는 데드락이 발생함.

진짜 찐최종 코드

task.py - 교착을 방지하기 위해 logging을 추가한다.


from celery import Celery  
import time  
from chromadb.utils.embedding_functions import OpenCLIPEmbeddingFunction
from chromadb.utils.data_loaders import ImageLoader
import chromadb
from celery.utils.log import get_task_logger

from huggingface_hub import login

hf_token = ""

login(token=hf_token)
chroma_client = None
CLIP = None
image_loader = None
imgae_vdb = None

app = Celery(
    'tasks',
    broker='redis://:123@localhost:6379/1',
    backend='redis://:123@localhost:6379/1'
)

logger = get_task_logger(__name__)
logger.info(f"==== [1] 시작' ====")

def init_db_and_model():
    global chroma_client, CLIP, image_loader, imgae_vdb
   
    if imgae_vdb is None:
        logger.info("==== 모델 및 데이터베이스 초기화 중 ====")
        hf_token = ""
        login(token=hf_token)
       
        chroma_client = chromadb.PersistentClient(path="img_vecdb3")
        image_loader = ImageLoader()
        CLIP = OpenCLIPEmbeddingFunction()
       
        imgae_vdb = chroma_client.get_or_create_collection(
            name='news_image',
            embedding_function=CLIP,
            data_loader=image_loader
        )
        logger.info("==== 초기화 완료 ====")

@app.task
def add(x , y):
    time.sleep(3)
    return x + y

@app.task
def get_query(query : str):
    logger.info(f"==== [1] get_query 태스크 시작: '{query}' ====")
    init_db_and_model()
    logger.info("==== [2] ChromaDB 쿼리 실행 중 ====")
    rt = imgae_vdb.query(
        query_texts=[query],
        n_results=3,
        include=['uris', 'distances']
    )
   
    logger.info("==== [3] ChromaDB 쿼리 완료 ====")
   
    safe_rt = {
        "ids": rt.get("ids", []),
        "distances": rt.get("distances", []),
        "uris": rt.get("uris", [])
    }
   
    return safe_rt

serving.py

from fastapi import FastAPI
from pydantic import BaseModel
from chromadb.utils.embedding_functions import OpenCLIPEmbeddingFunction
from chromadb.utils.data_loaders import ImageLoader
import chromadb
from huggingface_hub import login
from task import get_query

hf_token = ""

login(token=hf_token)
chroma_client = chromadb.PersistentClient(path="img_vecdb3")
image_loader = ImageLoader()
CLIP = OpenCLIPEmbeddingFunction()
imgae_vdb =chroma_client.get_or_create_collection(
    name='news_image',
    embedding_function=CLIP,
    data_loader=image_loader
)

# FastAPI 인스턴스 생성
app = FastAPI()

# POST 요청으로 받을 데이터 구조
class UserQuery(BaseModel):
    query : str

@app.post("/query/")
async def create_user(query : UserQuery):
    query_dict = query.model_dump()
    # query_dict.update({"status" : "성공"})
    # return query_dict

    # print(query_dict['query'])
    task = get_query.delay(query_dict['query'])

    return {
        'task_id' : task.id, 'messages' : '메시지를 전달했습니다.'
    }

명령어 보낼 시 잡힘

확인을 위해 serving.py에 get 방식을 추가해본다.

airflow 활용 데이터 수집→ 폴더 경로 바꿔서 스트림릿 연결

cips_news.py - airflow 폴더

from airflow.decorators import dag, task
from pendulum import datetime
import httpx
from bs4 import BeautifulSoup
from urllib import request
import re
import os
from tqdm  import tqdm
from chromadb.utils.embedding_functions import OpenCLIPEmbeddingFunction
from chromadb.utils.data_loaders import ImageLoader
import chromadb
from huggingface_hub import login
import pendulum
from datetime import date

local_tz = pendulum.timezone("Asia/Seoul")

@dag(dag_id="naver_news_image",
    start_date=datetime(2026,2,1, tz=local_tz),
    schedule= "13 14 * * *",
    catchup=False,
    tags=['naver', 'images', 'news']
)
def clip_news_pipeline():

    @task
    def get_news_img():
        p = re.compile("[0-9a-zA-Z가-힣]+")
        url = "https://news.naver.com/main/ranking/popularDay.naver"
        rt = BeautifulSoup(httpx.get(url).text).find_all("div", {'class' : 'list_content'})

        target_folder= f"./naver_news_img/{str(date.today()).replace("-", "")}/"
        os.makedirs(target_folder, exist_ok=True)
        for x in tqdm(rt):
            target_url = x.a['href']
            r2 = httpx.get(target_url)
            bs2 = BeautifulSoup(r2.text)
            tmp = bs2.find('article', id="dic_area").find_all("img")
            for x in tmp:
                filename = " ".join(p.findall(x['alt']))[:20]
                request.urlretrieve(x['data-src'], f"./{target_folder}/{filename}.jpg")
       
        return target_folder

    @task
    def store_vector_images(target_folder: str):
        hf_token = "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"
        login(token=hf_token)

        chroma_client = chromadb.PersistentClient(path="./naver_img_vector")
        image_loader = ImageLoader()
        CLIP = OpenCLIPEmbeddingFunction()
        imgae_vdb =chroma_client.get_or_create_collection(
            name='news_image',
            embedding_function=CLIP,
            data_loader=image_loader
        )

        ids = []
        uris = []
        for f in os.listdir(target_folder):
            # print(f)
       
            f_name = f.split(".")[0]
            if len(f_name) < 2:
                continue
            ids.append(f_name)
            uris.append(f"{target_folder}/{f}")
       

        imgae_vdb.add(
            ids=ids,
            uris=uris
        )
       

    first_task = get_news_img()
    second_task = store_vector_images(first_task)

    first_task >> second_task

clip_news_pipeline()

vectordb 전에 했던 코드 읽기.

chroma_client = chromadb.PersistentClient(path="/home/ming9/naver_img_vector")
image_loader = ImageLoader()
CLIP = OpenCLIPEmbeddingFunction()
  • 경로만 수정

경량화랑 재매개화 얘기하시네…

streamlit에 연결하기!

query넣으면 작동하는지 확인하는 1차 코드

img_chat.py


import streamlit as st
import requests

query  = st.text_input("이미지 기반 검색 시스템에 쿼리를 입력하세요")
post_url = "http://127.0.0.1:8000/query/"

if query:
    data = {
            "query": query
            }
    r = requests.post(post_url, json=data)

    st.markdown(r.text)

사진도 같이 나오게 기능 구현하기

import streamlit as st
import requests

query  = st.text_input("이미지 기반 검색 시스템에 쿼리를 입력하세요")
post_url = "http://127.0.0.1:8000/query/"
get_url = "http://127.0.0.1:8000/status/{}"
host = "http://127.0.0.1:8000/images/"
if query:
    data = {
            "query": query
            }
    r = requests.post(post_url, json=data)

    st.markdown(r.text)

    if r.status_code == 200:
        task_id = r.json()['task_id']

        status_placeholder = st.empty()

        while True:
            r2 = requests.get(get_url.format(task_id))

            if r2.status_code == 200:
                status_data = r2.json()
                status = status_data['status']
                if status == 'SUCCESS':
                    status_placeholder.success(f"{status_data['result']['uris']}")
                    for data in status_data['result']['uris'][0]:
                        st.markdown(
    f'''<a href="{host}{data.split("/naver_news_img/")[-1]}" target="_blank"> 사진보기 </a>''',
    unsafe_allow_html=True
)
                    break

결론

  1. 리눅스 기준 airflow에 dags 폴더 아래에 clips_news.py 실행해서 데이터 수집
  2. 수집한 데이터를 통해 시스템을 구현한다.
    1. vdb_serv 폴더 아래에 serving.py와 task.py를 둔다.
    2. 각각의 터미널에 실행해둔 상태로 둔다.
  3. img_chat.py를 실행해서 프론트엔드에서 실행한다.
profile
김다피
지루하게 선명하기보다는 흐릿해도 흥미롭게
이전 포스트

13주차 필기록 통합본

2개의 댓글

comment-user-thumbnail
ㅊH림
2026년 4월 13일

어떤 피크민이 이렇게 부지런할까

1개의 답글