서론 및 주저리 주저리

파이썬에서 제공하는 내장 프레임워크 티키타라는 것이 있는데
winform은 가능하면 c#으로 할 것, mac 만들꺼면 object c로 하던가
qtic? 이것을 C로 할 것 파이썬을 공부하고 나면 numpy, pandas, matplotlib, pytorch, 등을 활용하여 간단한 전이학습 된 모델을 하나 만드는 것이 가능하다. 우리가 만든 모델을 가지고 바로 서비스를 하는 것이 가능하다. 배우면 써먹어보자

Kaggle이란 것이 있다. AI는 사실 확률형 모델이다. 많은 데이터를 기반으로 상황을 판단하고 찍어주는 것이 가능하다.
데이터 셋을 가지고 공유를 하거나 처리를 하는 것이 가능하다. 어떤 종의 꽃밭으로 보니까 모이더라 이 종은 특징이 꽃밭인가 모이는 것을 보는 예제이다.
웹에서는 전통적으로 MySQL. MariaDB를 사용을 했다. postgre는 python에서 많이 쓰던 데이터베이스다.
요즘은 웹에서도 많이 사용한다고 한다. 따로 공부를 할 필요가 있겠다.

rainbow CSV → csv 보는 것 도움주는 플러그인이니 추후에 csv를 사용할 일이 있다면 설치해 주도록 하자.

파일로부터 데이터 받아오기

from langchain_community.document_loaders import CSVLoader
from langchain_core.documents import Document

from typing import List

csv_loader = CSVLoader(
    file_path = 'libs/iris.csv',
    encoding = 'utf-8',
)

csv_docs : List[Document] = csv_loader.load()

print(csv_docs[:5])

csv 파일에서 가져오고 이것을 document 객체로 저장을 하게 된다.

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import CharacterTextSplitter

text_loader = TextLoader(
    file_path='./libs/split_text.txt',
    encoding='utf-8',
)

doc = text_loader.load()[0]

splitter=CharacterTextSplitter(
    separator = '\n',
    chunk_size= 30,
    chunk_overlap=20,
)

chunk_lines = splitter.split_text(doc.page_content)

print(chunk_lines)

아래는 .txt 형태로 파일을 가져오고 chunk로 분리를 해서 객체를 생성하게 된다.
이렇게 떠있는 데이터는 의미단위로 chunking을 하려면 임베딩 과정을 거쳐야 한다. 그리고 이를 저장하고 싶다면 Vector 스토어에 저장을
하게 된다. 이 경우 벡터 데이터 간의 유사도를 검사하고 의미적 유사성으로 build를 찾게된다.

모든 임베딩 결과를 메모리에 계속 올려두는 방식은 단점이 있다. 이 경우 부담이 커져서 vectorDB를 사용하는 것이 적합하다

document 객체

RAG와 같은 경우 어느 파일, 몇 페이지, 작성자 누구?, 작성 문서와 같은 정보도 필요한데 이를 다루기 위해 나온 객체가
document이다. document는 크게 두가지로 이루어져 있다.

metadata
- 문서에 관한 부가 정보 저장
page_content
- 실제 llm이 읽는 텍스트

vectorDB 기준으로 document는 그대로 임베딩이 된다. 이때 메타데이터도 같이 저장되기 때문에 출처를 쉽게 확인하는 것이 가능하다.

LangGraph

Graph

애플리케이션 실행 구조를 정의 및 관리를 시행한다.
여러 개의 node와 edge를 담는 컨테이너 역할을 수행한다.
애플리케이션의 실행경로를 정의한다.
작업들의 순서와 조건을 결정하는 실행구조를 가진다.

StateGraph

상태 스키마를 만들고 상태 그래프를 만들 수 있게 된다. 상태 스키마를 만들고 그래프를 만들 수 있는 빌더가 만들어지게 된다.
이 경우 그래프가 만들어 질 수 있게 된다. 이렇게 그래프 인스턴스는 invoke(), batch(), stream() 이렇게 runnable()의 실행 메서드를 통해 실행할 수 있다.

TypedDict

State의 구조를 정의하는 타입, 모든 node가 하나의 state를 공유하고 각 node가 같은 state를 읽고 수정을 하게 된다.

from typing import TypedDict

class State(TypedDict):
    question: str
    documents: list
    answer: str

Reducer

각각의 노드에서 반환한 값을 state에 어떻게 합칠지 정하는 규칙이다.
LangGraph에서 State가 업데이트 될 때 덮어쓸지, 누적할지, 병합할지를 결정하는 함수

Node

실제 작업을 수행하는 실행 단위
단일 책임으로 구성하게 된다.
State를 입력 받아 State를 반환하게 된다.
python 함수로 구성한다.

Conditional Edge

조건에 따라서 어디로 가야할 지 결정하는 edge, 상태는 노드 뿐만 아니라 간선도 가져갈 수 있다.
여기서 conditional 함수가 반환되는 값을 실제 node 이름으로 매핑하는 기능이다.

def should_continue(state: State):
    if "error" in state["messages"][-1]:
        return "fix"
    return "exit"
  
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "fix": "repair_node",
        "exit": END
    }
)

간선 연결의 한계
- 복잡한 조건부 간선
- 상태 업데이터와 경로결정의 분리
- 동적 점프의 어려움

edge 기반 흐름제어는 graph라는 것 안에서 이루어지고 필요조건에 따라 명시적으로 선언되었다.
특정 상황에 따라 edge를 사용할 수 있다.

command 기반 Routing

node가 자신의 작업결과(State를 업데이트 하는 것)와 다음에 실행할 Node를 함께 반환하는 방식이다.
기존의 Conditional Edge에서 Node가 작업만 수행하고, 별도의 Router기능(condtional edge)로 다음에 실행할
Node를 결정한다.

ReAct pattern

사고, 행동, 추론 이 3가지를 반복하는 구조이다. 이벤트 루프, 조건절, 무엇을 만든다, 등 도출해낸 결과물을
보고 어떤 추론을 해야 하는지 모두 chain으로 만들고 반복한다.

도구를 활용하여 최신 정보, 계산, 검색 등을 수행
단계적으로 문제를 해결하여 복잡한 작업에 적합
반복적인 추론과 행동을 통해 더 정확한 결론을 얻을 수가 있다.

RetryPolicy

재시도와 관련된 설정을 담당한다. Jitter라는 것이 있다. 만약 동시 10명이 시도를 하는 경우 접근할 때 오류가 나게된다.
애플리케이션이 견딜 수 있는 부하는 최대 10개인 경우 큰 오류가 생기게 된다. jitter로 랜덤 지연을 부여하여 여러 사용자들이 각기 다른 시간에
접속하도록 조절하는 것이 가능하다.

langGraph code구현

def append_step_count(curr: int, new: int) -> int:
    return curr + new
    
from typing import Annotated
from operator import add

class TypedDictState(TypedDict):
    messages: Annotated[List[str], add]
    step_count: Annotated[int, append_step_count]

builder = StateGraph(TypedDictState)
builder. add_node(node_a)
builder. add_node(node_b)

builder.add_edge(START,"node_a")

builder.add_edge("node_a","node_b")
builder.add_edge("node_b",END)

agent = builder.compile()
result = agent.invoke({'messages': [], 'step_count': 0})

print(result)

동현의 나불나불

Contact