LangGraph?

내가 LangGraph를 좋아하는 이유는 필요한 모듈을 온라인에서 가져와 내 마음대로 커스터마이징하기 편하기 때문이다. LangChain처럼 모든 기능이 일체형으로 묶여 있지 않고, Node 단위로 구성되어 있어 코드 수정과 확장이 훨씬 유연하다.

다른 사람들의 코드를 참고하며 많은 도움을 받은 만큼, 나도 누군가의 프로젝트에 작은 기여를 할 수 있으면 좋겠다는 생각이 들었다. 물론 코드 어시스턴트 예제는 이미 LangGraph 공식 문서에 나와 있지만, 한국어 자료는 부족하고 공식 문서는 내용이 다소 무겁게 느껴질 수 있다. 그래서 이 글에서는 처음 접하는 사람도 부담 없이 이해할 수 있도록 더 가볍고 실용적인 코드 쿡북을 만들어봤다. 특히 데이터 시각화 측면에 초점을 맞춰 코드를 짜봤다.

이 코드가 누군가의 프로젝트에 인용된다면 정말 뿌듯할 것 같다.
그럼 시작해보자! 🚀

🧩 LangGraph의 기본 요소

Agentic Code Generator를 만들기 전에, LangGraph의 핵심 개념부터 가볍게 훑어보자.

1. 상태(State): 작업 중 데이터를 저장하는 공간이야. 예를 들어 사용자 입력, 전처리 결과, 코드 실행 결과 같은 게 여기에 담긴다.
2. 노드(Node): 특정 작업을 처리하는 단위다. 데이터 로드, 전처리, 코드 생성, 시각화 등 단계별 작업이 여기서 이뤄진다.
3. 간선(Edge): 노드들끼리 연결해주는 다리 같은 존재다. 이걸로 작업 순서나 조건 분기를 설정할 수 있어.

이 구조 덕분에 작업 흐름을 시각적으로 파악하기도 쉽고, 코드 관리도 훨씬 편하다. 🙌

🛠️ Agentic Code Generator 플로우 설계

Agentic Code Generator는 아래 흐름으로 돌아간다.

1. 데이터 로드
2. 데이터 전처리
3. 사용자 요구사항 분석 및 코드 생성
4. 코드 실행 및 결과 확인
5. 에러 발생 시 코드 재생성 및 재실행

🔍 플로우 다이어그램

위의 그림은 Agentic Code Generator의 플로우 다이어그램이다. 최대한 간소화해서 만들어보았다.

flowchart LR
    A(START) --> B[데이터 로드]
    B --> C[데이터 전처리]
    C --> D[코드 생성]
    D --> E[시각화]
    E -->|에러 발생?| D
    E -->|에러 없음| F(END)

에러가 발생하면 자동으로 코드 생성 단계로 돌아가 반복 실행한다. 🔄

💻 코드 구현 과정

1️⃣ 상태(State) 정의

먼저 작업에 필요한 데이터를 저장할 공간을 만들어보자.

from typing import TypedDict

class State(TypedDict):
    input: str
    data: str
    processed_data: str
    code: str
    is_error: bool
    result: str

• input: 사용자의 요청 (input query)
• data: 로드된 데이터
• processed_data: 전처리된 데이터
• code: 생성된 코드
• is_error: 에러 발생 여부
• result: 실행 결과

각각의 쓰임새가 있는데, 아래에서 조금씩 설명해보겠다.

2️⃣ 노드(Node) 함수 정의

각 작업 단계별로 함수를 만들어보자.

📥 데이터 로드 노드

import pandas as pd

def load_data(state: State) -> State:
    data = pd.read_csv("data/example.csv")
    return {**state, "data": data}

데이터를 가져오는 간단한 코드다. 데이터 시각화를 위해서 CSV 파일을 가져온다. 참고로 해당 example.csv 파일은 Kaggle의 Titanic 생존자 예측 데이터를 사용했다.

🧹 데이터 전처리 노드

def preprocess_data(state: State) -> State:
    processed_data = state["data"].head(100)
    return {**state, "processed_data": processed_data}

데이터가 너무 많으면 처리 시간이 길어지니까 상위 100개만 추려보자.

🧠 코드 생성 노드

from langchain_openai import ChatOpenAI

def analyze_data(state: State) -> State:
    model = ChatOpenAI(model="gpt-4o-mini")
    prompt = f"{state['input']} 데이터: {state['processed_data']} 단, 오직 Code만 출력하세요."
    response = model.invoke(prompt)
    return {**state, "code": response.content}

모델에게 코드 생성을 요청하면, 사용자 입력과 전처리된 데이터를 기반으로 코드를 뚝딱 만들어준다.

만약 코드를 한번만 만든다면 위처럼 정의하면 되겠지만, 우리는 다이어그램에서 에러가 발생하면 다시 코드를 작성하기로 했으니까 코드를 조금 더 추가해야 한다.

def analyze_data(state: State) -> State:
    # 데이터 분석 로직 구현
    model = ChatOpenAI(model="gpt-4o-mini")
    if state["is_error"] is False:
        # 일반 코드 제작 프롬프트
        response = model.invoke(
            (
                f"{state['input']} "
                "단, 오직 Code만 출력하세요. "
                f"데이터: {state['processed_data']} "
            )
        )
    else:
        # Visualization 노드에서 코드 에러 발생시 다시 코드를 생성하는 프롬프트
        response = model.invoke(
            (
                f"{state['input']} "
                "단, 오직 Code만 출력하세요. "
                f"데이터: {state['processed_data']} "
                "====================== "
                f"[AI] {state['code']} "
                f"{state['result']} "
                "====================== "
                "실행 중 발생한 버그를 참고하여 다시 코드를 작성해주세요."
            )
        )
    code = response.content
    return State(code=code)

만약 처음으로 코드를 생성하는 상태라면 state["is_error"]가 False이므로 첫번째 if 문에서 "일반 코드 제작 프롬프트"로 LLM을 호출한다.

하지만 visualization 노드에서 코드를 실행했는데 error가 뜬 상황이라면 state["is_error"]가 True이므로 else 문에서 "다시 코드를 생성하는 프롬프트"로 LLM을 호출한다. 여기에는 에러 메시지와 다시 코드를 작성하라는 문구가 포함되어 있는 것을 볼 수 있다.

📊 시각화 노드

def visualization(state: State) -> State:
    import seaborn as sns
    import matplotlib.pyplot as plt
    from langchain_experimental.tools.python.tool import PythonAstREPLTool

    code = state["code"]
    repl_tool = PythonAstREPLTool(locals={"sns": sns, "plt": plt})

    try:
        exec_result = repl_tool.invoke(code)
        return {**state, "result": exec_result, "is_error": False}
    except Exception as e:
        return {**state, "result": str(e), "is_error": True}

AI가 생성한 코드를 실행해서 시각화를 출력한다. 실행 중 문제가 있으면 에러 메시지가 반환된다.

🚀 Tips

PythonAstREPLTool이 코드를 실행하는 Tool이다. PythonREPLTool과의 차이점은 보안이 조금 더 강하다는 점이다. PythonAstREPLTool은 코드 실행 전에 추상 구문 트리(Abstract Syntax Tree, AST) 분석을 통해 잠재적으로 위험한 코드나 금지된 연산을 사전에 차단할 수 있다. 이를 통해 외부 라이브러리 접근, 파일 시스템 조작, 무한 루프 실행 등 의도치 않은 동작을 방지하며 보다 안전한 코드 실행 환경을 제공한다. 따라서, 사용자 입력을 기반으로 코드를 실행해야 하는 시나리오에서 PythonAstREPLTool은 안정성과 보안을 모두 확보할 수 있는 적절한 선택지이다.

3️⃣ 플로우 구성하기

각 노드를 연결해 흐름을 완성하자.

from langgraph.graph import StateGraph, START, END

def build_flow():
    flow = StateGraph(State)

    flow.add_node("load_data", load_data)
    flow.add_node("preprocess_data", preprocess_data)
    flow.add_node("analyze_data", analyze_data)
    flow.add_node("visualization", visualization)

    flow.add_edge(START, "load_data")
    flow.add_edge("load_data", "preprocess_data")
    flow.add_edge("preprocess_data", "analyze_data")
    flow.add_edge("analyze_data", "visualization")

    flow.add_conditional_edges(
        "visualization",
        lambda state: "analyze_data" if state["is_error"] else "end",
        {"analyze_data": "analyze_data", "end": END},
    )

    return flow.compile()

에러가 발생하면 자동으로 코드 생성 단계로 돌아가 재실행할 수 있다. flow.add_conditional_edges에서 "visualization" 노드로부터 이어지는 conditional edges를 정한다. state["is_error"]가 True라면 "analyze_data" 노드로 이어지고, False라면 END 로 이어진다.

🚀 Tips

LangGraph의 State 객체는 직렬화를 거치기 때문에 pandas.DataFrame처럼 직렬화가 까다로운 객체는 직접 저장하기 어려울 수 있다. 특히 해당 Graph를 SubGraph로 추가할 때 직렬화와 관련된 Pydantic Error가 발생한다면 아래 솔루션을 시도해보자. ✅ 해결 방법:

• CSV, JSON 등 문자열 형태로 변환하여 저장:

  state["data"] = df.to_json()  # 저장 시
  df = pd.read_json(state["data"])  # 불러올 때

  이렇게 하면 상태 전환 시 데이터 손실을 방지하면서도 안정적인 처리가 가능하다.

4️⃣ 실행 및 결과 확인

if __name__ == "__main__":
    flow = build_flow()
    initial_state = State(
        input="Titanic 데이터를 시각적으로 멋지게 보여줘!",
        data="",
        processed_data="",
        code="",
        is_error=False,
        result=""
    )

    for event in flow.stream(initial_state):
        for step, output in event.items():
            print(f"\n🚀 STEP: {step}\n{output}\n")

✅ 실행 결과 예시

🚀 STEP별 진행 로그

🚀 STEP: load_data
데이터 로드 완료!

🚀 STEP: preprocess_data
상위 100개 데이터 전처리 완료!

🚀 STEP: analyze_data
코드 생성 성공! 📝

🚀 STEP: visualization
시각화 성공 🎉 (에러 없음)

📄 실제 State 데이터 (Dump)

📥 STEP: load_data

데이터 로드 후 State에 저장된 내용:

{
  'data': 
      PassengerId  Survived  Pclass                                               Name     Sex  ...  Fare Cabin  Embarked
0              1         0       3                            Braund, Mr. Owen Harris    male  ...  7.2500   NaN      S
1              2         1       1  Cumings, Mrs. John Bradley (Florence Briggs Th...  female  ... 71.2833   C85      C
2              3         1       3                             Heikkinen, Miss. Laina  female  ...  7.9250   NaN      S
...          ...       ...     ...                                                ...     ...  ...     ...   ...    ...
890          891         0       3                                Dooley, Mr. Patrick    male  ...  7.7500   NaN      Q

[891 rows x 12 columns]
}

🧹 STEP: preprocess_data

상위 100개 데이터 전처리 후 State에 저장된 내용:

{
  'processed_data': 
      PassengerId  Survived  Pclass                                               Name  ...     Fare  Cabin Embarked
0              1         0       3                            Braund, Mr. Owen Harris  ...    7.2500   NaN       S
1              2         1       1  Cumings, Mrs. John Bradley (Florence Briggs Th...)  ...   71.2833    C85      C
2              3         1       3                             Heikkinen, Miss. Laina  ...    7.9250   NaN       S
...          ...       ...     ...                                                ...  ...      ...    ...     ...
99           100         0       2                                  Kantor, Mr. Sinai  ...   26.0000    NaN       S

[100 rows x 12 columns]
}

📊 STEP: analyze_data

분석 코드 생성 후 State에 저장된 내용:

{
  'code': 
"""
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns

# 데이터 프레임 생성
data = {
    'PassengerId': range(1, 101),
    'Survived': [0, 1, 1, 1, 0] * 20,
    'Pclass': [3, 1, 3, 1, 3] * 20,
    'Fare': [7.25, 71.2833, 7.925, 53.1, 8.05] * 20,
    'Embarked': ['S', 'C', 'S', 'S', 'S'] * 20
}
df = pd.DataFrame(data)

# 시각화
plt.figure(figsize=(12, 8))

# 1. 생존자 수
plt.subplot(2, 2, 1)
sns.countplot(data=df, x='Survived', palette='pastel')
plt.title('Survival Count')

# 2. 클래스별 생존율
plt.subplot(2, 2, 2)
sns.barplot(data=df, x='Pclass', y='Survived', palette='pastel')
plt.title('Survival Rate by Class')

# 3. 요금 분포
plt.subplot(2, 2, 3)
sns.histplot(data=df, x='Fare', kde=True, bins=30, color='purple')
plt.title('Fare Distribution')

# 4. 탑승 항구별 생존자 수
plt.subplot(2, 2, 4)
sns.countplot(data=df, x='Embarked', hue='Survived', palette='pastel')
plt.title('Survival Count by Embarked')

plt.tight_layout()
plt.show()
"""
}

🖼️ STEP: visualization

시각화 결과 및 실행 상태:

{
  'result': '',
  'is_error': False
}

✅ 📈 시각화된 Figure:

Titanic 데이터 시각화 결과:
🧍‍♂️ 생존자 수 비교 | 🏷️ 클래스별 생존율
💰 요금 분포 | 🚢 탑승 항구별 생존자 수

🌊 플로우 시각화

from IPython.display import display, Image
from main import build_flow

flow = build_flow()
display(Image(flow.get_graph(xray=False).draw_mermaid_png()))

• 각 단계(load_data → preprocess_data → analyze_data → visualization)에서 처리 결과가 State에 저장된다.
• 시각화 단계에서 에러가 발생하면 analyze_data 단계로 되돌아가 코드가 자동으로 수정 및 재실행된다.
• 정상적으로 실행될 경우 위와 같은 시각화 결과가 출력된다.

전체 코드

from typing import TypedDict
import pandas as pd

from langchain_openai import ChatOpenAI
from langchain_experimental.tools.python.tool import PythonAstREPLTool
from langgraph.graph import StateGraph, START, END


# State 정의
class State(TypedDict):
    input: str
    data: str
    processed_data: str
    code: str
    is_error: bool
    result: str


# 각 Node에 해당하는 함수 정의
def load_data(state: State) -> State:
    # 데이터 로드 로직 구현
    data = pd.read_csv("./data/example.csv")
    return State(data=data)


def preprocess_data(state: State) -> State:
    # 데이터 전처리 로직 구현
    data = state["data"]
    processed_data = data.head(100)
    return State(processed_data=processed_data)


def analyze_data(state: State) -> State:
    # 데이터 분석 로직 구현
    model = ChatOpenAI(model="gpt-4o-mini")
    if state["is_error"] is False:
        response = model.invoke(
            (
                f"{state['input']} "
                "단, 오직 Code만 출력하세요. "
                f"데이터: {state['processed_data']} "
            )
        )
    else:
        response = model.invoke(
            (
                f"{state['input']} "
                "단, 오직 Code만 출력하세요. "
                f"데이터: {state['processed_data']} "
                "====================== "
                f"[AI] {state['code']} "
                f"{state['result']} "
                "====================== "
                "실행 중 발생한 버그를 참고하여 다시 코드를 작성해주세요."
            )
        )
    code = response.content
    return State(code=code)


def visualization(state: State) -> State:
    # 코드 실행
    import seaborn as sns
    import matplotlib.pyplot as plt

    code = state["code"]
    processed_code = None
    if code.startswith("```"):
        processed_code = "\n".join(code.split("\n")[1:-1])
    else:
        processed_code = code

    python_repl_tool = PythonAstREPLTool(locals={"sns": sns, "plt": plt})
    try:
        exec_result = str(python_repl_tool.invoke(processed_code))
        return State(result=exec_result, is_error=False)
    except Exception as e:
        return State(result=e, is_error=True)


def build_flow():
    # flow init
    flow = StateGraph(State)

    # node 정의
    flow.add_node("load_data", load_data)
    flow.add_node("preprocess_data", preprocess_data)
    flow.add_node("analyze_data", analyze_data)
    flow.add_node("visualization", visualization)

    # edge 연결
    flow.add_edge(START, "load_data")
    flow.add_edge("load_data", "preprocess_data")
    flow.add_edge("preprocess_data", "analyze_data")
    flow.add_edge("analyze_data", "visualization")

    # codintional edge 연결
    def is_error_check(state: State) -> str:
        return "analyze_data" if state["is_error"] else "end"

    flow.add_conditional_edges(
        "visualization",
        is_error_check,
        {"analyze_data": "analyze_data", "end": END},
    )

    # compile
    return flow.compile()


if __name__ == "__main__":
    flow = build_flow()
    initial_state = State(
        input="다음 데이터를 전문적으로 시각화하여 하나의 Figure로 표현해주세요.",
        is_error=False,
        code="",
        reulst="",
    )
    for event in flow.stream(initial_state):
        for node_name, value in event.items():
            print(f"\n==============\nSTEP: {node_name}\n==============\n")
            print(value)

🔔 마무리

이번 글에서는 LangGraph를 활용한 데이터 처리 파이프라인 구축 방법과 단계별 시각화 과정을 다뤘다.
각 단계별 에러 처리 흐름, 조건부 플로우 구성, 그리고 State 직렬화 팁 등을 포함해 실전 프로젝트에 적용할 수 있는 내용을 정리했다. 🚀

🔗 추가 자료 & 코드 확인하기

📂 전체 코드 GitHub에서 확인하기:
🔗 GitHub - Agentic Code Generator

📚 참고 자료:
🔗 LangGraph Code Assistant Tutorial

💬 읽어주셔서 감사합니다!
🙌 도움이 되셨다면 ⭐️ 스타를 눌러주시면 큰 힘이 됩니다.
✅ 추가로 다뤄보면 좋을 주제나 궁금한 내용이 있다면 댓글로 알려주세요. 관련 글을 작성해 보겠습니다. 😊

🎁🎁 케라스 창시자에게 배우는 딥러닝 개정2판 5~6장 을 참고하였습니다!

전체적인 머신러닝 워크플로를 정리해봤다. 책을 많이 참고했고 책보다 최대한 재미있게 써보려고 노력했다. 제목 어그로 죄송

맨날 따라치는 실습만 하다가 이 책에서 전체적인 개념과 관점을 배울 수 있었다. 대학 강의처럼 딱딱하지도 않고 훨씬 재밌다. 딥러닝에 관심있다면 위에 책을 한 번 꼭 읽어보길 바란다.

1. 문제가 뭐지?

흔히 머신러닝 워크플로에서 데이터 정제가 가장 중요하다고 한다. 하지만 가장 간지나는 일은 모델 개발 아닌가? 그래서 다들 모델 개발 쪽에 공부하는 시간을 많이 투자하는 것 같다. 사실 나도 모델 개발 공부가 가장 전문적이고 재밌어보인다.

재밌는 일만 할 수는 없으니까.. 1번에서는 작업 시간의 대부분을 잡아먹는 문제 정의 및 데이터 정제 파트를 이야기한다. 나름 이 파트도 깊게 파고들면 확률적인 데이터 클리닝 방법론이 많다. 물론 여기서는 전체적인 워크플로를 볼 것이기 때문에 깊게 들어가진 않는다.

✔ 어떤 데이터가 문제를 잘 설명하지? 🧐

❓❓❓❗

머신러닝을 쓴다고 항상 문제가 해결되진 않는다. 오히려 전통적인 통계 모델링이 더 효과적일 수 있다. 그렇기에 문제를 자세히 파고들어 룰을 이해하고 가장 적합한 솔루션을 찾아야 한다. 물론 인공지능 성능이 너무 좋아서 대충 때려넣어도 해결될 때가 있지만... 그러면 안된다.

• 입력 데이터는 무엇인가? 무엇을 예측하려고 하는가?
• 이진분류인가? 다중분류인가? 스칼라 회귀인가? 벡터 회귀인가? 이미지 분할? 강화학습?
• 기존 솔루션은 어떤 것이 있을까?
• 고려해야 할 특별한 제약이 있을까? 모델이 엔드-투-엔드로 엄격히 암호화되어 있어 모델을 꼭 사용자의 디바이스에 넣어야 한다든지..

일단 문제를 들어가기 전에 물음표를 계속 던질 필요가 있다. 기획 단계에서 최-대한 꼼꼼하게 해두지 않아서 후반 단계에서 차질이 생기면 일을 다시해야 할 수도 있다.

또 이 문제를 머신러닝으로 풀겠다는 것은

1. 주어진 입력으로 타깃을 예측할 수 있고
2. 데이터가 결과에 대한 충분한 정보를 담고있다고 가정한다.

나는 이걸 제대로 안해서 프로젝트를 통째로 날려먹은 적이 있다...

라벨링 노가다 😫

문제가 뭔지 파헤치면서 작업 특성을 이해하고, 입력과 타깃이 무엇인지 알게되면 데이터를 수집하기 시작한다. 책에서는 '노동집약적인 파트'라고 고급지게 표현했지만 내 표현이 더 적절한 거 같다.

대부분 연구소에서는 이런 일을 막내에게 시키거나 외주를 맡긴다 (그게 나다). 혹시 학교 학부연구생에게 라벨링을 시킨다고 하면, 내가 연구소에 배우러 가는지 일하러 가는지 비율을 진지하게 고민해봐야 한다.

시간은 많이 들겠지만 최대한 편한 라벨링 작업을 하기 위해 개발자들은 고민해야한다. 예를 들어 이미지, 영상에서 바인딩 박스를 어떻게 효율적으로 체크할 수 있을지 고민해보자. 온라인에서 '바인딩 박스 툴'을 다운받아 열심히 수작업으로 할 수도 있다. 하지만 이는 한계가 있으니 바인딩 박스를 만들어주는 인공지능 모델을 따로 만들 수도 있다. 모델이 작업한 박스를 검수만 하면 되니 시간이 훨씬 빨라진다. 간단한 토이 프로젝트일 경우 전자가, 규모가 비교적 큰 프로젝트면 후자가 좋은 선택지일 것이다.

또 라벨링을 누구나 할 수 있는지, 아니면 전문지식이 있는 사람만 할 수 있는지도 고려해야한다. 위의 바인딩 박스는 누구나 할 수 있다. 한편 신약 개발을 위한 유력 시료 판단을 한다고 해보자. 시료 후보군이 1만개가 있고, 하나의 시료를 약으로 만들어 실험해보는데 3시간이 걸린다고 하자. 만약 1만개를 전부 실험해보려면 3만시간이 걸린다. 이 신약 개발 주제로 박사 논문을 쓰다 잘못하면 졸업까지 10년이 걸릴 수도 있다. 그러니 유력 시료를 판단하는 딥러닝 모델이 필요하다. 이 모델에 들어가는 데이터는 시료에 대한 전문 지식일 것이다. 이런 모델은 전문가만 라벨링할 수 있다. 과연 전문 인력을 라벨링에 쓸 정도로 좋은 주제인가? 고민해봐야 한다.

데이터의 실효성은 반드시 고민해야 한다. 온라인에 올라온 음식점 음식 사진들과 내가 찍은 음식 사진은 많이 차이가 난다. 포토샵과 전문성에서 많이 밀린다. 만약 온라인의 전문 사진사가 찍은 사진으로만 모델을 학습시키고, 내 사진으로 예측을 한다면? 못 맞힐 수도 있다. 그러니 실제 예측에 사용될 데이터와 모델 학습 데이터를 최대한 일치시켜야 한다.

마지막으로 '개념 이동(concept dift)'도 주의해야 한다. 2013년에 훈련한 음악 추천 엔진으로 지금 추천을 부탁하면 영 시원찮은 결과가 나올 수 있다. 어휘, 표현, 장르의 트렌드가 시기에 따라 변하기 때문이다. 모델의 데이터도 트렌드에 맞춰야 한다. 지속적인 데이터 수집과 재훈련이 필요하다.

✔ 우리의 문샷은 이거야! 🌚

앞서 말했듯 데이터에 대한 이해 없이 대충 모델에 때려넣으면 될거라는 믿음은 가지면 안된다. 모델 훈련에 들어가기 전, 데이터를 탐색하고 시각화하여 예측 능력을 가진 특성이 뭔지 고민해봐야한다. 이를 특성 공학이라고 한다.

• 이미지나 자연어 텍스트는 직접 확인해볼 수 있다.
• 수치는 히스토그램을 그려 범위, 빈도를 확인해볼 수 있다.
• 데이터가 위치 정보를 포함하면 지도에 그려본다.
• 누락된 값을 전처리도 해야한다.
• 분류 작업이라면 각 클래스 별로 데이터의 수가 균등하게 있는지 고민해본다. 만약 하나에만 데이터가 몰려있다면 정확도에 악영향을 준다.
• '타깃 누출'을 확인한다. 타깃에 너무 직접적인 영향을 주는 특성이 섞여있다면 그 특성에만 가중치가 몰릴 수 있다.

그 다음 성공 지표를 선택한다. 정확도, 정밀도, 재현율, 고객 재방문율 등 모델의 목적에 따라 성공 지표가 달라진다. 모델의 단순한 수치 연산 결과를 고수준의 목표로 연결시키려면 논리적인 설명이 뒷받침되어야 한다.

2. 딥-러닝 모델 개발!

✔ 데이터 클리닝 ✨

벡터화

데이터를 신경망에 넣으려면 부동 소수점 데이터로 이루어진 텐서 형태로 변환해줘야 한다. 이 단계를 데이터 벡터화라고 부른다.

값 정규화

이미지 데이터를 그레이스케일 인코딩인 0255 사이의 정수로 인코딩하거나, 각기 다른 특성들의 범위를 01의 확률공간으로 압축시켜버리는 등의 정규화 작업을 할 수 있다. 키(cm)랑 몸무게(kg)로 성별을 맞추는데 값을 정규화 없이 넣으면 비교적 숫자가 높은 키에 편향된 결과가 나올 수 있다.

누락된 값 처리

데이터에 NaN이나 Null 같은 빵꾸가 나있으면 학습이 덜 된다. 평균값, 최빈값, 앞에 값 끌어다 쓰기 등 일반적인 처리 방법부터 통계학적인 방법까지 다양한 처리 방법이 있다.

✔ 과대적합 모델 만들기 📈

데이터를 넣고 훈련을 할 때 신경망 가운데 층이 너무 작지도, 너무 크지도 않아야 한다. 적당한 크기를 찾기 위해 작은 모델부터 점점 크기를 키워가며 적당한 신경망을 찾는다. 크기를 기우는 방법은

1. 층 추가하기

2. 층 크기를 키우기!

3. 더 오래 훈련하기 (에포크 Up)

가 있다.

또 다른 방법으로 모델 규제와 하이퍼파라미터 튜닝이 있다.

1. 다른 구조를 시도해본다. 층을 추가하거나 제거한다.
2. 드롭아웃을 추가한다.
3. 모델이 작다면 L1 또는 L2규제를 추가한다.
4. 최적의 설정을 찾기 위해 하이퍼파라미터를 바꿔서 시도해본다.
5. 데이터 큐레이션이나 특성 공학을 시도해본다.

3. 유저님, 맛이 어떠신가요?

✔ 모델 설명하기

인공지능 모델을 다 만들었다고 끝난게 아니다. 마지막으로 유저님 밥상에 가져다 놔야 한다.

우선 AI를 잘 모르는 사람들에게 내가 만든 AI를 잘 설명해야 한다. 이때 "98% 정확도의 모델입니다"라고 설명하면 비전문가는 반올림해서 100%로 알아듣는다. 이는 자칫 오해를 초래한다. 매일 평균 1만건의 메일 중에서 스팸 메일을 판별하는 모델이 있다면 하루 평균 2천건의 거짓 판별이 나온다. 그러므로 비전문가에게 말할 때는 거짓 음성 비율과 거짓 양성 비율을 정확히 말할 필요가 있다. 책에는 다음과 같은 예시를 들어주었다.

"이런 설정에서는 부정 거래 감지 모델이 5% 거짓 음성 비율과 2.5%의 거짓 양성 비율을 달성합니다. 매일 평균 200건의 유효한 거래가 부정 거래로 표시되어 수동 검토를 위해 전달됩니다. 평균적으로 14개의 부정 거래를 놓치며 266개의 부정 거래를 정확하게 감지합니다."

✔ 모델 배포하기 🍻

제품의 환경이 파이썬을 지원하지 않는다면 파이썬이 아닌 다른 방식으로 모델을 배포해야 한다. 여러가지 방법이 있는데 한번 훑어보자.

서버에서 돌리기: REST API

서버나 클라우드 인스턴스에 텐서플로를 설치하고 REST API로 모델의 예측을 요청하는 방식이다. 요새는 클라우드에서 Docker를 활용해 그 안에 파이썬 코드를 넣는다. Docker 사용법은 내 velog에 나와있으니 확인해보길 바란다..ㅎㅎ

플라스크로 직접 서빙 앱을 만들거나 텐서플로 자체 라이브러리인 텐서플로 서빙을 사용하면 된다. 단, 이 방법을 사용하려면 클라이언트가 서버에 접속 가능해야 하고, 응답 속도 면에서도 고려해야 하며, 데이터가 크게 민감하지 않아야 클라우드로 주고 받을 수 있다. 이런 부분들을 고려해야 한다.

로컬에서 돌리기: 장치로 배포

모델을 장치에서 직접 구동해야할 경우 장치 안에 모델을 임베딩해야 한다. 이런 경우 모델을 장치에 들어갈 정도로 작게 최적화해야 한다. 장치의 사양에 따라 다양한 변수가 생길 수 있다.예를 들면, 바다를 가로지르는 배에 자율주행 모델을 심는다면 커다란 컴퓨터 장치를 싣을수 있으니 크기가 상관 없겠지만 드론은 꽤 제약이 있다. 해양에서는 클라우드로 육지의 서버와 통신도 어려우니 드론을 조종하는 장치에서 모든게 처리되어야 한다. 이런 작은 컴퓨터에서도 모델이 돌아갈 수 있도록 Llama2 같은 소형 모델들이 나오는 것이다.

아니면 회사 기밀 문서들을 학습시키는 모델 같은 경우 클라우드로 정보를 주고 받기 어렵다. 이러면 사내 장치로 모델을 돌릴 수 밖에 없다.

추론모델 최적화

로컬에서 모델을 돌리려면 모델을 작게 만드는게 좋다. 최적화를 위해 보통 가중치 가지치기, 가중치 양자화 기법이 사용된다.

• 가중치 가지치기: 드롭아웃처럼 신경망 내의 텐서를 죽인다. 단, 여기서는 랜덤이 아니라 철저하게 필요 없는 텐서를 삭제한다.
• 가중치 양자화: 딥러닝 모델은 부동 소수점(float32) 가중치로 훈련된다. 이를 정수(int8)로 반올림하면 크기가 1/4로 줄어든다.

모델 유지 관리

모델 배포를 마쳤다면 끊임없이 성과 지표를 확인하며 개선 방안을 모색한다. MLOps 직무가 이런 일을 하는 것 같다.

글을 정리하며

대학생이 이 전 과정을 거칠 수 있는 가장 좋은 방법은 공모전이 아닐까? Kaggle처럼 정제된 데이터로 연습할 수 있는 플랫폼이 있어 얼마나 다행인지 모른다.

끝!

랭체인(LangChain) 랭체인(LangChain)은 언어 모델을 간편하게 개발하기 좋은 프레임워크다. 현재 버전 0.0.256까지 릴리즈되어 비교적 신생이지만, 한 달에 400만 이상의 사람이 다운받는 등 관심도와 참여도로 보아 무섭게 성장할 것으로 보인다. 여러 종류의 언어 모델은 각기 다른 입출력 형식과 자잘한 차이점들을 가지고 있다. 서로 다른 모델을 모듈화하여 레고 조립하듯이 언어 모델 서비스를 개발할 수 있다.

그라디오와 랭체인을 활용한 챗봇 서비스 개발 이 글에서는 그라디오와 랭체인을 활용하여 기초적인 챗봇 서비스를 개발하는 과정을 다룬다. 그라디오로 UI(User Interface)를 구현하고, 랭체인으로 간단한 챗봇을 개발하여 그라디오에 연결시킨다.

1. 그라디오의 기본 구조

그라디오는 블록을 쌓듯이 UI를 구현할 수 있는 프레임워크다. React.js에서 컴포넌트를 만들어 구조화하듯이 그라디오에서 제공하는 컴포넌트를 쌓고 함수를 연결하여 사용한다. 그라디오를 사용하기 위해 다음 명령어를 명령 프롬프트(cmd)에 입력한다.

pip install gradio

아래 표는 자주 사용하는 그라디오 컴포넌트에 대한 설명이다. 이를 참고하여 아래 그라디오의 기본 기능이 들어간 예제 코드를 살펴보자.

# gradio library를 가져온다. gr로 줄여서 칭한다.
import gradio as gr

# gradio 내부에서 사용할 함수다.
def update(name): 
    return f"Welcome to Gradio, {name}!"

with gr.Blocks() as iface:
    gr.Markdown("Hello, World! I’m yelling in gradio!")
    with gr.Row():
        inp = gr.Textbox(placeholder="이름이 무엇인가요?")
        out = gr.Textbox()
    btn = gr.Button("제출")

    # 버튼에 이벤트 리스너를 추가한다. 
    # 버튼 클릭시 update함수를 호출하고, inp에 입력된 문자열을 파라미터로 보낸다. 함수의 반환값은 out에 출력한다.
    btn.click(fn=update, inputs=inp, outputs=out)

iface.launch()

위 코드를 실행시키고 웹 브라우저(Chrome)로 접속한 화면이다.

Input(왼쪽 텍스트박스)에 이름을 입력한 결과다. Input 입력 후 버튼 클릭시 update 함수를 호출하고, Input에 입력된 문자열을 파라미터로 보낸다. 함수의 반환값은 Output(오른쪽 텍스트박스)에 출력한다.

그라디오는 UI를 만들 때 블록을 쌓듯 컴포넌트를 세로로 차곡차곡 쌓는다. Fig3은 예제 코드의 컴포넌트가 어떻게 쌓여있는지 이미지로 보여준다.

2. 그라디오로 챗봇 UI 만들기

그라디오의 옛 버전에서는 Blocks 객체 내에서 여러 블록을 쌓고, 이벤트리스너(Event Listener)를 추가하여 챗봇을 구현하였다. 하지만 업데이트된 그라디오 3.39.0버전에서는 챗봇의 기능이 대부분 담긴 ChatInterface 컴포넌트가 생겼다. ChatInterface 컴포넌트 안에 챗봇UI, 입력, 버튼, 추가 옵션들을 모두 파라미터로 설정할 수 있다.

# 챗봇에 채팅이 입력되면 이 함수를 호출합니다. 
# message는 유저의 채팅 메시지, history는 채팅 기록, additional_input_info는 additional_inputs안 블록의 정보를 받습니다.
def response(message, history, additional_input_info):
    # additional_input_info의 텍스트를 챗봇의 대답 뒤에 추가합니다.
    return "챗봇을 미완성하였습니다 " + additional_input_info

gr.ChatInterface(
        fn=response,
        textbox=gr.Textbox(placeholder="말걸어주세요..", container=False, scale=7),
        title="어떤 챗봇을 원하심미까?",
        description="물어보면 답하는 챗봇임미다.",
        theme="soft",
        examples=[["안뇽"], ["요즘 덥다 ㅠㅠ"], ["점심메뉴 추천바람, 짜장 짬뽕 택 1"]],
        retry_btn="다시보내기 ↩",
        undo_btn="이전챗 삭제 ❌",
        clear_btn="전챗 삭제 💫",
        additional_inputs=[
            gr.Textbox("!!!", label="끝말잇기")
        ]
).launch()

ChatInterface로 만든 챗봇 화면이다. 원하는 값만 넣었는데 결과가 꽤 멋있다!

3. 랭체인(LangChain)으로 챗봇 함수 만들기

그라디오로 유저와 상호작용할 수 있는 UI를 만들었다. 랭체인은 유저에게서 받은 메시지를 가공할 함수를 만드는 언어모델 프레임워크다. 랭체인을 활용해 OpenAI의 GPT-3.5모델을 연결하여 답변을 생성한 후 반환하는 함수를 만들어보자. 랭체인은 언어 생성 모델과 임베딩 모델을 패키징해줘 쉽게 import해 쓸 수 있다. 랭체인을 다운받기 위해 아래 커맨드를 명령 프롬프트(cmd)에 입력한다.

pip install langchain

아래는 그라디오 예제의 response함수를 랭체인을 활용해 OpenAI의 GPT-3.5모델을 연결한 코드다.

from langchain.chat_models import ChatOpenAI
from langchain.schema import AIMessage, HumanMessage
import os

# os의 환경변수로 OPENAI_API_KEY를 설정한다. “...”에 api key를 입력해야 동작한다.
os.envrion["OPENAI_API_KEY"] = "..."  

# langchain의 GPT-3.5 모델을 가져온다.
# temperature는 챗봇 대답의 자유도를 의미한다.
llm = ChatOpenAI(temperature=1.0, model='gpt-3.5-turbo-0613')

# 그라디오에서 유저의 질문과 채팅기록을 받아 llm의 input형식에 맞게 변환 및 입력 후 GPT-3.5의 대답을 리턴한다.
def response(message, history, additional_input_info):
        history_langchain_format = []
        for human, ai in history:
                history_langchain_format.append(HumanMessage(content=human))
                history_langchain_format.append(AIMessage(content=ai))
        history_langchain_format.append(HumanMessage(content=message))
        gpt_response = llm(history_langchain_format)
        return gpt_response.content

랭체인과 그라디오를 활용한 챗봇 만들기 위에서 살펴본 랭체인과 그라디오를 합친 전체 코드다. additional_input에서 시스템 프롬프트를 입력받아 langchain에게 전달하는 기능을 추가했다.

최종! 랭체인과 그라디오를 활용한 챗봇 만들기

from langchain.chat_models import ChatOpenAI
from langchain.schema import AIMessage, HumanMessage, SystemMessage
import gradio as gr
import os

os.envrion["OPENAI_API_KEY"] = "..." 

llm = ChatOpenAI(temperature=1.0, model='gpt-3.5-turbo-0613')

def response(message, history, additional_input_info):
        history_langchain_format = []
        # additional_input_info로 받은 시스템 프롬프트를 랭체인에게 전달할 메시지에 포함시킨다.
        history_langchain_format.append(SystemMessage(content= additional_input_info))
        for human, ai in history:
                history_langchain_format.append(HumanMessage(content=human))
                history_langchain_format.append(AIMessage(content=ai))
        history_langchain_format.append(HumanMessage(content=message))
        gpt_response = llm(history_langchain_format)
        return gpt_response.content

gr.ChatInterface(
        fn=response,
        textbox=gr.Textbox(placeholder="말걸어주세요..", container=False, scale=7),
        # 채팅창의 크기를 조절한다.
        chatbot=gr.Chatbot(height=1000),
        title="어떤 챗봇을 원하심미까?",
        description="물어보면 답하는 챗봇임미다.",
        theme="soft",
        examples=[["안뇽"], ["요즘 덥다 ㅠㅠ"], ["점심메뉴 추천바람, 짜장 짬뽕 택 1"]],
        retry_btn="다시보내기 ↩",
        undo_btn="이전챗 삭제 ❌",
        clear_btn="전챗 삭제 💫",
        additional_inputs=[
            gr.Textbox("", label="System Prompt를 입력해주세요", placeholder="I'm lovely chatbot.")
        ]
).launch()

꽤 괜찮다

정리하며

챗봇 프로토타입 개발 과정을 살펴보았다. 이를 응용하여 다양한 챗봇을 만들 수 있을 것이다. 그라디오와 랭체인은 메인 인공지능 설계를 제외한 작업을 간단히 만들어준다. 그러면 인공지능 설계에 들일 시간과 노력을 확보할 수 있다. Fig5의 짜장면과 짬뽕 질문에 챗봇이 두툼한 카라멜이라는 엉뚱한 대답을 내놓은 것처럼, UI와 기능을 간결하게 만들어도 핵심은 인공지능의 성능이다. 프레임워크를 활용해 인공지능을 연구할 시간을 확보할 수 있다.

참고 문헌 LangChain. LangChain Docs. 2023.08.09 검색, https://python.langchain.com/docs Gradio. Gradio Docs. 2023.08.09 검색, https://www.gradio.app/docs

끝

도커(Docker)에서 웹서버 호스팅하기

도커(Docker)는 컨테이너 단위로 애플리케이션을 실행해주는 플랫폼이다. 로컬 환경에서 개발한 애플리케이션을 다른 환경에서 실행하면 패키지의 버전이 충돌하여 오류가 뜨는 경우가 많다. 라이브러리가 업데이트되며 이전 버전에서 사용하던 함수가 사장되거나, 각기 다른 라이브러리가 함께 의존중인 라이브러리에게 요구하는 버전이 다를 수 있다. 이런 충돌을 방지하고자 도커는 컨테이너에 애플리케이션을 위한 최적 환경을 구성한다. 그리고 컨테이너의 환경 정보와 애플리케이션이 모두 저장된 이미지 파일을 주고받는다. 도커는 로컬에서 작업한 애플리케이션을 클라우드 서버로 옮겨 실행하거나, 수십 년 전 작성된 코드를 당시 환경의 컨테이너에서 쾌적하게 실행할 수 있도록 돕는다. 어떤 작업이든 초기 환경설정에 소요되는 시간을 크게 절약할 수 있다.

도커의 컨테이너는 호스트 컴퓨터(컨테이너를 실행시킨 컴퓨터) 안에 독자적인 환경을 구축하여 애플리케이션을 실행한다. 때문에 컨테이너에서 웹서버를 호스팅할 때 호스트에서 하는 것과 몇 가지 차이점이 있다. 이 글에서 챗봇을 호스팅해보고 과정에서 필요한 설정들을 알아볼 것이다.

1. 도커 파일 작성

FROM python:3.10.11-bullsey
# FROM은 도커의 베이스 이미지를 나타낸다. 
# 챗봇은 python(.py)으로 작성됐으므로 python:3.10.11-bullseye을 사용했다.

COPY docker_chatbot_package docker_chatbot_package
# COPY는 로컬파일에서 작업한 파일들을 이미지에 저장하겠다는 의미다.
# 첫 번째 ‘docker_chatbot_package’는 로컬에서 복사할 폴더를 의미하고 두 번째는 컨테이너 안에 저장할 위치다.

WORKDIR /docker_chatbot_package
# WORKDIR은 컨테이너 안에서 커맨드의 시작 위치를 바꾼다. 
# ‘docker_chatbot_package’ 폴더 안에서 시작하도록 설정했다.

RUN pip install -r /docker_chatbot_package/requirements.txt
# RUN 은 커멘드 라인을 실행할 수 있도록 해준다. 
# requirements.txt 안에 적힌 패키지를 모두 다운받도록 했다. requirements.txt 의 내용은 다음과 같다.

EXPOSE 7860
# EXPOSE는 웹서버를 호스팅할 포트를 명시한다. 
# 7860번 포트로 호스팅을 하였다. 여기서 명시하더라도 컨테이너를 실행할 때 –p 옵션으로 컨테이너와 호스트 컴퓨터의 포트를 연결해야한다. 밑에서 컨테이너를 실행할 때 추가로 설명하겠다.

CMD ["python3", "web_test_gradio_ui_memo.py"]
# CMD는 컨테이너가 실행될 때 한 번만 실행할 명령어를 기술한다.
# ‘python3’로 ‘web_test_gradio_ui_memo.py’ 파일을 실행하도록 했다.

# ‘requirements.txt ’의 내용이다.
# 다운로드 받을 패키지 목록이다.
gradio
tiktoken
openai
pandas
scipy

2. Gradio를 활용한 챗봇 호스팅

이 글에서 챗봇의 구조를 자세히 다루지는 않는다. 하지만 호스팅 부분을 설명하기 위해 챗봇 제작에 사용한 Gradio를 간단히 소개하겠다. Gradio는 인공지능을 개발하고 테스트하기에 유용한 파이썬 프레임워크다. 파이썬으로 블록을 쌓듯이 UI를 구현하고, 호스팅할 수 있도록 해준다. 아래는 Fig 4.를 구현한 Gradio 코드의 일부다.

import gradio as gr
with gr.Blocks() as iface:
  … 화면 구현 …
iface.launch(share=True, server_name="0.0.0.0")

iface.launch()의 매개변수가 호스팅과 관련있다. ‘Share=True’로 설정시 월드와이드웹(World Wide Web, WWW)에 Gradio가 호스팅해준다. ‘Server_name’ 옵션은 파일을 호스팅할 로컬 주소를 뜻한다. ‘127.0.0.1‘로 설정하면 로컬에서 호스팅이 된다. ’0.0.0.0‘으로 설정하면 모든 IP주소에서 접속이 가능하다.

컨테이너라는 독자적 생태계

컨테이너는 고유의 환경 변수와 IP주소를 갖고 있다. 컨테이너 안의 IP주소는 도커가 부여한 가상의 값이기 때문에 호스트 컴퓨터의 IP주소 체계와 별개다. 컨테이너에서 ‘127.0.0.1’로 IP주소를 설정하면 컨테이너 자신을 의미한다.

> docker run –p 127.0.0.1:8000:8000 [IMAGE_NAME]
# 이미지를 빌드할 때, –p 옵션으로 컨테이너의 8000번 포트와 호스트 컴퓨터의 8000번 포트를 연결한다.
# 호스트 컴퓨터에서만 서버에 접속할 수 있다.

> docker run –p 8000:8000 [IMAGE_NAME]
# 마찬가지로 컨테이너와 호스트 컴퓨터의 8000번 포트를 연결한다.
# 외부 컴퓨터(IP가 127.0.0.1이 아닌 컴퓨터)도 포트를 통해 호스팅 서비스에 접근 가능하다.

Gradio에서 연결 IP주소를 ‘0.0.0.0’으로 설정해 두면 어떤 IP주소를 가진 컴퓨터도 접근할 수 있다. 하지만 컨테이너에 접속하려면 호스트 컴퓨터를 통해야 하므로 결국 호스트 컴퓨터만 접근이 가능하다. ‘docker run –p’ 옵션에서 호스트의 IP주소를 127.0.0.1:8000으로 명시하면 호스트만 포트를 연결한다. 이를 생략하면 호스트 이외의 컴퓨터도 ‘외부 컴퓨터-호스트-컨테이너’ 순서로 컨테이너의 서비스에 접근이 가능하다.

3. 도커파일 빌드 및 챗봇 컨테이너 호스팅

앞의 도커 파일을 빌드 하여 챗봇 이미지를 만든다. 이미지를 RUN하여 컨테이너를 만들고 호스팅해보자. 아래는 이와 관련된 코드다.

> docker build -t dokdo_chatbot .  
# 현재 폴더(마지막 . 이 현재 폴더 의미)의 ‘dockerfile’을 찾아 이미지로 build한다. 
# 이미지의 이름은 ‘dokdo_chatbot’이다 (-t 옵션).

> docker images
# 이미지 목록을 출력한다.
# 이미지 목록에 ‘dokdo_chatbot’이 있다면 이미지가 정상적으로 빌드된 것이다.

> docker volume create chatbot_chatlog
# 볼륨을 생성한다. 이름은 ‘chatbot_chatlog’이다.

> docker run -d –p 7860:7860 –v chatbot_chatlog:/dokdo_chatbot_package/data  --name my_chatbot dokdo_chatbot
# 챗봇 이미지를 실행한다. -p 옵션으로 7860포트를 연결한다.
# -d 옵션으로 컨테이너를 백그라운드에서 실행했다. 이 옵션을 없애면 컨테이너의 로그를 볼 수 있다.
# -v 옵션으로 컨테이너에 볼륨을 마운트했다. ‘chatbot_chatlog’ 볼륨을 컨테이너의 /dokdo_chatbot_package/data에 배치했다.
# --name 옵션으로 컨테이너의 이름을 ‘my_chatbot’으로 만들었다.

실행된 어플리케이션은 Fig 4.와 같다. ‘localhost:7860’에서 실행된 것을 확인할 수 있다.

정리하며

챗봇을 예제로 웹서버를 호스팅해보았다. 응용한다면 다양한 서비스를 도커에서 운영할 수 있을 것이다.

참고문헌 엘튼 스톤맨. 『도커 교과서』. 심효섭(역). 길벗, 2022.

참고한 책: 케라스 창시자에게 배우는 딥러닝 2판

케라스(Keras)는 최고(Best)다.

지난 글에서 정리한 1~2장은 수학적 배경지식과 딥러닝에 대한 설명이었다면 3장에서는 텐서플로우와 케라스에 대한 이해도를 높일 수 있었다. 이 장을 읽으면서 단순히 케라스의 모델을 import해서 쓰는게 아니라 케라스가 어떻게 동작하는지, tensorflow로 단순한 케라스 객체를 따라 만들어보며 이해할 수 있었다.

케라스를 만든 이 책의 저자는 얼마나 이론을 깊게 이해하고 또 구현 능력을 갖춘 것일까? 나도 열심히 공부해서 케라스나 텐서플로우 같은 대형 프레임워크의 컨트리뷰터로 활동하고 싶다. 그 날을 위해 오늘도 열심히 공부하자!

텐서플로우 (TensorFlow)

텐서플로우는 넘파이처럼 수치를 나타내기 위한 라이브러리다. 넘파이처럼 빠른 병렬 연산이 가능하다. 그럼 넘파이와 차이점은 무엇일까?

• 미분 가능한 식을 넣으면 자동으로 Gradient를 계산해주는 "GradientTape" 기능이 있다.
• GPU와 TPU에서 실행할 수 있다.
• 연산을 분산하기 쉽다. (tf로 정의한 식을 분산 연산한다.)
• 배포가 쉽다.

텐서플로우 연산

텐서플로우에서 사용하는 기본 연산을 알아보자. 기본적인 연산 예제와 주석을 달아두었다. 간편하게 복습하기 좋다.

import tensorflow as tf

a = tf.zeros(shape=(2, 1))     # (2, 1) 행렬을 만들고 0으로 채움
b = tf.ones(shape=(2, 1))     # (2, 1) 행렬을 만들고 1로 채움

c = tf.Variable(3.) # 변수
d = tf.constant(3.) # 상수

e = tf.random.normal(shape=(3,1), mean=0, stddev=1)  # 평균 0, 표준편차 1 정규분포에서 (3, 1) 행렬 뽑음
f = tf.random.uniform(shape=(3,1), minval=0, maxval=1) # 0~1 균등분포에서 (3, 1) 행렬 뽑음

tf.~ 식으로 쉽게 행렬을 만들 수 있다. 0으로 채우기, 1로 채우기, 정규분포로 채우기, 균등분포로 채우기 등 다양한 데이터를 만들 수 있다. 변수는 tf.Variable()로 첫번 째 문자가 대문자이다!

a = tf.Variable(a)
a.assign(tf.ones(((2,1))))     # numpy와 다르게 tf는 object형태라 assign 메소드로 따로 할당
a[0,0].assign(3.)             # 개별 할당 가능
a.assign_add(a)     # +=
a.assign_sub(a)     # -=
a = tf.square(a)      # 제곱
a = tf.sqrt(a)        # 제곱근
b = tf.transpose(a)    # 전치
e = tf.matmul(a, b) # 점곱
e *= e                # 원소별 곱셈

tf는 변수와 상수 모두 tf 객체로 만들기 때문에 assign 메소드로 값을 할당해야 한다. 인덱싱 기능을 제공해서 행렬 원소만 할당할 수 있다.

# GradientTape는 미분 가능한 식을 넣으면 자동미분해준다.
# 아래는 위치, 속도, 가속도 예제이다.

time = tf.Variable(0.)
with tf.GradientTape() as outer_tape:
  with tf.GradientTape() as inner_tape:
    position = 4.9 * time ** 2
  speed = inner_tape.gradient(position, time)
acceleration = outer_tape.gradient(speed, time)

위는 GradienTape를 활용한 속도, 가속도 구하기 예제이다. with ~ tape: 안에 수식을 넣고 밖에서 tape의 gradient(미분값)을 구할 수 있다.

수식은 미분 가능해야 한다. 수식의 tf 변수 객체만 미분 값을 제공하므로 timed을 tf.constant()로 지정하면 미분이 되질 않는다.

케라스 대신 텐서플로우로 딥러닝 모델을 구현해보면 깊이 공부한 건 줄 알았는데, 텐서플로우 안에 들어있는 수식 함수들과 C언어 연산이라는 심해를 보았다. GradientTape 안에 들어있는 함수를 만들 수 있는 사람은 얼마나 될까? 경쟁력은 이런 기초에서 나오는 것이다! 정말이지 공부할 건 넘쳐나는 것 같다. 우선 텐서플로우와 수학적 원리를 조금씩 이해해보자! 라는 마음으로 책을 읽었다.

텐서플로우로 선형분류기 구현하기

그럼 이제 본격적으로 텐서플로우를 활용해 선형분류기를 만들자! 선형분류기는 케라스의 기본적인 객체 중 하나이다. 일반적인 레이어 하나가 선형분류기라고 보면 된다. 물론 층을 쌓을 수록 활성화함수를 거쳐 비선형이 된다. 여기서는 하나의 레이어가 어떻게 구성되어있는지 텐서플로우로 구현하며 이해하겠다.

import numpy as np

num_samples_per_class = 1000
negative_samples = np.random.multivariate_normal(
    mean=[0, 3],                # 평균
    cov=[[1, 0.5], [0.5, 1]],    # 공분산
    size=num_samples_per_class    # 데이터 개수
)
positive_samples = np.random.multivariate_normal(
    mean=[3, 0],
    cov=[[1, 0.5], [0.5, 1]],
    size=num_samples_per_class
)

우선 임의의 데이터를 만든다. 두 개의 집단을 만들고 집단 당 데이터는 1000개씩 만든다. 각 집단은 두 개의 특성을 갖고있다 (x축 y축으로 통칭). negative_samples는 정규분포에서 뽑으며 평균, 공분산, 자료의 개수를 지정해주었다. positive_samples도 마찬가지다.

inputs = np.vstack((negative_samples, positive_samples)).astype(np.float32)
targets = np.vstack((np.zeros((num_samples_per_class, 1), dtype="float32"),
                   np.ones((num_samples_per_class, 1), dtype="float32")))

inputs에 두 자료를 뭉쳐놓고, targets에서 0과 1로 라벨링을 해준다.

import matplotlib.pyplot as plt

plt.scatter(inputs[:, 0], inputs[:, 1], c=targets[:, 0])
plt.show()

plt.scatter( )에서 c 옵션을 target[ : , 0 ], 즉 라벨 번호로 지정해준다. c 옵션의 활용법을 배웠다.

input_dim = 2    # 특성 2개
output_dim = 1    # 결과는 Class 분류

W = tf.Variable(initial_value=tf.random.uniform(shape=(input_dim, output_dim)))
b = tf.Variable(initial_value=tf.zeros(shape=(output_dim,)))

W는 initial_value의 shape가 (2, 1)이다. x축과 y축 각각의 특성에 대한 W1, W2가 필요하기 때문이다. 반면 b는 (1, )로 하나만 붙는다.

# 선형 분류기
def model(inputs):
  return tf.matmul(inputs, W) + b

정방향 패스의 연산을 반환한다. inputs의 shape은 (1000, 2) W는 (2, 1)이므로 점곱을 하면 (1000, 1)이 된다. 이후 b를 더해준다.

# 손실함수
def square_loss(targets, predictions):
  per_sample_losses = tf.square(targets - predictions)
  return tf.reduce_mean(per_sample_losses)

손실함수는 target과 prediction의 오차를 계산해 제곱한다. 여기서 target은 0 또는 1 뿐이다. model은 앞서 만든 선형 분류기이므로 랜덤이다. tf.reduce_mean( ) 함수가 모든 loss의 평균을 Scala값으로 만들어준다.

learning_rate = 0.1

def training_step(inputs, targets):
  with tf.GradientTape() as tape:
    predictions = model(inputs)
    loss = square_loss(targets, predictions)
  grad_loss_wrt_W, grad_loss_wrt_b = tape.gradient(loss, [W, b])
  W.assign_sub(grad_loss_wrt_W * learning_rate)
  b.assign_sub(grad_loss_wrt_b * learning_rate)
  return loss

GradientTape( )를 이용해서 loss에 대한 W와 b의 편미분값을 계산한다. loss를 최대한 줄이는게 목적이다.

for step in range(40):
  loss = training_step(inputs, targets)
  print(f"{step} 번째 스텝의 손실: {loss:.4f}")

0 번째 스텝의 손실: 2.1274
1 번째 스텝의 손실: 0.2540
2 번째 스텝의 손실: 0.1287
3 번째 스텝의 손실: 0.1079 
...
37 번째 스텝의 손실: 0.0276
38 번째 스텝의 손실: 0.0273
39 번째 스텝의 손실: 0.0270

40번 학습시킨다. 여기서는 게산의 편의를 위해 전체 데이터를 계속 학습했지만, 데이터가 커지면 미니배치 경사하강법으로 학습을 시키는게 일반적이다.

predictions = model(inputs)
plt.scatter(inputs[:, 0], inputs[:, 1], c=predictions[:, 0] > 0.5)
plt.show()

잘 분류해낸 모습이다. 위의 원래 데이터에서 특이값(보라색이지만 노란 집단에 가까움)은 잡아내지 못했다. 선형분류기의 한계이다. 딥러닝은 직선이 아니라 고차원의 기준을 만들어내므로 특이값에 대한 분류가 더 용이할 것이다. 물론 여기서는 데이터가 너무 깔끔해서 그럴 필요는 없어보인다.

x = np.linspace(-1, 4, 100)
y = - W[0] / W[1] * x + (0.5 - b) / W[1]    # x * W[0] + y * W[1] + b = 0.5 식 이항
plt.plot(x, y, "-r")
plt.scatter(inputs[:, 0], inputs[:, 1], c=predictions[:, 0] > 0.5)
plt.show()

x * W[0] + y * W[1] + b = 0.5    #prediction의 예측 값. 0.5 미만이면 class 0, 초과면 class 1 이다.

선형분류기의 기준선을 그려보았다. x축 [-1, 4] 범위에 점을 100개 찍어서 선을 만든다. 이에 대응하는 y축 값들을 계산하여 plot한다. 이 기준선이 선형분류기가 데이터의 클래스를 판별하는 기준이다.

선형분류기 객체로 만들기

class SimpleDense(keras.layers.Layer):    # keras.layers.Layer 상속
  def __init__(self, units, activation=None):
    super().__init__()
    self.units = units                    # 출력 차원의 크기
    self.activation = activation        # 활성화함수

  def build(self, input_shape):
    input_dim = input_shape[-1]            
    self.W = self.add_weight(shape=(input_dim, self.units), initializer="random_normal")
    self.b = self.add_weight(shape=(self.units, ), initializer="zeros")

  def call(self, inputs):
    y = tf.matmul(inputs, self.W) + self.b
    if self.activation is not None:
      y = self.activation(y)
    return y

# keras.layers.Layer class의 __call__ 메소드
def __call__(self, inputs):
    if not self.built:
        self.build(inputs.shape)
        self.built = True
    return self.call(inputs)

앞서 살펴본 선형 분류기를 객체로 만든다면 이런 모습일 것이다. 만약 keras.layers.Layer class를 상속하지 않고 call() 함수에서 직접 build했다면 input_dim을 파라미터로 받아야 한다. 수동적으로 파라미터를 받지 않고 동적으로 input_dim을 추론하기 위해 call과 build 메소드를 따로 만든 것이다.

케라스 (Keras)

그럼 케라스는 무엇인가?

• TensorFlow 기반 딥러닝 프레임워크
• 텐서플로우에서 한단계 더 나아간 사람 친화적 프레임워크!

텐서플로우로 선형분류기를 구현해보았다. 그 다음에 케라스를 이용한다면? 훨씬 편하다! 케라스를 사용해서 앞서 만든 선형분류기를 구현해보자.

케라스로 선형분류기 구현하기

from tensorflow.keras import layers
layer = layers.Dense(32, activation="relu")

케라스는 레이어가 이미 구현되어 객체로 포장되어 있다. layers.Dense가 앞서 만든 SimpleDense와 유사한 구조를 갖고 있다. 물론 더 다양한 기능들이 포함되어 있다. 오픈소스 코드에 들어가서 코드를 뜯어보는 것도 재밌는 작업일 것 같다.

from tensorflow.keras import models
from tensorflow.keras import layers

model = models.Sequential([
    layers.Dense(32, activation="relu"),
    layers.Dense(32)
])

Sequential을 통해 여러 레이어를 블록 쌓듯이 연결한다.

신경망 설계에 필요한 것들

가설공간 정의하기

신경망을 설계할 때 가설공간을 정의하는 직관은 매우 중요하다. 이 직관은 튼튼한 기초 지식과 경험에서 나온다. 엔지니어의 진짜 실력 중 하나라고 말할 수 있다. 예를 들면 위에서 두 개의 집단이 선형으로 충분히 분류가 가능하다고 판단하면 단순히 선형분류기(아핀 변환)으로 해결이 가능하다. 하지만 데이터의 분포가 복잡해질 수록 적합한 신경망 모델의 설계가 중요해진다.

손실함수 (=목적함수)

손실함수는 최소화할 값을 의미한다. 책에 재밌는 예가 나온다. 만약 인공지능에게 "인류의 평균 행복지수 높이기"라는 목적을 갖고 손실함수로 인류의 평균 행복지수를 설정하면 어떻게 될까? 행복한 인간을 제외한 모든 인간을 사살한다는 답을 도출할 수도 있다. 손실함수를 어떻게 섬세하게 설정하느냐에 따라 인공지능의 결과가 만족스러울 수도 있고, 불만족스러울 수도 있다. 아래는 대표적인 손실함수 모델 리스트다.

CategoricalCrossentropy,
SparseCategoricalCrossentropy,
BinaryCrossentopy,
MeanSquareError,
KLDivergence,
CosineSimilarity,
...

옵티마이저

옵티마이저는 손실함수를 업데이트하는 방법이다. 기본적으로 신경망은 확률적 경사하강법(Stochastic Gradient descent)의 발전된 모델들을 많이 사용한다. 이전 포스트에서 정리한 모멘텀(Momentum) 개념 등 다양한 개선 모델이 있다. 아래는 대표적인 옵티마이저 리스트다.

SGD,
RMSprop,
Adam,
Adagrad,
...

측정 지표

측정 지표는 훈련 과정에서 모니터링할 성공의 척도다. 손실함수는 항상 0에 수렴하도록 작동한다. 측정 지표는 분류가 얼마나 정확히 되었는지 퍼센트로 표시하는 등의 역할을 한다. 측정 지표는 모델 최적화에 직접 관여하는게 아니라 모니터링 용도이므로 미분 가능하지 않아도 된다.

CategoricalAccuracy,
SparseCategoricalAccuracy,
BinaryAccuraccy,
AUC,
Precision,
Recall

케라스 메서드

케라스 모델에는 여러 메서드가 있다. 알아보자.

model.compile( )

메서드 훈련 과정을 설명하는 메서드다.

model.compile(
    optimizer="rmsprop",
    loss="meann_squared_error",
    metrics=["accuracy"]
)

model.fit( )

메서드 훈련 루프를 설명하는 메서드다.

history = model.fit(
    inputs,
    targets,
    epochs=5,
    batch_size=128,
    validation_data=(val_inputs, val_targets)
)

validation = model.evaluate()

model.predict( )

예측 값을 반환하는 메서드다. 그냥 model()로 전체 예측 값을 출력해도 되지만 한번에 병렬로 출력하려면 GPU에 부담이 갈 수 있다. predict 메서드는 batch_size로 계산하기 때문에 부담이 덜하다.

정리하며

그냥 모델 임포트해서 쓰는 것보다 이렇게 하나씩 텐서플로우로 만들고 이해하려 하니까 더 공부한 느낌이 든다. 나중에 나도 저런 모델을 직접 연구할 날이 올 것이다. 그 때를 위해 꾸준히 공부하자!

끝

딥러닝은 머신러닝의 하위 개념이다

질리도록 들었지만 딥러닝은 머신러닝의 하위 개념이다. 머신러닝은 3가지로 이루어진다.

• 입력 데이터 포인트: 데이터 전처리 및 가공 = 어떤 특성이 중요한가?

• 기대 출력: 라벨링 = 어떤 답을 원하는가?
• 알고리즘의 성능 측정: 학습(손실함수 값 0으로 만들기) = 얼마나 정확한가?

어떤 데이터를 넣으면 어떤 답이 나와야 하는지 알려주고, 머신러닝(ML) 모델이 가져온 해결안에 대해 평가를 하는 방식이다.

기존에 데이터와 해결안을 미리 프로그래머가 정의해두고, 새로운 데이터가 들어오면 코드를 돌리기만 하던 방식과는 다르다. 해답이 미리 주어져야 한다.

머신러닝의 역사를 뼈대만 살펴보면 아래와 같다.

• 확률적 모델링 시작: 나이브베이즈(베이즈 정리 활용), 로지스틱 회귀(분류 모델)
• 초창기 신경망 제안: LeNet
• 커널 방법: 서포트 벡터 머신(SVM) -> 특성 공학(유용한 특성만 추출)
• 결정트리, 랜덤포레스트, 그레디언트 부스팅 머신
• 발전한 신경망(딥러닝)

정말 키워드만 나열해봤다. 책에서는 더 자세히 나와있지만 나의 관심사는 딥러닝이니 저 부분은 나중에 공부하기로 마음먹고 넘어갔다.

커널 방법

커널 방법만 간단하게 소개하고 넘어가겠다. 분류 문제를 간단하게 만들기 위해서는 데이터를 고차원 표현으로 매핑하는 기법이 이론상으로는 좋아 보이지만 실제로는 컴퓨터로 구현하기 어려운 경우가 많다. 그래서 커널 기법이 등장했다. 새로운 공간에서의 두 데이터 포인트 사이의 거리를 계산만 할 수 있다면, 굳이 새로운 공간의 좌표 자체를 구할 필요가 없다. 커널 함수는 두 포인트 사이의 거리를 효율적으로 계산하는 함수다. (나중에 공부 ^^)

딥러닝은 신경망 레이어를 깊게(Deep) 연결한 머신러닝 모델이다. 위 그림에서 보면 딥러닝에서는 특징 추출 부분이 분류와 합쳐졌다. 서포트 벡터 머신은 좋은 성능을 가졌지만 올바른 특성을 일일이 떠먹여줘야하는 수고가 있었다. 예를 들면 손글씨 숫자를 분류할 때 원시 픽셀 값을 사용할 수는 없다. 픽셀 히스토그램처럼 문제를 쉽게 만드는 유용한 표현을 수동으로 먼저 찾아야한다. 이런 작업을 특성 공학이라고 한다. 딥러닝은 이런 특성을 알아서 찾아낸다. 책에서 마음에 들었던 비유가 있다.

빨간색 종이와 파란색 종이 두개를 마구 구겨서 공으로 만들어냈다고 해보자. 이 공을 다시 빨간색과 파란색 종이로 다시 분리하려면, 조금씩 종이를 살살 펼칠 것이다. 아무리 꼬여있는 데이터라 하더라도 거기서 최대한 올바른 정보를 추출해낸다. 물론 그럼에도 아무런 의미가 없다면 Garbage-In-Grabage-Out일 것이다.

이런 딥러닝이 발전할 수 있는 이유라면 3가지다.

• 하드웨어의 발전
• 데이터의 축적
• 알고리즘: 병렬 연산, 활성화 함수, 가중치 초기화, 옵티마이저 등

신경망의 수학적 이해

물론 여기서 수식을 엄청나게 이해하지는 못했다. 간단하게 훑어본 기분으로 넘어가보자. 우선 케라스로 구현한 간단한 신경망 코드를 살펴보자.

from tensorflow import keras
from tensorflow.keras import layers
from tensorflow.keras.datasets import mnist # 기본적인 손글씨 판별 데이터셋 MNIST import

model = keras.Sequential([
    layers.Dense(512, activation="relu") # relu로 비선형 함수 만들기, 512차원 layer 하나
    layers.Dense(10, activation="softmax") # softmax로 10개의 class에 각각 확률 값 계산하기 -> 0~9까지 숫자 예측
])

model.compile(
    optimizer="rmsprop", # 최적화함수 (역전파 기법)
    loss = "sparse_categorical_crossentropy", # 손실함수
    metrics = ["accuracy"] # 정확도 지표만 고려해서 훈련과 테스트 과정 모니터링
)

train_images = train_images.reshape((60000, 28 * 28))
train_images = train_images.astype("float32") / 255
test_images = test_images.reshape((60000, 28 * 28))
test_images = test_images.astype("float32") / 255

model.fit(train_images, train_labels, epochs=5, batch_size=128)

일단 완성체를 한번 지르고 하나씩 코드를 뜯어보자.

텐서 연산

keras.layers.Dense(512, activation="relu")

for output in range(512):
    output = relu(dot(W, input) + b)

위 두 코드는 같은 결과를 갖는다. 딥러닝에서 위와 같은 layer를 겹겹이 쌓아 수천만 그 이상의 output 계산이 생긴다. 이를 선형으로 하나씩 계산하면 시간이 많이 걸린다. 병렬 계산을 위해 Numpy가 필요하다. Numpy는 BLAS(Basic Linear Algebra Subprogram)을 매우 효율적이고 저수준의 텐서 조작으로 구현해놓았다. 포트란이나 C언어로 구현되어있다. 케라스는 넘파이 기반으로 병렬 연산을 지원한다.

브로드 캐스팅

큰 텐서와 작은 텐서를 연산할 때 작은 텐서의 차원을 큰 텐서에 맞춰줘야 한다. 이 작업을 브로드 캐스팅(broadcasting)이라고 부른다.

X -> (32, 10) 행렬
y -> (10, ) 벡터

위 두 텐서를 연산하려면 두 가지 과정을 거친다.

1. 큰 텐서의 ndim에 맞도록 작은 텐서에 브로딩캐스팅 축이 추가된다.
2. 작은 텐서가 새 축을 따라서 큰 텐서의 크기에 맞도록 반복된다.

코드로 보면 다음과 같다.

y = np,expand_dims(y, axis=0) # y에 브로드캐스팅 축 추가 -> y의 크기가 (1, 10)이 됨
Y = np.concatenate([y] * 32, axis=0) # 축 0을 따라 y를 32번 반복하여 크기가 (32,10)인 Y를 얻음

행렬로 이해하면 행마다 똑같은 열을 32번 복사한 행렬을 만든 것이다. 정말 계산의 편의를 위한 작업이다.

점곱

텐서 곱셈(tensor product) 또는 점곱(dot product)라고 불리는 연산은 가장 널리 사용되고 유용한 텐서 연산이다.

output = relu(dot(W, input) + b)

여기서 dot이 점곱이다. 이름이 점곱인 이름은 수학 기호가 점이다.

z = x ⋅ y

점곱은 어떻게 하는가? 이 책은 참 특이한게 설명이 수식보다는 코드다. 이 글에서도 코드로 간다.

for i in range(x.shape[0]):
    z += x[i] * y[i]

for 문으로 이해하기 좋다. 4차원 텐서는 4중 for문이고 5차원 텐서는 5중 for문일 것이다. x 와 y 두 행렬이 있을 때 이렇게 계산한다. 그림으로 보자.

수식은 이렇다.

텐서의 형태로 보면 아래와 같다. 마지막 차원과 첫번째 차원이 맞물린다.

(a, b, c, d) ⋅ (d, e) = (a, b, c, e)

텐서 크기 변환

x = np.array([
    [0., 1.],
    [2., 3.],
    [4., 5.]
])
x.shape # (3, 2)
x.reshape((2, 3))

>>> #reshape 결과
x = 
[[0., 1., 2.],
[3., 4., 5.]]

텐서의 크기를 변환한다는 것은 특정 크기에 맞게 열과 행을 재배열한다는 뜻이다.

x = np.array([
    [0., 1.],
    [2., 3.],
    [4., 5.]
])
x = np.transpose(x)

>>> #transpose == 전치 결과
x = 
[[[0., 2., 4.],
[1., 3., 5.]]

전치는 행과 열을 바꾸는 연산이다. 그냥 reshape하는 것과 모양이 조금 다르다.

텐서 연산의 기하학적 해석

이동, 회전, 크기변형, 선형변환, 아핀변환, relu 활성화 함수를 사용하는 Dense층 등의 내용이 들어있다.

.. 수식 쓰기가 너무 귀찮았다.

딥러닝의 기하학적 해석은 여기서도 종이 구겼다 펼치는 예시가 적용된다.

경사하강법

신경망의 모든 함수 (변환)은 미분가능한 수식이다. 텐서 함수의 도함수 == 그레디언트(gradient)

y_pred = dot(W, x) # W = 가중치, x = 입력 데이터
loss_value = loss(y_pred, y_true) # y_pred는 예측 값, y_true는 라벨

위 수식에서 x, y_true는 상수다. 그러므로 다음 수식은

loss_value = f(w)

라고 할 수 있다. 여기서는 도함수를

grad(loss_value, W)

로 표현한다.

연쇄 법칙 (chain rule)

자세한 설명은 생략한다..ㅎ

계산 그래프를 활용한 자동 미분

계산 그래프로 표현하고 이를 분산(병렬)처리할 수 있다는 개념을 알아갔다.
계산 과정을 도표로 작성 및 동일 계산을 "분산"해서 병렬 처리하는 코드? tensorflow에 탑재되어있다. GradientTape라고 부른다. 나중에 공부해볼 수도 있겠다.

계산 그래프를 활용하니 순방향, 역전파 설명이 더 잘 되었다. 나중에 설명할 일 생기면 써먹자.

끝

도커는 이미지와 컨테이너 형태로 작동하기 때문에 환경에 구애받지 않고 쉽게 공유가 가능하다. Docker File을 통하여 현재 작업물을 이미지로 빌드하고, 도커 허브에 업로드하여 사람들과 공유할 수 있다. 그 과정을 이 글에서 실습한다.

1. 도커 파일을 작성한다.

도커 파일은 이미지를 빌드하기 위한 명세서이다. 이미지의 베이스가 되는 이미지 레이어(Python버전 등), 환경 변수, 필요한 파일들을 어떻게 조합하여 새로운 이미지를 만들어낼지 작성하는 레시피다. 일반 메모장이나 선호하는 텍스트편집기에서 내용을 작성하고 파일 이름을 ‘dockerfile’로 만든다. 아래는 책 “도커 교과서”에 나오는 도커 파일 예제이다.

FROM diamol/node

ENV TARGET=“blog.sixeyed.com”
ENV METHOD=“HEAD”
ENV INTERVAL=“3000”

WORKDIR /web-ping
COPY app.js .

CMD [“node”, “/web-ping/app.js”]

자주쓰는 도커 명령어

• FROM: 모든 이미지는 다른 이미지로부터 출발한다. Python, Node, Ubuntu 등 다양한 시작 이미지 레이어를 지정한다.

• ENV: 환경 변수 값을 지정하기 위한 인스트럭션이다. 값을 지정하기 위해 [key]=“[value]”형식을 따른다.

• WORKDIR: 컨테이너 이미지 파일 시스템에 디렉터리를 만들고, 해당 디렉터리를 작업 디렉터리로 지정하는 인스트럭션이다. 리눅스는 “/temp”와 같은 방식으로 폴더를 만들고, 윈도 컨테이너에서는 “C:\temp”와 같은 방식으로 폴더를 만든다.

• COPY: 로컬 파일 시스템의 파일 혹은 디렉터리를 컨테이너 이미지로 복사하는 인스트럭션이다. [원본경로] [복사경로] 형식으로 지정하면 된다. 예시 스크립트에서는 로컬 파일 시스템에 있는 app.js 파일을 이미지의 작업 디렉터리로 복사했다.

• RUN: 쉘(shell)에서 커맨드를 실행하는 것처럼 이미지 빌드 과정에서 필요한 커맨드를 실행하기 위해 사용됩니다.

• ENTRYPOINT: 이미지를 컨테이너로 띄울 때 항상 실행되야 하는 커맨드를 지정할 때 사용합니다. Docker 이미지를 마치 하나의 실행 파일처럼 사용할 때 유용합니다. 왜냐하면 컨테이너가 뜰 때 ENTRYPOINT 명령문으로 지정된 커맨드가 실행되고, 이 커맨드로 실행된 프로세스가 죽을 때, 컨테이너를 따라서 종료되기 때문입니다.

• CMD: 해당 이미지를 컨테이너로 듸울 때 디폴트로 실행할 커맨드나, ENTRYPOINT명령문으로 지정된 커맨드에 디폴트로 넘길 파라미터를 지정할 때 사용합니다. CMD 명령문은 많은 경우, ENTRYPOINT 명령문과 함께 사용하게 되는데, ENTRYPOINT 명령문으로는 커맨드를 지정하고, CMD 명령문으로 디폴트 파리미터를 지정해주면 매우 유연하게 이미지를 실행할 수 있게 됩니다.

도표 정리

2. 도커를 연다.

(Ubuntu) DockerFile이 있는 디렉토리로 이동하여 커맨드라인에서 도커를 실행한다. (Window) DockerFile이 있는 디렉토리로 이동하여 cmd창을 켠다.

3. 도커에서 이미지를 빌드한다.

> docker image build –t test .

docker image build로 도커파일에서 이미지를 빌드한다. -t는 image tag를 test로 지정한다는 뜻이다. 마지막 상대경로 ‘ . ’은 현재 디렉토리에서 DockerFile을 찾아서 빌드한다는 뜻이다.

4. 도커 허브에 이미지를 업로드한다.

우선 도커 허브에 가입해야한다. 가입 후 ‘유저이름’과 ‘비밀번호’를 잘 기억해둔다. 로컬 cmd에서 도커에 로그인한다.

> docker login —username [유저 이름]
> [비밀번호]

이제 내 계정으로 도커 허브에 이미지를 업로드(Push)할 수 있다. 우선 로컬에서 빌드한 이미지의 이름을 도커허브 형식에 맞게 변경한다.

> docker image tag test-image [dockerID/image-name:tag]

그리고 이미지를 도커허브에 업로드한다.

> docker image push [dockerID/image-name:tag]

도커 허브 웹페이지에서 내 이미지가 업로드 된 것을 확인할 수 있다.

5. 도커 허브에서 이미지를 다운받는다.

이제 다른 컴퓨터에서도 도커 허브 접근 권한만 있다면 자유롭게 이미지를 다운받아 바로 파일을 실행할 수 있다.

> docker container run [dockerID/image-name:tag]

도커를 쓰는 이유

도커의 최고 장점은 간편한 환경 설정이다.

도커를 사용하면 15년 전에 작성된 Java코드와 1년전에 작성된 Go코드를 하나의 컴퓨터에서 돌릴 수 있다. 코드의 배경이 되는 Library Dependency를 컨테이너별로 설정할 수 있기 때문이다. 라이브러리의 버전 충돌때문에 낭비되는 시간을 줄일 수 있다.

규모가 있는 프로젝트를 인수인계할 때 특히 빛을 발한다. 인수인계를 하면 대여섯 가지나 되는 도구를 특정한 버전으로 설치해야 하고 운영 팀도 또 다른 도구 대여섯 가지를 설치해야 하는 경우가 다반사다. 도커를 도입하면서 모든 툴체인이 중앙집중화돼 어떤 역할의 조직에서도 이런 환경설정 작업이 간단해졌다.

도커와 Virtual Machine(VM)의 차이점

VM도 도커와 마찬가지로 서로 다른 환경에서 동작하는 코드들을 하나의 컴퓨터에서 실행시키기 위한 툴이다. 도커와 VM의 차이를 살펴보자.

VM은 컴퓨터 자원을 Guest OS로 분할하여 컴퓨터 자원을 사용한다. Guest OS간에는 어떠한 간섭도 없이 별개의 컴퓨터처럼 사용된다. OS위에 OS가 올라가기 때문에 비교적 무겁다. Docker는 Guest OS를 만들지 않는다. 그래서 같은 컴퓨터 자원으로 5배 정도 많은 어플리케이션을 실행할 수 있다.

도커의 구조는 Linux 커널로 도커 Container들을 관리하는 방식이다. 리눅스 커널 기능으로 각 컨테이너들은 격리된다.

하나의 컨테이너는 OS처럼 고유의 환경변수, IP주소를 갖는다. 컨테이너 내부에서 사용되는 환경변수, IP주소는 도커가 임의로 부여한 가상의 값들이다. 그래서 Host(컴퓨터)의 환경변수, IP주소와 호환되지 않는다.

도커의 이미지와 레이어, 컨테이너

이미지(image) = 정적인 상태

서버 프로그램, 소스코드 및 라이브러리, 컴파일된 모든 실행 파일을 묶어둔 Read-Only 파일이다. 보통 용량은 수백MB ~ 수GB으로 VM의 이미지보다는 작다. 파일들의 상태를 저장해둔 패키지라 이해하면 된다. Ubuntu이미지는 Ubuntu를 실행하기 위한 모든 파일을 가지고 있으며, Oracle이미지는 Oracle을 실행하는데 필요한 파일과 실행명령어, port정보를 가지고 있다. Ubuntu, Oracle 같은 유명한 이미지는 도커 허브 등의 공식 레지스트리에서 쉽게 다운받아 사용할 수 있다.

레이어(layer)

이미지는 팬케이크처럼 여러겹의 레이어로 이루어진다. 기존 이미지(레이어들)에 파일 하나를 추가하기 위해서 전체 이미지를 다시 빌드하지 않고, 추가 파일 레이어 하나만 쌓아서 이미지를 만든다. 레이어는 읽기 전용(read only)이므로 각각의 레이어를 수정할 수 없다. 새로운 레이어를 쌓기만 할 수 있다. 도커 파일(Docker File)은 레이어를 어떤 순서로 쌓아서 이미지를 만들지 적는 파일이다. DockerHub 및 개인 레지스트리에서 이미지를 공유할 때 바뀐 부분만 주고받을 수 있다.

도커 컨테이너(Docker container) = 동적인 상태

이미지(Image)의 구조를 가져와 실행한 상태이다. 이미지는 Read-Only 파일로 직접 코드를 작성할 수 없다. 이미지의 레시피를 그대로 요리해 실제 음식으로 만든게 컨테이너다. “Docker conatiner run [이미지 이름]”을 통해 실행한다.

도커 파일(Docker File)

도커 이미지를 자동으로 빌드하기 위한 파일이다. 이 파일에서 라이브러리의 버전 충돌을 제어하여 쉽게 이미지를 빌드할 수 있다.

컨테이너의 구성과 볼륨

이미지에서 컨테이너를 구성할 때 이미지의 레시피를 가져와 컨테이너를 만든다고 했다. 이때 이미지의 내용을 복사하는 것이 아니라 Read-Only 형태로 참조하는 방식을 사용한다. 그래서 똑같은 내용으로 용량을 두번 차지하는 비효율을 없엤다.

컨테이너에서 수정되는 파일은 Container layer에서 이루어지고, 원래 이미지에서 수정할 내용이 있다면 Container layer로 가져와 수정한다. 이미지 레이어 자체는 건드리지 않는다.

컨테이너의 생명주기가 다해서 Exit되면 Container layer도 사라진다. 만약 DB와 같은 중요 정보를 컨테이너에서 수정하다가, DB의 이미지가 업데이트되어 새로운 컨테이너로 옮길 때 기존의 수정된 정보가 손실된다면 대참사가 일어날 것이다. 이를 방지하기 위해 도커 볼륨(Docker Volume)이 필요하다.

도커 볼륨은 이미지, 컨테이너와 다른 주기를 갖는 디렉토리이다. 도커 버전의 USB라고 생각하면 된다. 이 글에서는 내용이 너무 길어지므로 개념만 설명하고 넘어가겠다.

도커 사용하기 (Container의 5가지 상태)

• Created: Image의 상태를 받아 Container를 생성한 상태이다.
• Running: Container에 실제 컴퓨터 자원을 할당하여 프로그램을 실행시킨 상태이다.
• Paused: Container에 할당된 컴퓨터 자원을 빼앗지 않으나 잠깐 프로그램의 실행을 멈춘 상태이다. 다시 Running상태로 돌아가면 프로그램을 그대로 실행할 수 있다. (Window 절전모드)
• Exited: Container에 할당된 컴퓨터 자원을 빼앗은 상태이다 (Memory Free). 다시 Running 상태로 돌아가도 실행하던 프로그램을 이어서 실행 할 수 없다. (Window 전원 끈 상태)
• Deleted: Container 자체가 삭제된 상태이다.

도커 실습하기

도커 기본 명령어 형식

> docker [container/image/volume...] [명령어]

1) 컨테이너 실행하기

> docker run [OPTIONS] [IMAGE:TAG]

run 명령어는 image를 컨테이너로 실행시킨다. 이미지가 로컬 registry에 없을 경우, 자동으로 docker hub를 통해 image를 pull 받아온 뒤 실행시킨다.

예시 실행해보기

> docker run ubuntu:18.04

> docker ps -a

ps는 실행중인 컨테이너를 확인하는 명령어이다. -a는 모든 상태의 컨테이너를 확인한다.

2) 컨테이너 Pause하기

> docker stop [CONTAINER ID]

docker ps의 CONTAINER ID를 확인하여 ID의 앞 4자리정도만 입력하면 알아서 식별하여 컨테이너를 stop한다.

3) Pause된 컨테이너 삭제하기

> docker rm [CONTAINER ID]

stop과 동일하다.

4) 실행중인 컨테이너 Exit

> docker kill [CONTAINER ID]

ㅇㅇ

5) Docker File로 image 만들기

도커를 쓰면 우분투나 파이썬 하나만 있는 이미지를 쓰지 않는다. 도커파일로 모든 레이어를 기술하여 한번에 빌드할 수 있다. (도커파일 사용법 알아보기)

> docker build –t [Docker File]

docker file을 빌드할 수 있다.

빌드를 완료하면 위처럼 image가 생성된다.

6) 컨테이너 이미지로 만들기

이미지로 만들기 전에 컨테이너를 Pause(멈춤) 상태로 만든다.

> docker stop [컨테이너 이름 또는 아이디]

컨테이너의 이름 또는 아이디를 알아낸다.

> docker ps –a
또는
> docker container ls -all

컨테이너의 아이디를 넣어서 이미지로 만든다.

> docker commit container_id image_name:tag

끝!

(참고자료) 도커 교과서, 엘튼 스톤맨, 링크

기본 설치

(선택사항) root 계정에 접속한다.

> su
> [password 입력]

다운로드에 필요한 패키지들을 설치한다. (ca-certificates, curl, gnupg, lsb_release)

>  sudo apt-get update
>  sudo apt-get install ca-certificates curl gnupg

docker download script를 docker 홈페이지에서 다운 받아 실행한다.

>  curl -fsSL https://get.docker.com -o get-docker.sh
>  sudo sh get-docker.sh

설치가 되었는지 확인하기 위해 hello-world:image를 실행시켜본다.

>  sudo docker run hello-world

“”“ Unable to find image 'hello-world:latest' locally latest: Pulling from library/hello-world 2db29710123e: Pull complete Digest: sha256:--- Status: Downloaded newer image for hello-world:latest Hello from Docker! This message shows that your installation appears to be working correctly. To generate this message, Docker took the following steps:

1. The Docker client contacted the Docker daemon.
2. The Docker daemon pulled the "hello-world" image from the Docker Hub. (amd64)
3. The Docker daemon created a new container from that image which runs the executable that produces the output you are currently reading.
4. The Docker daemon streamed that output to the Docker client, which sent it to your terminal. “”“

도커가 잘 실행됐다면 위 메시지가 나온다. 실행이 안됐다면 아래 방법들을 시도해서 해결할 수 있다.

에러 해결하기

Hyper-v 사용권한 설정하기

Hyper-v를 사용할 수 있도록 설정한다.

>  bcdedit /set hypervisorlaunchtype auto

BIOS에서 VM 사용권한 설정하기

그리고 다음 과정을 통해 BIOS 환경에서 VM 사용 제한을 해제한다.

1. BIOS에서 VM을 설정하기 위해 컴퓨터 리부팅한다.
2. 리부팅 과정(검정색 화면)에서 F2와 또는 Delete 키를 연타한다. (연타 방법 동영상)
3. 이후 BIOS 설정에서 advanced mode에서 Intel Virtualization Technology 설정을 enable로 변경한다. (과정 동영상)

다시 컴퓨터를 키면 Docker를 실행할 수 있다.

>  docker version

리눅스 관리자 권한 설정하기

일반 사용자가 관리자 권한이 없어 sudo 명령어로만 docker가 실행된다면 사용자를 Docker 그룹에 추가할 수 있다.

>  sudo usermod -aG docker $USER

AIhub에 있는 '이상행동 CCTV 동영상 데이터셋'에서 이상행동을 하는 부분만 잘라내어 학습 데이터로 쓰려고 합니다.

1. 데이터 살펴보기

• 데이터 종류와 크기

• 라벨 (XML파일) 라벨 <action>안에 <actionname>과 <frame>이 필요한 정보입니다. 영상 안에서 Kick이나 punch등 다양한 이상행동들이 나와요. 우리는 영상에서 이상행동이 나오는 부분을 잘라 행동 종류별로 각각 폴더에 담을겁니다.

2. FFmpeg 간단 소개

FFmpeg는 영상 처리를 위한 라이브러리입니다. 자세하게 알고 싶다면 여기 wikidocs로 들어가 보세요. 저는 대략 이해하는데 도움이 됐어요.

!ffmpeg -i {input_file} -vf scale=3840x2160 
-vf trim=start_frame={start_frame}:end_frame={end_frame} {output_file}

보통 cmd창에서 영상을 편집하는데 저는 python의 매직커맨드 기능을 사용해서 cmd 명령어를 python에서 사용했어요. 앞에 '!' 를 붙이면 python에서 cmd 명령어를 사용할 수 있어요.

코드를 설명은 아래 있어요.

• -i 뒤에 편집할 동영상 경로 입력
• vf scale= 뒤에 화면 사이즈 입력 (영상 속성에서 확인 가능)
• vf trim= 뒤에 시작 frame과 끝 frame을 입력
• output_file칸에 저장할 파일 이름 입력 (확장자까지 입력)

다른 포스트 보니까 -ss 같은 옵션을 쓰면 초단위로 설정해서 자를 수도 있다고 하네요.

!ffplay -i {file_path} -video_size 3840x2160 -loop 0

편집한 영상을 코드로 실행하려면 위 코드를 사용하시면 됩니다.

3. Python에서 XML파일 읽기

우선 파일들을 가져옵시다.

import os

data_base_folder = 'C:\\Users\\luna_f1\\Desktop\\aihub_video_edit'
video_path = data_base_folder + '\\videos'
label_path = data_base_folder + '\\labels'
save_folder = data_base_folder + '\\save'

video_file_list = os.listdir(video_path)
label_file_list = os.listdir(label_path)

video_file_list.sort()
label_file_list.sort()

print('img len: {}, label len: {}\n'.format(len(video_file_list), len(label_file_list)))
print(video_file_list[:10])
print(label_file_list[:10])


## 출력 예시
## img len: 1, label len: 1
## ['488-3_cam03_vandalism01_place09_day_winter.mp4']
## ['488-3_cam03_vandalism01_place09_day_winter.xml']

label_file_list에는 xml파일들이, video_file_list에는 영상들 파일 이름이 들어있어요.

그리고 먼저 영상 편집을 할 함수를 설계해봅시다.

def cut_video(start_frame, end_frame, input_file, output_file):
    !ffmpeg -i {input_file} -vf scale=3840x2160 
    -vf trim=start_frame={start_frame}:end_frame={end_frame} {output_file}
    return

함수에서 시작프레임, 끝프레임, 입력 파일, 출력 파일을 매개변수로 받아 편집하도록 만들었어요.

그러면 xml파일을 읽어서 함수를 호출하기만 하면 됩니다. 아래는 그 코드에요.

import xml.etree.ElementTree as ET
import os

# 영상과 xml 라벨을 짝꿍을 맞춰 가져옴
for video_file, label_file in zip(video_file_list, label_file_list):
    video_file_path = video_path + '\\' + video_file
    label_file_path = label_path + '\\' + label_file
    label = ET.parse(label_file_path).getroot()

   # 잘라낼 영상 부분을 actionname에 따라 분류함.
    highlights = { }
    for clss in label.iter("actionname"):
        clss = clss.text
        highlights[clss] = []

        # 영상들을 저장할 폴더를 actionname별로 각각 만듦
        try:
            os.mkdir(save_folder + '\\' + clss)
        except:
            pass

    # 영상 안의 이상행동들의 시작프레임, 끝프레임을 리스트로 묶어 이상행동 종류별로 dict에 저장함.
    # 구조: highlights['punching'] = [[start, end], [start, end]...]
    for action in label.iter("action"):
        hlight_class = action.find('actionname').text
        for start, end in zip(action.iter('start'), action.iter('end')):
            highlights[hlight_class].append([start.text, end.text])

    # 이상행동 부분들을 잘라서 now_save_folder_path에 저장함
    for clss in highlights.keys():
        now_save_folder_path = save_folder + '\\' + clss
        #video_file_path = video_path + '\\' + video_file
        for [start, end] in highlights[clss]:
            cut_video(start, end, video_file_path, now_save_folder_path + '\\' + start + end + '.yuv')

OpenCV 뻘짓 기록

이미지 데이터에서 OpenCV를 써서 동영상 자르는것도 이걸로 하려고 했는데 생각보다 뻘짓이 길어졌네요. 내가 실력이 모자라서 그런거 같습니. 혹시 누군가에게 도움이 될지 모르니 실패한 과정을 적어놓겠습니다.

######### 동영상 처리는 OpenCV보다는FFmpeg를 쓰자!


import cv2
import numpy as np
import xml.etree.ElementTree as ET

for video_file, label_file in zip(video_file_list, label_file_list):
    video_file_path = video_path + '\\' + video_file
    label_file_path = label_path + '\\' + label_file

    cap = cv2.VideoCapture(video_file_path)
    label = ET.parse(label_file_path).getroot()

   # 잘라낼 영상 부분을 punching, kicking, pushing 3가지로 분류함. (각각 여러개 있을 수 있음)
    highlight = {
        'punching' : [],
        'kicking' : [],
        'pushing' : []
    }
    # 구조: highlight['punching'] = [[start, end], [start, end]...]
    for action in label.iter("action"):
        hlight_class = action.find('actionname').text
        for start, end in zip(action.iter('start'), action.iter('end')):
            highlight[hlight_class].append([int(start.text), int(end.text)])

    #재생할 파일의 넓이와 높이
    width = cap.get(cv2.CAP_PROP_FRAME_WIDTH)
    height = cap.get(cv2.CAP_PROP_FRAME_HEIGHT)
    fourcc = cv2.VideoWriter_fourcc(*'DIVX')

    # 영상을 여러개 출력해야하므로 영상 하나마다 out writer 하나를 지정함.
    outs = {
        'punching' : [],
        'kicking' : [],
        'pushing' : []
    }
    for hlight_class in highlight.keys():
        for i in range(len(highlight[hlight_class])):
            outs[hlight_class].append(cv2.VideoWriter(save_folder + "\\output_"+ hlight_class + str(i) + '_'+ video_file, fourcc, 30.0, (int(width), int(height))))

    while cap.isOpened():
        success, image = cap.read()
        if not success:
            print("카메라를 찾을 수 없습니다.")
            break

        now_frame = int(cap.get(cv2.CAP_PROP_POS_FRAMES))
        if not now_frame % 20 == 0: continue

        # 필요에 따라 성능 향상을 위해 이미지 작성을 불가능함으로 기본 설정합니다.
        image.flags.writeable = False
        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)

        # 비디오 저장 파트입니다.
        cv2.imshow('video_show', image)
        for hlight_class in highlight.keys():
            for i in range(len(outs[hlight_class])):
                if now_frame >= highlight[hlight_class][i][0] and now_frame <= highlight[hlight_class][i][1]: # event frame 값이면
                    outs[hlight_class][i].write(image) # 이 부분이 write하는 부분.
        if cv2.waitKey(2) & 0xFF == 27:
            break

    cap.release()
    for key in outs.keys():
        for out in outs[key]:
            out.release()
    cv2.destroyAllWindows()

영상을 쭉 진행하다가 현재 프레임이 이상행동 프레임과 일치하면 동영상으로 저장하는 식으로 설계했습니다. MediaPipe에서 쓰던 코드를 재활용하다 이렇게 비효율적을 짠것같습니다. 여러모로 고칠점이 많아서 배울것도 많았어요.

• MediaPipe를 사용하여 영상 속 사람의 특징점을 점으로 인식하고 그린다.
• cv2를 활용해 동영상을 불러들이고 쓸 수 있다.

1. MediaPipe 설치

MediaPipe는 Google에서 만든 사람 인식 Library이다. 포즈 뿐만 아니라 다양한 카테고리를 인식할 수 있다.

# 다운로드 코드
!pip install mediapipe

홈페이지에 자세하게 MediaPipe에 대한 설명이 나와있지만 영어다. 한국어로 간단하게 설명해보겠다.

간단 MediaPipe 설명

MediaPipe는 아래와 같은 사람의 특징점들을 인식하는 머신러닝 라이브러리다.

동영상의 경우 각 프레임별로 아래의 프로세스가 수행된다.

1. 가장 중심이 되는 사람을 감지(detect)한다.
2. 그 사람의 특징점을 찾는다. 여기서 사람 하나를 먼저 감지하는 단계가 있다. 이 부분을 수정해 여러 사람을 감지하게 만들면 한 프레임 안의 여러사람에게서 특징점을 도출할 수 있다. 그만큼 시간이 오래 걸리겠지만.

2. 코드

MediaPipe 홈페이지에 사용 코드가 나와있다. 하지만 동영상을 돌리려면 코드를 약간 수정할 필요가 있다. 아래는 내가 수정한 코드다. 설명은 주석으로 달아놓았다.

import cv2
import mediapipe as mp
import numpy as np
mp_drawing = mp.solutions.drawing_utils
mp_drawing_styles = mp.solutions.drawing_styles
mp_pose = mp.solutions.pose

#파일 위치 미리 지정
input_video_path = "/content/in_theft.mp4"
save_video_path = '/content/in_theft_output.mp4'

cap = cv2.VideoCapture(input_video_path)

#재생할 파일의 넓이와 높이
width = cap.get(cv2.CAP_PROP_FRAME_WIDTH)
height = cap.get(cv2.CAP_PROP_FRAME_HEIGHT)
#video controller
fourcc = cv2.VideoWriter_fourcc(*'DIVX')
out = cv2.VideoWriter(save_video_path, fourcc, 30.0, (int(width), int(height)))


with mp_pose.Pose(
        min_detection_confidence=0.7,
        min_tracking_confidence=0.7) as pose:
    while cap.isOpened():
        success, image = cap.read()
        if not success:
            print("카메라를 찾을 수 없습니다.")
            # 웹캠을 불러올 경우는 'continue', 동영상을 불러올 경우 'break'를 사용합니다.
            break

        # 필요에 따라 성능 향상을 위해 이미지 작성을 불가능함으로 기본 설정합니다.
        image.flags.writeable = False
        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        results = pose.process(image)

        # 포즈 주석을 이미지 위에 그립니다.
        image.flags.writeable = True
        image = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)
        mp_drawing.draw_landmarks(
            image,
            results.pose_landmarks,
            mp_pose.POSE_CONNECTIONS,
            landmark_drawing_spec=mp_drawing_styles.get_default_pose_landmarks_style())
        # 보기 편하게 이미지를 좌우 반전합니다.
        cv2.imshow('MediaPipe Pose', image) #코랩에서 돌릴거면 imshow()문은 주석처리할 것.
        out.write(image)
        if cv2.waitKey(5) & 0xFF == 27:
            break
cap.release()
out.release()
cv2.destroyAllWindows()

결과는 아래 사진을 클릭하면 볼 수 있다.

보면 잘못 인식한 부분이 보인다. 사람이 여러명이라 그런듯 하다. 다음에 고쳐보자.!

목표

• 위 사진처럼 detect하는 YOLO모델을 실행시켜본다.

기본적인 detect를 하는 코드는 아래와 같다.

from IPython.display import Image
import os

!python /content/drive/MyDrive/yolov5/detect.py --weights /content/carplate_best.pt --source /content/drive/MyDrive/crop_img/temp/same_data_masking

3개의 파일 주소를 넣어야하는데 차례대로 다음과 같다.

• 욜로를 git에서 클론할때 딸려온 detect.py의 위치
• 학습시킨 pt의 위치
• yolo에 넣을 파일들 위치

참고로 학습시킨 pt의 위치는 yolo > train > 학습시킨 모델이름 > weight에 .pt형태로 저장되어있다.

여기에 아래와 같은 옵션을 넣으면 추가로 좌표를 활용한 이미지를 저장할 수 있다. 여러개 넣어도 폴더가 따로 생성되며 전부 저장된다.

--save-text --save-crop

아래는 코드 예시이다.

from IPython.display import Image
import os

!python /content/drive/MyDrive/yolov5/detect.py --weights /content/carplate_best.pt --source /content/drive/MyDrive/crop_img/temp/same_data_masking --save-crop

from IPython.display import Image
import os

!python /content/drive/MyDrive/yolov5/detect.py --weights /content/carplate_best.pt --source /content/drive/MyDrive/crop_img/temp/same_data_masking --save-text

왼쪽이 yolo 실행 전, 오른쪽이 yolo 실행 후다.

저장된 사진의 위치는 python log 마지막에 나온다.

• 라벨링한 이미지 1000장으로 Crop과 Mask를 한다.
• 클래스 0은 차량 번호판으로, Mask할 것이다.
• 클래스 1, 2는 각각 차량 앞면부와 차량 전체로, Crop할 것이다.

환경은 코랩으로 해보았다.

이미지 업로드와 다운로드하는데 시간이 오래걸리니 로컬(주피터 노트북 등등)에서 돌리는게 낫다.

나는 로컬에 cv2 설정하다가 안돼서 그냥 코랩에 돌렸다. 코랩에서는 imshow()도 안되니 더더욱 로컬에서 돌리자.

1. 구글 드라이브 연결하기

구글 드라이브에 이미지 압축파일을 업로드 한다. 압축 안해서 올리면 엄청 오래걸린다. 압축파일을 올린 후 밑에서 코드로 압축을 해제해준다.

그리고 코랩에 연결해준다. (버튼 클릭 후 생성된 코드 실행)

2. 이미지 리스트 불러오기 & 정렬

import os

base_path = '/content/drive/MyDrive/yolov5/data/car+plate+carfront/'
img_path = base_path + 'images'
label_path = base_path + 'labels'

img_file_list = os.listdir(img_path)
label_file_list = os.listdir(label_path)

나는 하나의 디렉토리에 images와 labels를 분리해서 넣었다. os.listdir로 각각 폴더 위치를 지정해 폴더의 하위 파일 이름을 리스트로 가져온다.

img_file_list.sort()
label_file_list.sort()

print('img len: {}, label len: {}\n'.format(len(img_file_list), len(label_file_list)))
print(img_file_list[:100])
print(label_file_list[:100])

그리고 리스트를 정렬해준다.

정렬 안하면 밑에서 이미지와 라벨 짝이 안맞아 이상 부분을 crop & mask한다.

2_1. 검수코드

## 이미지 파일과 라벨 파일 이름 똑같음
cnt=0

for a, b in zip (img_file_list, label_file_list):
  a = a.rstrip('jpgtxt')
  a = a.rstrip('.')
  b = b.rstrip('jpgtxt')
  b = b.rstrip('.')
  if a != b: print(cnt)
  cnt+=1

• 이미지와 라벨 짝이 맞는지 검사하는 코드다.

아무것도 출력이 안되면 잘 짝이 맞는 것이다. 짝이 안맞을 경우 해당 인덱스를 출력한다.

3. 이미지 좌표 구하기

import cv2

cnt = 0

for img_file, label_file in zip(img_file_list, label_file_list):

    cnt += 1
    print(cnt)


    img = cv2.imread(img_path + '/' + img_file)

    with open(label_path + '/' + label_file, 'r') as f:
        labels = f.readlines()
        car_whole = list(map(float, labels[2].split()[1:]))  ## 차 전체 이미지 바운딩박스
        car_front = list(map(float, labels[1].split()[1:]))  ## 차 앞부분 이미지 바운딩박스
        carplate_masking = list(map(float, labels[0].split()[1:]))  ## 차 앞부분 이미지 바운딩박스
        h, w, _ = img.shape  ## 사진 전체 크기 (heigt, width)

우선 for문에서 이미지 파일과 라벨 파일을 정렬한 순서대로 하나씩 선택한다. cv2.imread로 img에 이미지를 저장한다. with open~ 으로 f에 라벨을 저장한다.

  ## 라벨 예시
  0 0.281785 0.303789 0.193897 0.092345
  1 0.425525 0.287692 0.581912 0.378795
  2 0.455954 0.285394 0.730389 0.509354

• 라벨 설명
  • 라벨 클래스는 0(자동차 번호판), 1(자동차 앞면 그릴), 2(자동차 전체)이다.
  • 뒤에 소수점으로 표시된 좌표는 각각 (바운딩박스 중점 x좌표, 바운딩박스 중점 y좌표, 바운딩박스 가로 넓이, 바운딩박스 세로 넓이) 이다.
  • 전부 소수점으로, 사진 전체 중 비율을 기준으로 라벨링이 되어있다.

각각 list로 변수에 좌표를 저장한다.

3_1. 좌표 진짜로 구하기

원하는 부분을 사각형모양으로 crop하거나 mask하려면 왼쪽 위 꼭짓점 좌표와 오른쪽 아래 꼭짓점 좌표를 알아야한다.

다음은 이를 계산한 좌표다.

starty = car_whole[1] - (car_whole[3] / 2)
        startx = car_whole[0] - (car_whole[2] / 2)
        endy = car_whole[1] + (car_whole[3] / 2)
        endx = car_whole[0] + (car_whole[2] / 2)

그리고 비율이 0과 1을 넘지 않도록 다음 함수를 통과시켜준다.

def size_control(sx, sy, ex, ey):
  if sx < 0: sx = 0
  if sy < 0: sy = 0
  if ex > 1: ex = 1
  if ey > 1: ey = 1
  return sx, sy, ex, ey

합치면 다음과 같다.

def size_control(sx, sy, ex, ey):
  if sx < 0: sx = 0
  if sy < 0: sy = 0
  if ex > 1: ex = 1
  if ey > 1: ey = 1
  return sx, sy, ex, ey


starty = car_whole[1] - (car_whole[3] / 2)
        startx = car_whole[0] - (car_whole[2] / 2)
        endy = car_whole[1] + (car_whole[3] / 2)
        endx = car_whole[0] + (car_whole[2] / 2)

        startx, starty, endx, endy = size_control(startx, starty, endx, endy)

4. 이미지 Crop하기 & 저장

이미지 크롭은 슬라이싱으로 간단하게 할 수 있다. 이미지는 numpy array로 되어있어 슬라이싱이 가능하다.

        whole_crop_img = img[int(starty * h):int(endy * h), int(startx * w):int(endx * w)]

int 안붙이면 슬라이싱이 안먹힌다. 라벨링이 비율로 되어있으므로 h와 w를 곱해준다.

cv2.imwrite('/content/drive/MyDrive/crop_img/car_whole/' + 'whole_crop_' + img_file, whole_crop_img)

그리고 저장해준다.

5. 이미지 mask하기

이미지 마스크는 원하는 영역에 속이 꽉찬 사각형을 그려서 할 수 있다.

masking_img = img
        cv2.rectangle(masking_img, (int(startx * w), int(starty * h)), (int(endx * w), int(endy * h)),(255,255,255), -1)

뒤에 (255,255,255)는 하얀색을 가리킨다. (0,0,0)으로 하면 까만색 사각형을 그린다.

cv2.imwrite('/content/drive/MyDrive/crop_img/carplate_masking/' + 'carplate_masking_' + img_file, masking_img)

그리고 저장

crop & mask 전체 코드

import cv2

cnt = 0

for img_file, label_file in zip(img_file_list, label_file_list):

    cnt += 1
    print(cnt)


    img = cv2.imread(img_path + '/' + img_file)

    with open(label_path + '/' + label_file, 'r') as f:
        labels = f.readlines()
        car_whole = list(map(float, labels[2].split()[1:]))  ## 차 전체 이미지 바운딩박스
        car_front = list(map(float, labels[1].split()[1:]))  ## 차 앞부분 이미지 바운딩박스
        carplate_masking = list(map(float, labels[0].split()[1:]))  ## 차 앞부분 이미지 바운딩박스
        h, w, _ = img.shape  ## 사진 전체 크기 (heigt, width)

        ### car_whole
        starty = car_whole[1] - (car_whole[3] / 2)
        startx = car_whole[0] - (car_whole[2] / 2)
        endy = car_whole[1] + (car_whole[3] / 2)
        endx = car_whole[0] + (car_whole[2] / 2)
        startx, starty, endx, endy = size_control(startx, starty, endx, endy)


        whole_crop_img = img[int(starty * h):int(endy * h), int(startx * w):int(endx * w)]
        try:
          cv2.imwrite('/content/drive/MyDrive/crop_img/car_whole/' + 'whole_crop_' + img_file, whole_crop_img)
        except:
          pass


        ### car_front
        starty = car_front[1] - (car_front[3] / 2)
        startx = car_front[0] - (car_front[2] / 2)
        endy = car_front[1] + (car_front[3] / 2)
        endx = car_front[0] + (car_front[2] / 2)
        startx, starty, endx, endy = size_control(startx, starty, endx, endy)

        carfront_crop_img = img[int(starty * h):int(endy * h), int(startx * w):int(endx * w)]
        try:
          cv2.imwrite('/content/drive/MyDrive/crop_img/car_front/' + 'carfront_crop_' + img_file, carfront_crop_img)
        except:
          pass


        ## carplate_masking
        starty = carplate_masking[1] - (carplate_masking[3] / 2)
        startx = carplate_masking[0] - (carplate_masking[2] / 2)
        endy = carplate_masking[1] + (carplate_masking[3] / 2)
        endx = carplate_masking[0] + (carplate_masking[2] / 2)

        masking_img = img
        cv2.rectangle(masking_img, (int(startx * w), int(starty * h)), (int(endx * w), int(endy * h)),(255,255,255), -1)
        cv2.imwrite('/content/drive/MyDrive/crop_img/carplate_masking/' + 'carplate_masking_' + img_file, masking_img)

        f.close()

욜로 사용해보는 post는 다음 포스트에서

from google.colab import drive
drive.mount('/content/drive')

• 구글 드라이브를 연결합니다.
• 저 버튼 누르면 자동으로 코드가 생깁니다.

2. 파일 업로드 및 압축해제

• 구글 드라이브에 들어가서 이미지와 라벨 압축 파일을 업로드합니다.
• 그냥 올려도 괜찮지만, 시간이 오래걸려서 압축파일로 업로드합니다.

# 압축파일들 풀 폴더 지정 (폴더 새로 안생기고 파일 바로 풀림 주의)
%cd /content/drive/MyDrive/yolov5/data/images   
!unzip -qq "/content/drive/MyDrive/temp/images.zip"

%cd /content/drive/MyDrive/yolov5/data/labels
!unzip -qq "/content/drive/MyDrive/temp/labels.zip"

• 코드로 압축해제를 합니다.
• %cd로 이동한 경로에 압축 파일이 풀립니다. (폴더가 따로 안생기니 미리 만든 폴더로 이동하세요.)
• unzip -qq 뒤에 압축 해제할 파일 경로를 지정합니다.

잘못 풀린 파일 삭제하기

import os
import glob

[os.remove(f) for f in glob.glob('/content/drive/MyDrive/yolov5/data/*.jpg')] #jpg파일
[os.remove(f) for f in glob.glob('/content/drive/MyDrive/yolov5/data/*.txt')] #txt파일

• 압축해제를 잘못한 경우 이 코드를 사용하면 됩니다.
• 경로상의 jpg, txt 파일을 모두 삭제합니다. (데이터셋이 아닌 txt폴더, jpg폴더도 모두 삭제되니 주의하세요.)

파일 개수 확인하기

# 이미지 파일 개수 확인

import glob 

path = '/content/drive/MyDrive/yolov5/data/car+plate+carfront/images_1/*.jpg'
file = glob.glob(path)

print(len(file))

# 라벨 파일 개수 확인

import glob 

path = '/content/drive/MyDrive/yolov5/data/car+plate+carfront/labels_1/*.txt'
file = glob.glob(path)

print(len(file))

• 압축이 잘 풀렸나 코드로 확인해봅니다.

3. YOLOv5 깃 클론하기 & 환경설정

YOLOv5 GitHub 바로가기

# YOLOv5 깃 클론

%cd /content/drive/MyDrive
!git clone https://github.com/ultralytics/yolov5  # clone repo
%cd yolov5
!git reset --hard fbe67e465375231474a2ad80a4389efc77ecff99  # YOLOv5 브랜치 설정

• 깃헙에서 YOLOv5를 클론해옵니다.
• YOLOv5도 계속 업데이트가 되기 때문에 에러를 방지하고자 브랜치를 임의로 설정합니다.

%cd /content/drive/MyDrive/yolov5/
!pip install -r requirements.txt

• requirements.txt에 YOLOv5를 돌리기 위한 라이브러리들이 적혀있습니다.
• 코드를 실행하면 필요한 파일을 전부 다운받습니다.

4. yaml파일 설정

%cat /content/drive/MyDrive/yolov5/data/data.yaml

• yaml파일은 YOLOv5에게 학습시킬 파일과 class 등을 지정해주는 파일입니다.
• 왼쪽 파일창에서 drive > MyDrive > yolov5 > data > data.yaml파일 을 더블클릭합니다.
• 더블클릭하면 오른쪽에 메모장처럼 편집할 수 있는 창이 뜹니다.

• names -> 라벨 이름 설정 (순서대로)
• nc -> number of classes, class 개수
• 밑에는 train, val 이미지 데이터 경로 설정

에러 주의

데이터 저장은 다음 사항을 준수해야합니다.

1. 이미지와 라벨은 이름이 같고 확장자만 달라야함. (img.jpg, img.txt) 2_1. 이미지와 라벨을 같은 폴더에 넣고 그걸 train, val 데이터 경로로 설정한다. 2_2. 같은 디렉토리 하위에 images폴더와 labels폴더를 만들고 그 안에 사진과 라벨을 저장한다(폴더이름 꼭 images, labels이어야함). 그리고 images폴더를 train, val 경로로 설정한다.

5. YOLOv5 학습

%cd /content/drive/MyDrive/yolov5
!python /content/drive/MyDrive/yolov5/train.py --epochs 50 --data /content/drive/MyDrive/yolov5/data/data.yaml --name car+carfront+plate

• 첫번째 경로는 yolov5 하위의 train.py의 경로를 지정한다.
• --epochs는 학습을 몇번 반복할지 결정한다.
• --data는 yaml파일의 경로를 지정한다.
• --name은 완성할 모델의 이름을 지정한다.
• --weights는 시작할 기본 가중치 파일을 설정한다. 없으면 랜덤으로 정해진 값으로 시작한다.

6. 결과 확인하기

%load_ext tensorboard
%tensorboard --logdir /content/drive/MyDrive/yolov5/runs/train/carfront_basic

• --logdir 뒤에 모델의 위치를 지정하면 멋진 그래프가 나온다.