Forem: matias yoon

LangGraph 워크플로우 템플릿 (v38)

matias yoon — Mon, 25 May 2026 19:38:55 +0000

LangGraph 워크플로우 템플릿 (v38)

Python 개발자를 위한 LangGraph 기반 AI 에이전트 워크플로우 템플릿

LangChain과 LangGraph를 사용한 Python 기반 AI 에이전트 개발을 위한 실전 가이드입니다. 이 템플릿은 실제 개발자들이 겪는 문제를 해결하기 위해 설계되었습니다.

1. LangGraph 아키텍처 개요

LangGraph는 상태 기반 워크플로우 시스템으로, 다음과 같은 핵심 구성 요소로 구성됩니다:

핵심 구성 요소:

Nodes: 워크플로우의 단계
Edges: 노드 간의 연결
State: 워크플로우의 상태 관리
Checkpointing: 상태 저장 및 복원

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class GraphState(TypedDict):
    messages: Annotated[list, operator.add]
    context: str

# 그래프 생성
workflow = StateGraph(GraphState)

2. 템플릿 1: 간단한 RAG 에이전트 (검색 → 생성 → 검증)

대부분의 LLM 응용 프로그램은 RAG(검색 기반 생성) 패턴을 기반으로 합니다:

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

class RAGAgent:
    def __init__(self, llm):
        self.llm = llm
        self.retriever = self._setup_retriever()
        self.prompt = PromptTemplate.from_template(
            "주어진 컨텍스트를 사용하여 질문에 답하세요:\n\n{context}\n\n질문: {question}"
        )

    def retrieve(self, state):
        question = state["messages"][-1].content
        docs = self.retriever.invoke(question)
        context = "\n\n".join([doc.page_content for doc in docs])
        return {"context": context}

    def generate(self, state):
        chain = self.prompt | self.llm | StrOutputParser()
        result = chain.invoke({
            "question": state["messages"][-1].content,
            "context": state["context"]
        })
        return {"messages": [result]}

    def validate(self, state):
        # 간단한 검증 로직
        if len(state["messages"][-1].content) < 10:
            return {"messages": ["질문에 대한 충분한 정보가 없습니다."]}
        return {"messages": [state["messages"][-1].content]}

# 워크플로우 정의
def create_rag_workflow():
    workflow = StateGraph(GraphState)

    workflow.add_node("retrieve", RAGAgent.retrieve)
    workflow.add_node("generate", RAGAgent.generate)
    workflow.add_node("validate", RAGAgent.validate)

    workflow.set_entry_point("retrieve")
    workflow.add_edge("retrieve", "generate")
    workflow.add_edge("generate", "validate")
    workflow.add_edge("validate", END)

    return workflow.compile()

3. 템플릿 2: 다중 도구 에이전트 (계획 → 실행 → 관찰 → 결정)

복잡한 작업을 처리하는 다중 도구 에이전트:

from langchain.tools import Tool
from langchain_core.messages import ToolMessage
import json

class MultiToolAgent:
    def __init__(self):
        # 예시 도구들
        self.tools = [
            Tool(
                name="weather",
                func=lambda location: f"{location}의 날씨는 맑습니다.",
                description="날씨 정보 조회"
            ),
            Tool(
                name="calculator",
                func=lambda expression: f"결과: {eval(expression)}",
                description="수학 계산"
            )
        ]

    def plan(self, state):
        # 작업 계획 생성
        plan = {
            "tasks": [
                {"name": "weather", "params": {"location": "서울"}},
                {"name": "calculator", "params": {"expression": "2+2"}}
            ]
        }
        return {"plan": plan}

    def execute(self, state):
        # 도구 실행
        results = []
        for task in state["plan"]["tasks"]:
            tool = next(t for t in self.tools if t.name == task["name"])
            result = tool.func(**task["params"])
            results.append(result)
        return {"execution_results": results}

    def observe(self, state):
        # 결과 관찰 및 분석
        return {"analysis": "모든 작업이 완료되었습니다."}

    def decide(self, state):
        # 결정 로직
        if len(state["execution_results"]) > 0:
            return {"messages": ["작업이 성공적으로 완료되었습니다."]}
        return {"messages": ["작업 실패."]}

4. 템플릿 3: 인간-중개 워크플로우 (일시정지 → 검토 → 계속)

사람의 개입이 필요한 중요한 결정을 위한 워크플로우:

class HumanInLoopAgent:
    def __init__(self):
        self.user_feedback = None

    def pause_for_review(self, state):
        # 사용자 검토를 위한 일시정지
        return {"status": "pending_review"}

    def review(self, state):
        # 사용자 검토 처리
        if self.user_feedback is not None:
            return {"messages": [self.user_feedback]}
        return {"messages": ["사용자 검토가 필요합니다."]}

    def continue_workflow(self, state):
        # 워크플로우 재개
        return {"status": "processing"}

# 사용자 피드백 설정
def set_user_feedback(feedback):
    agent = HumanInLoopAgent()
    agent.user_feedback = feedback
    return agent

5. 템플릿 5: 병렬 실행 에이전트 (분기 → 처리 → 집계)

대량 데이터를 병렬로 처리:

from concurrent.futures import ThreadPoolExecutor
import asyncio

class ParallelAgent:
    def __init__(self, max_workers=4):
        self.max_workers = max_workers

    def fan_out(self, state):
        # 입력 데이터 분기
        data_chunks = [
            {"chunk_id": i, "data": f"chunk_{i}"}
            for i in range(4)
        ]
        return {"chunks": data_chunks}

    def process_chunk(self, chunk_data):
        # 개별 청크 처리
        time.sleep(0.1)  # 시뮬레이션
        return {"result": f"processed_{chunk_data['data']}"}

    async def parallel_process(self, state):
        # 병렬 처리
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = [
                executor.submit(self.process_chunk, chunk)
                for chunk in state["chunks"]
            ]
            results = [future.result() for future in futures]
        return {"processed_results": results}

    def aggregate(self, state):
        # 결과 집계
        aggregated = {
            "total_count": len(state["processed_results"]),
            "results": [r["result"] for r in state["processed_results"]]
        }
        return {"messages": [f"총 {aggregated['total_count']}개 처리 완료"]}

6. 상태 관리 패턴

효율적인 상태 관리:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint import BaseCheckpointSaver

class StateManager:
    def __init__(self):
        # 메모리 체크포인트 사용
        self.checkpointer = MemorySaver()

    def get_state(self, thread_id):
        # 상태 조회
        return self.checkpointer.get(thread_id)

    def update_state(self, thread_id, new_state):
        # 상태 업데이트
        self.checkpointer.put(thread_id, new_state)

    def rollback(self, thread_id, version):
        # 이전 버전으로 롤백
        pass

# 상태 저장 설정
def setup_workflow_with_checkpoint():
    workflow = StateGraph(GraphState)
    workflow.add_checkpoint(MemorySaver())
    return workflow

7. 스트리밍 및 실시간 업데이트

실시간 처리 지원:


python
from langchain_core.callbacks import BaseCallbackHandler

class StreamingHandler(BaseCallbackHandler):
    def on_chain_start(self, serialized, inputs, **kwargs):
        print("체인 시작")

    def on_chain_end(self, outputs, **kwargs):
        print("체인 종료")

    def on_llm_new_token(self, token, **kwargs):
        print(f"토큰: {token}", end="", flush=True)

class StreamingAgent:
    def __init__(self):
        self.streaming_handler = StreamingHandler()

    def stream_response(self, query):
        # 스트리밍 응답
        llm = ChatOpenAI(callbacks=[self.streaming_handler])
        response = llm.invoke(query)
        return response

# 사용 예시
async def stream_example():


---

📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)

터미널 AI 에이전트 구축 (v37)

matias yoon — Mon, 25 May 2026 19:24:08 +0000

터미널 AI 에이전트 구축 (v37)

터미널에서 AI 에이전트를 구축하는 것은 개발자에게 매우 실용적인 도구를 제공합니다. 이 가이드는 로컬 LLM을 활용한 CLI AI 에이전트를 구축하고, 실전 워크플로우에 적용하는 방법을 단계별로 설명합니다.

1. CLI AI 에이전트 생태계

현재 CLI AI 에이전트는 여러 형태로 존재합니다:

Aider: GitHub에서 개발된 코드 생성 도구로, 실제 파일에 직접 수정을 적용하는 기능이 강점입니다.

Continue.dev: VSCode 확장 프로그램이지만, 터미널에서도 사용 가능한 커스터마이징 가능한 에이전트입니다.

OpenCode: 직접 구축한 AI 에이전트로, 빠르고 간단한 커스터마이징이 가능합니다.

커스텀 스크립트: 가장 유연한 접근 방식으로, 특정 워크플로우에 맞춘 로직을 직접 구현할 수 있습니다.

2. 로컬 LLM API 엔드포인트 설정

먼저 로컬 LLM 서버를 실행해야 합니다. 이 예제에서는 LLaMA.cpp를 사용하겠습니다:

# 1. LLaMA.cpp 설치
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp
make

# 2. 모델 다운로드 및 변환
wget https://huggingface.co/TheBloke/Mistral-7B-v0.1-GGUF/resolve/main/mistral-7b-v0.1.Q4_K_M.gguf

# 3. 로컬 서버 시작
./server -m mistral-7b-v0.1.Q4_K_M.gguf -c 2048 --host 127.0.0.1 --port 1234

API 엔드포인트를 테스트하기 위해 curl 명령어를 사용합니다:

curl http://localhost:1234/v1/completions \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "Write a Python function that calculates fibonacci numbers",
    "max_tokens": 200,
    "temperature": 0.7
  }'

3. 간단한 Python CLI 에이전트 만들기

다음은 함수 호출 기능을 포함한 기초적인 Python CLI 에이전트입니다:

# agent.py
import requests
import json
import subprocess
import sys

class TerminalAgent:
    def __init__(self, api_url="http://localhost:1234/v1/completions"):
        self.api_url = api_url

    def call_model(self, prompt, max_tokens=300):
        """LLM API 호출"""
        headers = {"Content-Type": "application/json"}
        data = {
            "prompt": prompt,
            "max_tokens": max_tokens,
            "temperature": 0.3
        }

        response = requests.post(self.api_url, headers=headers, data=json.dumps(data))
        return response.json()['choices'][0]['text'].strip()

    def execute_command(self, command):
        """명령어 실행 및 결과 반환"""
        try:
            result = subprocess.run(
                command, 
                shell=True, 
                capture_output=True, 
                text=True, 
                check=True
            )
            return result.stdout
        except subprocess.CalledProcessError as e:
            return f"Error: {e.stderr}"

    def run(self, task):
        """주어진 작업을 실행"""
        print(f"작업 요청: {task}")

        # 작업 분석 및 분할
        analysis_prompt = f"""
        다음 작업을 분석하고 필요한 단계를 나열해주세요:
        {task}

        단계별로 JSON 형식으로 응답해주세요:
        {{
            "steps": ["step1", "step2", ...],
            "summary": "작업 요약"
        }}
        """

        analysis = self.call_model(analysis_prompt, 150)
        print(f"분석 결과: {analysis}")

        # 각 단계별 실행
        for i, step in enumerate(json.loads(analysis).get("steps", [])):
            print(f"\n단계 {i+1}: {step}")
            result = self.execute_command(step)
            print(f"결과: {result}")

if __name__ == "__main__":
    agent = TerminalAgent()
    task = "Python으로 주어진 디렉토리 내 모든 .py 파일을 찾아서 'print('Hello World')'를 추가해주세요"
    agent.run(task)

실행 방법:

python agent.py

4. tmux와 통합

tmux를 사용하여 에이전트가 작업을 수행하는 동안 터미널 세션을 유지하고, 다른 작업을 병행할 수 있습니다:

# tmux 세션 생성
tmux new-session -s ai_agent -d

# 에이전트 실행
tmux send-keys -t ai_agent "python agent.py" Enter

# 세션에 연결
tmux attach -t ai_agent

다음은 tmux 세션을 관리하는 스크립트입니다:

# tmux_helper.py
import subprocess
import os

class TmuxHelper:
    def __init__(self, session_name="ai_agent"):
        self.session_name = session_name

    def create_session(self):
        """새 tmux 세션 생성"""
        cmd = f"tmux new-session -s {self.session_name} -d"
        subprocess.run(cmd, shell=True, check=True)

    def send_command(self, command):
        """세션에 명령어 전송"""
        cmd = f"tmux send-keys -t {self.session_name} '{command}' Enter"
        subprocess.run(cmd, shell=True, check=True)

    def attach(self):
        """세션 연결"""
        cmd = f"tmux attach -t {self.session_name}"
        subprocess.run(cmd, shell=True, check=True)

    def kill_session(self):
        """세션 종료"""
        cmd = f"tmux kill-session -t {self.session_name}"
        subprocess.run(cmd, shell=True, check=True)

# 사용 예시
if __name__ == "__main__":
    helper = TmuxHelper("my_ai")
    helper.create_session()
    helper.send_command("python agent.py")

5. 커스텀 도구 개발

다음은 코드 검색, git 연동, 파일 작업을 위한 커스텀 도구들입니다:

# tools.py
import os
import subprocess
import json
import re

class CodeSearchTool:
    @staticmethod
    def find_files(directory, extension=".py"):
        """지정된 확장자 파일 찾기"""
        files = []
        for root, dirs, filenames in os.walk(directory):
            for filename in filenames:
                if filename.endswith(extension):
                    files.append(os.path.join(root, filename))
        return files

    @staticmethod
    def grep_in_files(pattern, files):
        """파일 내에서 패턴 검색"""
        results = {}
        for file_path in files:
            try:
                with open(file_path, 'r') as f:
                    content = f.read()
                    matches = re.findall(pattern, content)
                    if matches:
                        results[file_path] = matches
            except Exception as e:
                print(f"Error reading {file_path}: {e}")
        return results

class GitTool:
    @staticmethod
    def get_git_status():
        """Git 상태 확인"""
        result = subprocess.run(["git", "status", "--porcelain"], 
                              capture_output=True, text=True)
        return result.stdout.strip()

    @staticmethod
    def git_diff():
        """Git diff 출력"""
        result = subprocess.run(["git", "diff"], 
                              capture_output=True, text=True)
        return result.stdout

class FileOperationTool:
    @staticmethod
    def create_directory(path):
        """디렉토리 생성"""
        os.makedirs(path, exist_ok=True)
        return f"Directory {path} created"

    @staticmethod
    def read_file(path):
        """파일 읽기"""
        try:
            with open(path, 'r') as f:
                return f.read()
        except Exception as e:
            return f"Error reading {path}: {e}"

    @staticmethod
    def write_file(path, content):
        """파일 쓰기"""
        try:
            with open(path, 'w') as f:
                f.write(content)
            return f"File {path} written successfully"
        except Exception as e:
            return f"Error writing {path}: {e}"

# 사용 예시
if __name__ == "__main__":
    search_tool = CodeSearchTool()
    files = search_tool.find_files(".", ".py")
    print(f"Found {len(files)} Python files")

    git_tool = GitTool()
    print("Git status:", git_tool.get_git_status())

6. 컨텍스트 윈도우 관리

대규모 코드베이스에서 성능을 최적화하기 위해 컨텍스트 윈도우를 관리해야 합니다:




---

📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)

RAG 시스템 실전 구축 (v37)

matias yoon — Mon, 25 May 2026 19:09:22 +0000

RAG 시스템 실전 구축 (v37)

실제로 구축할 수 있는 RAG 시스템 구현 가이드

1. RAG 기초 개념: 검색 → 보완 → 생성 루프

RAG(Retrieval-Augmented Generation)는 대규모 언어 모델(LLM)이 외부 지식을 활용하여 정확한 답변을 생성할 수 있도록 하는 아키텍처입니다. 이 루프는 세 가지 주요 단계로 구성됩니다:

# RAG 루프 구조
class RAGPipeline:
    def __init__(self):
        self.retriever = None
        self.generator = None

    def process_query(self, query):
        # 1. 검색: 쿼리와 유사한 문서 검색
        retrieved_docs = self.retriever.retrieve(query)

        # 2. 보완: 검색된 문서를 쿼리와 결합
        augmented_context = self.augment_context(query, retrieved_docs)

        # 3. 생성: LLM이 보완된 문맥을 기반으로 답변 생성
        answer = self.generator.generate(query, augmented_context)

        return answer

2. 청킹 전략: 의미적, 재귀적, 에이전트 기반

청킹은 문서를 검색 가능한 단위로 나누는 작업입니다:

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter
)
import tiktoken

class ChunkingStrategies:
    @staticmethod
    def semantic_chunking(documents, model="text-embedding-3-small"):
        """의미적 청킹: 의미 단위로 문서 분할"""
        # 의미적 단위를 찾기 위한 임베딩 기반 분할
        from sentence_transformers import SentenceTransformer
        encoder = SentenceTransformer(model)

        # 문서를 의미 단위로 분할
        # 예시: 문장 단위 분할
        sentences = doc.split('.')
        embeddings = encoder.encode(sentences)

        # 임베딩 간 유사도를 기준으로 청킹
        return sentences

    @staticmethod
    def recursive_chunking(documents, chunk_size=512, chunk_overlap=50):
        """재귀적 청킹: 계층적 구조로 문서 분할"""
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", " ", ""]
        )
        return text_splitter.split_text(documents)

    @staticmethod
    def agentic_chunking(documents, chunk_size=512):
        """에이전트 기반 청킹: 문서 구조를 고려한 청킹"""
        # 문서 헤더, 본문, 결론 구분하여 청킹
        sections = documents.split('\n\n')
        chunks = []
        for section in sections:
            if len(section) > chunk_size:
                chunks.extend(
                    [section[i:i+chunk_size] 
                     for i in range(0, len(section), chunk_size)]
                )
            else:
                chunks.append(section)
        return chunks

# 사용 예시
strategy = ChunkingStrategies()
documents = "..."
chunks = strategy.recursive_chunking(documents, chunk_size=512)

3. 임베딩 모델 선택과 비교

임베딩 모델은 문서와 쿼리를 벡터 공간으로 변환하는 핵심 요소입니다:

from sentence_transformers import SentenceTransformer
from transformers import AutoTokenizer, AutoModel
import numpy as np
import time

class EmbeddingComparison:
    def __init__(self):
        self.models = {
            'all-MiniLM-L6-v2': SentenceTransformer('all-MiniLM-L6-v2'),
            'all-mpnet-base-v2': SentenceTransformer('all-mpnet-base-v2'),
            'text-embedding-3-small': SentenceTransformer('text-embedding-3-small'),
            'gte-small': SentenceTransformer('thenlper/gte-small')
        }

    def compare_models(self, texts):
        """모델별 성능 비교"""
        results = {}

        for model_name, model in self.models.items():
            start_time = time.time()
            embeddings = model.encode(texts)
            end_time = time.time()

            results[model_name] = {
                'embedding_size': embeddings.shape,
                'processing_time': end_time - start_time,
                'memory_usage': embeddings.nbytes
            }

        return results

    def benchmark_embeddings(self, query, corpus, top_k=5):
        """임베딩 성능 벤치마킹"""
        # 모든 모델로 벡터 생성
        vectors = {}
        for name, model in self.models.items():
            vectors[name] = model.encode(corpus)

        # 쿼리 벡터 생성
        query_vector = self.models['all-MiniLM-L6-v2'].encode([query])

        # 유사도 계산
        similarities = {}
        for name, embeddings in vectors.items():
            similarity = np.dot(embeddings, query_vector.T).flatten()
            top_indices = np.argsort(similarity)[-top_k:][::-1]
            similarities[name] = top_indices.tolist()

        return similarities

# 사용 예시
benchmark = EmbeddingComparison()
texts = ["문서 1", "문서 2", "문서 3"]
results = benchmark.compare_models(texts)

4. 벡터 데이터베이스 비교

다양한 벡터 데이터베이스의 특징과 성능 비교:

# Chroma 벡터 데이터베이스
import chromadb
from chromadb import Client

class ChromaVectorDB:
    def __init__(self, persist_directory="./chroma_db"):
        self.client = Client()
        self.collection = self.client.get_or_create_collection(
            name="rag_collection",
            metadata={"hnsw:space": "cosine"}
        )

    def add_documents(self, documents, embeddings, ids):
        self.collection.add(
            documents=documents,
            embeddings=embeddings,
            ids=ids
        )

    def search(self, query_embedding, n_results=5):
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=n_results
        )
        return results

# Qdrant 벡터 데이터베이스
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue

class QdrantVectorDB:
    def __init__(self, host="localhost", port=6333):
        self.client = QdrantClient(host=host, port=port)

    def create_collection(self, collection_name, vector_size=384):
        self.client.recreate_collection(
            collection_name=collection_name,
            vectors_config={"size": vector_size, "distance": "Cosine"}
        )

    def search(self, query_vector, collection_name, limit=5):
        results = self.client.search(
            collection_name=collection_name,
            query_vector=query_vector,
            limit=limit
        )
        return results

# pgvector 예시
import psycopg2
from psycopg2.extras import Json

class PGVectorDB:
    def __init__(self, connection_string):
        self.conn = psycopg2.connect(connection_string)

    def create_table(self):
        with self.conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS documents (
                    id UUID PRIMARY KEY,
                    content TEXT,
                    embedding VECTOR(384),
                    metadata JSONB
                )
            """)
            cur.execute("CREATE INDEX IF NOT EXISTS idx_embedding ON documents USING ivfflat (embedding vector_cosine_ops)")

    def insert_document(self, doc_id, content, embedding, metadata):
        with self.conn.cursor() as cur:
            cur.execute("""
                INSERT INTO documents (id, content, embedding, metadata)
                VALUES (%s, %s, %s, %s)
                ON CONFLICT (id) DO UPDATE SET
                    content = EXCLUDED.content,
                    embedding = EXCLUDED.embedding,
                    metadata = EXCLUDED.metadata
            """, (doc_id, content, embedding, Json(metadata)))

    def search(self, query_embedding, limit=5):
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT content, metadata, embedding <-> %s as distance
                FROM documents
                ORDER BY distance
                LIMIT %s
            """, (query_embedding, limit))
            return cur.fetchall()

5. 전체 RAG 파이프라인 코드

실제 구현 예시:


python
import os
from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List, Dict, Any

class FullRAGPipeline:
    def __init__(self, 
                 embedding_model_name="all-MiniLM-L6-v2",
                 vector_db_type="chroma",
                 local_model_path=None):

        # 임베딩 모델 로드
        self.embedding_model = SentenceTransformer(embedding_model_name)

        # 벡터 데이터베이스 초기화
        if vector_db_type == "chroma":
            self.vector_db = ChromaVectorDB()
        elif vector_db_type == "qdrant":
            self.vector_db = QdrantVectorDB()
        else:


---

📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($7)

로컬 LLM 셋업 가이드 (v37)

matias yoon — Mon, 25 May 2026 18:54:35 +0000

로컬 LLM 셋업 가이드 (v37)

1. 개요 및 사전 요구사항

로컬 LLM 실행은 AI 에이전트 개발 및 프로덕션 배포에 있어 필수적인 기술입니다. 이 가이드는 Linux 기반 시스템에서 높은 성능과 포터블성을 갖춘 로컬 LLM 구성을 안내합니다.

사전 요구사항:

Linux 운영체제 (Ubuntu 20.04 이상 권장)
GPU (NVIDIA RTX 30xx 이상 권장)
최소 16GB RAM (32GB 이상 권장)
100GB 이상 여유 디스크 공간

2. 프레임워크 비교

프레임워크	장점	단점	적합성
llama.cpp	최소 의존성, 높은 포터빌리티, C++ 기반	명령줄 중심, 관리 용이성 부족	초급자 및 포터블 개발
Ollama	Docker 기반, 쉬운 설치, GUI 지원	성능 저하, 메모리 사용량 증가	빠른 테스트 및 개발
vLLM	최고의 추론 성능	복잡한 설치, CPU 중심	고성능 요구사항
LocalAI	API 호환성, 다중 모델 관리	복잡한 설정, 높은 리소스 소모	프로덕션 및 에이전트 개발

추천: llama.cpp + LocalAI 조합을 사용하여 성능과 관리 용이성을 균형 있게 유지.

3. 설치 단계 (llama.cpp 기반)

3.1 기본 시스템 준비

# 시스템 업데이트
sudo apt update && sudo apt upgrade -y

# 의존성 설치
sudo apt install git cmake build-essential libopenblas-dev liblapack-dev -y

# CUDA 설치 (NVIDIA GPU 사용 시)
wget https://developer.download.nvidia.com/compute/cuda/12.4.1/local_installers/cuda_12.4.1_550.54.15_linux.run
sudo sh cuda_12.4.1_550.54.15_linux.run

3.2 llama.cpp 설치

# 소스코드 다운로드
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp

# 빌드
make clean
make -j$(nproc)

# CUDA 지원 빌드 (GPU 사용 시)
make -j$(nproc) CUDA=1

3.3 LocalAI 설치 (API 포워딩용)

# LocalAI 다운로드
curl -s https://raw.githubusercontent.com/go-skynet/LocalAI/main/install.sh | sh

# 로컬 실행
./local-ai server --port 8080

4. 모델 선택 가이드

모델	크기	추론 성능	추천 사용 사례
Llama3-8B	4.7GB	높음	일반적인 에이전트 개발
Mistral-7B	4.1GB	높음	코딩 및 분석 작업
Phi-3	3.8GB	높음	빠른 추론 필요
Gemma-2B	2.2GB	중간	리소스 제한 환경

# 모델 다운로드 예시
wget https://huggingface.co/TheBloke/Llama-3-8B-GGUF/resolve/main/llama-3-8b.Q4_K_M.gguf

5. 퀀타이제이션 타입 설명

# 퀀타이제이션 유형별 성능 비교
Q4_K_M: 최적화된 4bit, 손실 최소화
Q5_K_M: 5bit, 더 높은 정확도
Q8_0: 8bit, 최대 정확도
Q2_K: 2bit, 최소 리소스 사용

추천: Q4_K_M으로 시작하여 성능과 정확도 균형 찾기

6. API 설정 및 통합

6.1 LocalAI API 설정

# config.yaml
models:
  - name: llama3-8b
    path: /models/llama-3-8b.Q4_K_M.gguf
    context_length: 4096
    n_gpu_layers: 40

server:
  port: 8080
  host: 0.0.0.0

6.2 Python 클라이언트 예제

# localai_client.py
import requests
import json

class LocalAIClient:
    def __init__(self, base_url="http://localhost:8080"):
        self.base_url = base_url

    def chat(self, prompt, model="llama3-8b"):
        response = requests.post(
            f"{self.base_url}/v1/chat/completions",
            json={
                "model": model,
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0.7
            }
        )
        return response.json()

# 사용 예시
client = LocalAIClient()
result = client.chat("Hello, how are you?")
print(result['choices'][0]['message']['content'])

7. Systemd 서비스 설정

# /etc/systemd/system/localai.service
[Unit]
Description=LocalAI Server
After=network.target

[Service]
Type=simple
User=developer
WorkingDirectory=/opt/localai
ExecStart=/opt/localai/local-ai server --port 8080
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

# 서비스 활성화
sudo systemctl daemon-reload
sudo systemctl enable localai
sudo systemctl start localai
sudo systemctl status localai

8. 모니터링 및 성능 튜닝

8.1 성능 모니터링

# CPU 및 GPU 모니터링
watch -n 1 nvidia-smi

# 메모리 사용량
free -h

# 추론 성능 측정
time ./llama.cpp/main -m models/llama-3-8b.Q4_K_M.gguf -p "Hello world" -n 100

8.2 성능 최적화 설정

# 환경 변수 설정
export CUDA_VISIBLE_DEVICES=0
export GGML_CUDA_F16=1

# 추론 최적화
./llama.cpp/main \
  -m models/llama-3-8b.Q4_K_M.gguf \
  -p "Your prompt here" \
  -n 100 \
  --temp 0.7 \
  --threads 8 \
  --n-gpu-layers 40 \
  --ctx-size 4096

9. 실제 사용 예시

9.1 에이전트 테스트

# 간단한 테스트 쿼리
./llama.cpp/main \
  -m models/llama-3-8b.Q4_K_M.gguf \
  -p "Write a Python function to calculate Fibonacci numbers" \
  -n 50 \
  --temp 0.3

9.2 API 연동 테스트

# JSON 요청 테스트
curl -X POST http://localhost:8080/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "llama3-8b",
    "messages": [{"role": "user", "content": "Explain quantum computing in simple terms"}],
    "temperature": 0.7
  }'

10. 문제 해결 팁

10.1 메모리 부족 문제

# 메모리 확인
free -h
nvidia-smi

# GPU 레이어 수 조정
./llama.cpp/main -m model.gguf -n-gpu-layers 20

10.2 성능 저하 해결

# CPU 스레드 수 조정
export OMP_NUM_THREADS=8

이 가이드를 따르면 빠르게 로컬 LLM 환경을 구축할 수 있으며, 프로덕션 준비가 완료된 에이전트 개발에 활용할 수 있습니다. 각 단계별 명령어와 설정은 실제 개발 환

📥 Get the full guide on Gumroad: https://gumroad.com/l/auto ($7)

LangGraph 워크플로우 템플릿 (v37)

matias yoon — Mon, 25 May 2026 18:39:48 +0000

LangGraph 워크플로우 템플릿 (v37)

LangGraph 아키텍처 개요

LangGraph는 LangChain의 고급 워크플로우 시스템으로, 상태 기반 그래프 기반 AI 에이전트를 구축하는 데 최적화되어 있습니다. 핵심 구성 요소는 다음과 같습니다:

노드 (Nodes): 각각의 작업 단계. 입력 상태를 받아 출력 상태를 반환합니다.
엣지 (Edges): 노드 간의 흐름을 정의합니다.
상태 (State): 에이전트의 현재 상태를 나타내는 Pydantic 모델입니다.
체크포인팅 (Checkpointing): 상태를 저장하고 복원할 수 있도록 합니다.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]

# 기본 그래프 구조
graph = StateGraph(AgentState)

템플릿 1: 간단한 RAG 에이전트 (검색 → 생성 → 검증)

이 템플릿은 문서 검색 및 생성을 수행하는 기본적인 RAG 에이전트입니다.

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableLambda
import operator

# RAG 상태 정의
class RagState(TypedDict):
    query: str
    context: str
    answer: str
    validation: bool

# 검색 노드
def retrieve(state: RagState):
    # 문서 검색 로직 (예: FAISS, Pinecone)
    vectorstore = ... # Vector store 초기화
    docs = vectorstore.similarity_search(state["query"], k=3)
    context = "\n\n".join([doc.page_content for doc in docs])
    return {"context": context}

# 생성 노드
def generate(state: RagState):
    prompt = PromptTemplate.from_template(
        "질문: {query}\n\n문맥: {context}\n\n답변:"
    )
    model = ChatOpenAI(model="gpt-4")
    answer = model.invoke(prompt.format(query=state["query"], context=state["context"]))
    return {"answer": answer.content}

# 검증 노드
def validate(state: RagState):
    # 생성된 답변의 검증 로직
    prompt = PromptTemplate.from_template(
        "다음 답변이 질문에 대한 적절한 답변인지 평가하세요:\n\n질문: {query}\n\n답변: {answer}\n\n평가: (예/아니요)"
    )
    model = ChatOpenAI(model="gpt-4")
    validation = model.invoke(prompt.format(query=state["query"], answer=state["answer"]))
    return {"validation": validation.content.startswith("예")}

# RAG 그래프 생성
rag_graph = StateGraph(RagState)
rag_graph.add_node("retrieve", retrieve)
rag_graph.add_node("generate", generate)
rag_graph.add_node("validate", validate)

# 엣지 정의
rag_graph.set_entry_point("retrieve")
rag_graph.add_edge("retrieve", "generate")
rag_graph.add_edge("generate", "validate")

# 조건부 엣지
def route_validation(state: RagState):
    if state["validation"]:
        return "end"
    else:
        return "retrieve"  # 다시 검색

rag_graph.add_conditional_edges(
    "validate",
    route_validation,
    {"end": END, "retrieve": "retrieve"}
)

템플릿 2: 멀티-도구 에이전트 (계획 → 실행 → 관찰 → 결정)

이 템플릿은 다중 도구를 사용하여 작업을 수행하는 에이전트입니다.

from typing import List
from langchain_core.tools import Tool

class ToolAgentState(TypedDict):
    messages: list[BaseMessage]
    plan: List[str]
    execution_log: List[str]
    final_answer: str

# 도구 정의
def web_search(query: str) -> str:
    # 웹 검색 로직
    return f"검색 결과: {query}"

def calculator(expression: str) -> str:
    # 계산기 로직
    return f"계산 결과: {eval(expression)}"

tools = [
    Tool(name="web_search", func=web_search, description="웹에서 정보 검색"),
    Tool(name="calculator", func=calculator, description="수학 계산")
]

# 계획 노드
def plan(state: ToolAgentState):
    # 작업 계획 생성
    prompt = PromptTemplate.from_template(
        "다음 작업을 수행하기 위한 계획을 세우세요:\n\n{messages}"
    )
    model = ChatOpenAI(model="gpt-4")
    plan = model.invoke(prompt.format(messages=state["messages"]))
    return {"plan": [task.strip() for task in plan.content.split('\n') if task.strip()]}

# 실행 노드
def execute(state: ToolAgentState):
    # 계획된 작업 실행
    execution_log = []
    for task in state["plan"]:
        if "search" in task.lower():
            result = web_search(task)
        elif "calculate" in task.lower():
            result = calculator(task)
        execution_log.append(f"{task}: {result}")
    return {"execution_log": execution_log}

# 관찰 노드
def observe(state: ToolAgentState):
    # 실행 결과 분석
    prompt = PromptTemplate.from_template(
        "다음 실행 결과를 분석하세요:\n\n{execution_log}"
    )
    model = ChatOpenAI(model="gpt-4")
    analysis = model.invoke(prompt.format(execution_log=state["execution_log"]))
    return {"final_answer": analysis.content}

# 멀티-도구 에이전트 그래프
tool_agent = StateGraph(ToolAgentState)
tool_agent.add_node("plan", plan)
tool_agent.add_node("execute", execute)
tool_agent.add_node("observe", observe)

tool_agent.set_entry_point("plan")
tool_agent.add_edge("plan", "execute")
tool_agent.add_edge("execute", "observe")

템플릿 3: 인간-중개 워크플로우 (일시정지 → 검토 → 계속)

이 템플릿은 인간의 입력을 포함한 결정 프로세스를 구현합니다.

from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import AIMessage, HumanMessage
from typing import Optional

class HumanInLoopState(TypedDict):
    messages: list[BaseMessage]
    user_review: Optional[str]
    final_decision: Optional[str]

# 일시정지 노드
def pause_for_review(state: HumanInLoopState):
    # 사용자 검토 요청
    human_message = HumanMessage(content="다음 작업을 검토해주세요:")
    return {"messages": [human_message]}

# 검토 노드
def review(state: HumanInLoopState):
    # 사용자 검토 처리
    return {"user_review": state["messages"][-1].content}

# 결정 노드
def decide(state: HumanInLoopState):
    # 사용자 피드백 기반 결정
    prompt = PromptTemplate.from_template(
        "사용자 피드백: {user_review}\n\n결정하세요:"
    )
    model = ChatOpenAI(model="gpt-4")
    decision = model.invoke(prompt.format(user_review=state["user_review"]))
    return {"final_decision": decision.content}

# 인간-중개 워크플로우
human_loop = StateGraph(HumanInLoopState)
human_loop.add_node("pause", pause_for_review)
human_loop.add_node("review", review)
human_loop.add_node("decide", decide)

human_loop.set_entry_point("pause")
human_loop.add_edge("pause", "review")
human_loop.add_edge("review", "decide")

템플릿 5: 병렬 실행 에이전트 (팬아웃 → 처리 → 집계)

이 템플릿은 여러 작업을 병렬로 처리하고 결과를 집계합니다.


python
from concurrent.futures import ThreadPoolExecutor
import asyncio

class ParallelAgentState(TypedDict):
    tasks: list[str]
    results: list[dict]
    aggregated_result: str

# 병렬 처리 노드
def fan_out(state: ParallelAgentState):
    # 작업 분할
    tasks = state["tasks"]
    return {"tasks": tasks}

# 처리 노드 (병렬 실행)
def process_task(state: ParallelAgentState):
    # 각 작업을 병렬로 처리
    def process_single_task(task):
        # 작업 처리 로직
        return {"task": task, "result": f"처리 완료: {task}"}

    # 병렬 실행
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(process_single_task, task) for task in state["tasks"]]
        results = [future.result() for future in futures]

    return {"results": results}

# 집계 노드
def aggregate_results(state: ParallelAgentState):
    # 결과 집계
    aggregated = "\n".join([f"{r['task']}: {r['result']}" for r in state

---

📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)

터미널 AI 에이전트 구축 (v36)

matias yoon — Mon, 25 May 2026 18:25:01 +0000

터미널 AI 에이전트 구축 (v36)

터미널에서 작동하는 AI 에이전트를 구축하는 것은 현대 개발 워크플로우에서 핵심적인 도구로 자리 잡고 있습니다. 이 가이드는 실질적인 비용 ($3-$7)의 가치를 제공하는 터미널 기반 AI 에이전트를 구축하는 방법을 다룹니다.

1. CLI AI 에이전트 생태계

현재 CLI AI 에이전트는 다양한 솔루션으로 구성되어 있습니다:

Aider: Git 기반 코드 생성 도구

pip install aider
aider --help

Continue.dev: VS Code 확장이지만 CLI에서도 사용 가능

npm install -g continue

OpenCode: 직접 구축한 커스텀 에이전트

# 커스텀 스크립트 예시
pip install openai

내부 에이전트: 최소한의 기능을 가진 자체 구축 도구

2. 로컬 LLM API 엔드포인트 설정

로컬 LLM을 위한 API 엔드포인트를 설정하여 성능과 비용을 최적화합니다:

# Ollama 설치 (가장 간단한 방법)
curl -fsSL https://ollama.com/install.sh | sh

# 로컬 모델 실행
ollama run llama3.2:3b
ollama run codellama:7b

# API 엔드포인트 시작
ollama serve

API 엔드포인트 구성:

# server.py
from flask import Flask, request, jsonify
import requests

app = Flask(__name__)

@app.route('/chat', methods=['POST'])
def chat():
    data = request.json
    prompt = data.get('prompt', '')

    response = requests.post(
        'http://localhost:11434/api/generate',
        json={
            'model': 'llama3.2:3b',
            'prompt': prompt,
            'stream': False
        }
    )

    return jsonify(response.json())

if __name__ == '__main__':
    app.run(port=8000)

3. 함수 호출 기능을 가진 간단한 Python CLI 에이전트

# agent.py
import openai
import json
import subprocess
from typing import List, Dict
import os

class TerminalAgent:
    def __init__(self, api_key: str = None):
        self.client = openai.OpenAI(
            base_url="http://localhost:8000",
            api_key="ollama"
        )

    def execute_command(self, command: str) -> str:
        """명령어 실행 및 결과 반환"""
        try:
            result = subprocess.run(
                command, 
                shell=True, 
                capture_output=True, 
                text=True
            )
            return result.stdout + result.stderr
        except Exception as e:
            return f"Error: {str(e)}"

    def call_function(self, function_name: str, args: Dict) -> str:
        """함수 호출"""
        functions = {
            'execute_command': self.execute_command,
            'list_files': self.list_files,
            'git_status': self.git_status
        }

        if function_name in functions:
            return functions[function_name](**args)
        return f"Function {function_name} not found"

    def list_files(self, path: str = ".") -> str:
        """파일 목록 조회"""
        try:
            files = os.listdir(path)
            return json.dumps(files, indent=2)
        except Exception as e:
            return f"Error listing files: {str(e)}"

    def git_status(self) -> str:
        """Git 상태 조회"""
        return self.execute_command("git status --porcelain")

    def chat(self, prompt: str, functions: List[Dict] = None) -> str:
        """AI 채팅 메시지 처리"""
        messages = [
            {
                "role": "system",
                "content": "You are a helpful terminal assistant that can execute commands and provide code assistance."
            },
            {
                "role": "user",
                "content": prompt
            }
        ]

        response = self.client.chat.completions.create(
            model="llama3.2:3b",
            messages=messages,
            functions=functions,
            function_call="auto"
        )

        return response.choices[0].message.content

# 사용법
if __name__ == "__main__":
    agent = TerminalAgent()
    result = agent.chat("git 상태를 확인해줘")
    print(result)

4. tmux/터미널 멀티플렉서 통합

터미널 에이전트와 tmux를 통합하여 복잡한 작업 처리:

# tmux 스크립트 생성
cat > tmux_agent.sh << 'EOF'
#!/bin/bash
# tmux 세션에 에이전트 실행
SESSION_NAME="ai_agent"

# 새 세션 생성
tmux new-session -d -s $SESSION_NAME

# 에이전트 실행
tmux send-keys -t $SESSION_NAME "python agent.py" Enter

# 세션을 포그라운드로 전환
tmux attach -t $SESSION_NAME
EOF

chmod +x tmux_agent.sh

# enhanced_agent.py
import subprocess
import os
import sys
import time

class EnhancedTerminalAgent(TerminalAgent):
    def create_tmux_session(self, session_name: str, command: str):
        """tmux 세션 생성"""
        try:
            # 새 세션 생성
            subprocess.run([
                'tmux', 'new-session', '-d', '-s', session_name
            ], check=True)

            # 명령어 실행
            subprocess.run([
                'tmux', 'send-keys', '-t', session_name, command, 'Enter'
            ], check=True)

            return True
        except subprocess.CalledProcessError:
            return False

    def get_tmux_session_status(self, session_name: str) -> str:
        """tmux 세션 상태 확인"""
        try:
            result = subprocess.run([
                'tmux', 'list-sessions', '-F', '#{session_name}: #{session_attached}'
            ], capture_output=True, text=True)
            return result.stdout
        except:
            return "Error checking session"

# 사용 예시
agent = EnhancedTerminalAgent()
agent.create_tmux_session("coding_session", "vim main.py")

5. 커스텀 도구 개발


python
# custom_tools.py
import os
import glob
from pathlib import Path

class CustomTools:
    @staticmethod
    def find_files(pattern: str, directory: str = ".") -> str:
        """파일 찾기"""
        try:
            search_path = os.path.join(directory, pattern)
            files = glob.glob(search_path, recursive=True)
            return json.dumps(files, indent=2)
        except Exception as e:
            return f"Error: {str(e)}"

    @staticmethod
    def search_code(pattern: str, directory: str = ".") -> str:
        """코드 검색"""
        try:
            results = []
            for root, dirs, files in os.walk(directory):
                for file in files:
                    if file.endswith(('.py', '.js', '.ts', '.html', '.css')):
                        filepath = os.path.join(root, file)
                        with open(filepath, 'r') as f:
                            content = f.read()
                            if pattern in content:
                                results.append({
                                    'file': filepath,
                                    'line': content.count('\n', 0, content.find(pattern)) + 1
                                })
            return json.dumps(results, indent=2)
        except Exception as e:
            return f"Error: {str(e)}"

    @staticmethod
    def get_project_info(directory: str = ".") -> str:
        """프로젝트 정보 조회"""
        info = {
            'current_directory': os.getcwd(),
            'git_status': subprocess.getoutput('git status --porcelain'),
            'file_count': len([f for f in Path(directory).rglob('*') if f.is_file()]),
            'python_files': len([f for f in Path(directory).rglob('*.py') if f.is_file()]),
            'node_modules': os.path.exists(os.path.join(directory, 'node_modules'))
        }
        return json.dumps(info, indent=2)

# 도구 등록
functions = [
    {
        "name": "execute_command",
        "description": "명령어 실행",
        "parameters": {
            "type": "object",
            "properties": {
                "command": {"type": "string"}
            },
            "required": ["command"]
        }
    },
    {
        "name": "find_files",
        "description": "파일 찾기",
        "parameters": {
            "type": "object",
            "properties": {
                "pattern": {"type": "string"},
                "directory": {"type": "string"}
            },
            "required": ["pattern"]
        }
    },
    {
        "name": "search_code",
        "description": "코드 검색",
        "parameters": {
            "type": "object",
            "properties": {


---

📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)

RAG 시스템 실전 구축 (v36)

matias yoon — Mon, 25 May 2026 18:10:13 +0000

RAG 시스템 실전 구축 (v36)

1. RAG 기초 개념

RAG (Retrieval-Augmented Generation)은 정보 검색과 생성 모델을 통합한 아키텍처로, 대규모 언어 모델(LLM)이 외부 데이터를 기반으로 정확한 응답을 생성할 수 있도록 합니다. 이 시스템은 다음과 같은 핵심 루프를 따릅니다:

질문 → 검색 → 증강 → 생성

검색 단계에서는 질문과 관련된 문서 청크를 벡터 데이터베이스에서 찾습니다. 증강 단계에서는 검색된 문서를 프롬프트에 포함시켜 LLM에게 추가 정보를 제공합니다. 생성 단계에서는 LLM이 증강된 프롬프트를 기반으로 답변을 생성합니다.

2. 청킹 전략

2.1 Semantic Chunking

의미 기반 청킹은 문장 단위로 청크를 분할하여 의미적 연관성을 유지합니다.

import tiktoken
from typing import List, Dict

class SemanticChunker:
    def __init__(self, chunk_size: int = 512, overlap: int = 64):
        self.chunk_size = chunk_size
        self.overlap = overlap
        self.tokenizer = tiktoken.get_encoding("cl100k_base")

    def chunk_text(self, text: str) -> List[str]:
        tokens = self.tokenizer.encode(text)
        chunks = []

        for i in range(0, len(tokens), self.chunk_size - self.overlap):
            chunk_tokens = tokens[i:i + self.chunk_size]
            chunk_text = self.tokenizer.decode(chunk_tokens)
            chunks.append(chunk_text)

        return chunks

# 사용 예시
chunker = SemanticChunker(chunk_size=512, overlap=64)
document = "..." # 긴 문서
chunks = chunker.chunk_text(document)

2.2 Recursive Chunking

재귀적 청킹은 문서를 단계적으로 분할하여 최적의 청크 크기를 찾습니다.

import re
from typing import List

class RecursiveChunker:
    def __init__(self, max_chunk_size: int = 1024):
        self.max_chunk_size = max_chunk_size

    def chunk_text(self, text: str) -> List[str]:
        # 문장 단위 분할
        sentences = re.split(r'[.!?]+', text)
        chunks = []
        current_chunk = ""

        for sentence in sentences:
            if len(current_chunk) + len(sentence) < self.max_chunk_size:
                current_chunk += sentence + ". "
            else:
                if current_chunk:
                    chunks.append(current_chunk.strip())
                current_chunk = sentence + ". "

        if current_chunk:
            chunks.append(current_chunk.strip())

        return chunks

2.3 Agentic Chunking

엔티티 기반 청킹은 문서에서 특정 엔티티(사람, 장소, 조직)를 기준으로 분할합니다.

import spacy
from typing import List, Dict

class AgenticChunker:
    def __init__(self):
        self.nlp = spacy.load("en_core_web_sm")

    def chunk_by_entities(self, text: str, entity_types: List[str] = None) -> List[str]:
        doc = self.nlp(text)
        chunks = []
        current_chunk = ""

        for ent in doc.ents:
            if entity_types is None or ent.label_ in entity_types:
                if len(current_chunk) > 0:
                    chunks.append(current_chunk.strip())
                current_chunk = ent.text + " "

        if current_chunk:
            chunks.append(current_chunk.strip())

        return chunks

3. 임베딩 모델 선택 및 비교

3.1 모델 비교 테스트

from sentence_transformers import SentenceTransformer
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class EmbeddingBenchmark:
    def __init__(self):
        self.models = {
            'all-MiniLM-L6-v2': SentenceTransformer('all-MiniLM-L6-v2'),
            'sentence-t5-base': SentenceTransformer('sentence-t5-base'),
            'multi-qa-MiniLM-L6-v2': SentenceTransformer('multi-qa-MiniLM-L6-v2')
        }

    def benchmark_models(self, texts: List[str], query: str) -> Dict[str, float]:
        results = {}
        query_embedding = self.models['all-MiniLM-L6-v2'].encode([query])

        for name, model in self.models.items():
            embeddings = model.encode(texts)
            similarity = cosine_similarity(query_embedding, embeddings)[0]
            results[name] = float(np.mean(similarity))

        return results

# 성능 비교
benchmark = EmbeddingBenchmark()
texts = ["AI technology", "Machine Learning", "Deep Learning"]
query = "artificial intelligence"
scores = benchmark.benchmark_models(texts, query)
print(scores)

4. 벡터 데이터베이스 비교

4.1 Chroma vs Qdrant vs pgvector

# Chroma 구현
import chromadb
from chromadb.utils import embedding_functions

class ChromaVectorDB:
    def __init__(self, collection_name: str):
        self.client = chromadb.Client()
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            embedding_function=embedding_functions.DefaultEmbeddingFunction()
        )

    def add_documents(self, documents: List[str], ids: List[str]):
        self.collection.add(
            documents=documents,
            ids=ids
        )

    def search(self, query: str, limit: int = 5):
        results = self.collection.query(
            query_texts=[query],
            n_results=limit
        )
        return results

# Qdrant 구현
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Filter, FieldCondition

class QdrantVectorDB:
    def __init__(self, collection_name: str):
        self.client = QdrantClient(":memory:")  # 메모리 기반
        self.collection_name = collection_name

        # 컬렉션 생성
        self.client.recreate_collection(
            collection_name=collection_name,
            vectors_config=VectorParams(size=384, distance="Cosine")
        )

    def add_documents(self, documents: List[str], ids: List[str]):
        points = [
            {
                "id": idx,
                "vector": self.get_embedding(doc),
                "payload": {"text": doc}
            }
            for idx, doc in zip(ids, documents)
        ]
        self.client.upsert(self.collection_name, points)

    def search(self, query: str, limit: int = 5):
        query_vector = self.get_embedding(query)
        results = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_vector,
            limit=limit
        )
        return results

    def get_embedding(self, text: str):
        # 임베딩 모델 구현 필요
        pass

5. 전체 RAG 파이프라인 구현


python
import asyncio
from typing import List, Dict, Any
import logging

class RAGPipeline:
    def __init__(self, embedding_model: str, vector_db: Any, llm: Any):
        self.embedding_model = SentenceTransformer(embedding_model)
        self.vector_db = vector_db
        self.llm = llm
        self.chunker = SemanticChunker()
        self.logger = logging.getLogger(__name__)

    async def build_index(self, documents: List[Dict[str, str]]):
        """문서 인덱싱"""
        all_chunks = []
        ids = []

        for doc in documents:
            chunks = self.chunker.chunk_text(doc['content'])
            for i, chunk in enumerate(chunks):
                all_chunks.append(chunk)
                ids.append(f"{doc['id']}_{i}")

        # 임베딩 생성
        embeddings = self.embedding_model.encode(all_chunks)

        # 벡터 DB에 저장
        self.vector_db.add_documents(all_chunks, ids)
        self.logger.info(f"인덱싱 완료: {len(all_chunks)}개 청크")

    async def retrieve_and_generate(self, query: str, top_k: int = 5) -> Dict[str, Any]:
        """검색 및 생성"""
        # 검색
        results = self.vector_db.search(query, top_k)

        # 검색 결과 정리
        retrieved_docs = [doc['text'] for doc in results['documents'][0]]

        # 프롬프트 생성
        context = "\n".join(retrieved_docs)
        prompt = f"""
        다음 문맥을 기반으로 질문에 답하세요:
        문맥: {context}
        질문: {query}
        답변:
        """

        # 생성
        response = self.llm.generate(prompt)

        return {
            'query': query,
            'retrieved_docs': retrieved_docs,
            'answer': response,
            'confidence': self.calculate_confidence(results)
        }

    def calculate_confidence(self, search_results: Dict) -> float:
        # 유사도 기반 신뢰도 계산
        if 'distances' in search_results

---

📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($7)

로컬 LLM 셋업 가이드 (v36)

matias yoon — Mon, 25 May 2026 17:55:27 +0000

로컬 LLM 셋업 가이드 (v36)

1. 개요 및 사전 준비

로컬 LLM(대형 언어 모델)은 데이터 프라이버시, 비용 절감, 실시간 처리가 필요한 개발자들에게 이상적인 선택입니다. 이 가이드는 최소한의 리소스로도 동작하는 로컬 LLM 셋업을 제공합니다.

사전 요구사항:

OS: Ubuntu 20.04 이상
GPU: NVIDIA GTX 1060 이상 (CUDA 11.8 필요)
RAM: 최소 16GB (24GB 이상 권장)
저장공간: 최소 50GB (모델별로 2-10GB 필요)

2. 프레임워크 비교

프레임워크	장점	단점	추천 사용 사례
llama.cpp	가볍고, C++ 기반, 최대한의 커스터마이징 가능	복잡한 설정 필요, API 부족	고급 사용자, 성능 튜닝
Ollama	간단한 CLI, 모델 관리 용이	제한된 API 기능, 배포 복잡	빠른 실험, 개발용
vLLM	초고속 추론, 멀티 GPU 지원	복잡한 설치, 메모리 요구량 큼	프로덕션, 대량 처리
LocalAI	API 중심, 다양한 모델 지원	리소스 소모 큼	API 통합이 필요한 경우

3. 추천 설정: llama.cpp + Ollama

llama.cpp로 실행 가능한 최적화된 모델 + Ollama로 간단한 API 관리

llama.cpp 설치:

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

Ollama 설치:

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

4. 모델 선택 가이드

모델	크기	추천 사용 사례	추론 속도
Qwen-1.5B	1.5GB	기본 테스트, API 실험	빠름
Gemma-2B	2GB	실시간 챗봇	빠름
LLaMA3-8B	8GB	정확도가 중요한 작업	보통
Mistral-7B	7GB	복잡한 프롬프트	보통

5. 양자화 유형 설명

유형	품질	크기	사용 사례
Q4_K_M	매우 높음	4.5GB	높은 정확도 필요
Q5_K_M	높음	5.5GB	중간 정확도
Q6_K	보통	6.5GB	빠른 속도 필요
Q8_0	낮음	8GB	최소한의 정확도

6. API 설정

llama.cpp API 서버 실행:

# 모델 다운로드
mkdir -p models
wget https://huggingface.co/TheBloke/Qwen-1.5B-Chat-GGUF/resolve/main/qwen-1.5b-chat.Q4_K_M.gguf -P models/

# 서버 실행
./server -m models/qwen-1.5b-chat.Q4_K_M.gguf -c 2048 --port 8080

Ollama API 사용:

# 모델 푸시
ollama push qwen:1.5b

# API 호출
curl http://localhost:11434/api/generate \
  -H "Content-Type: application/json" \
  -d '{
    "model": "qwen:1.5b",
    "prompt": "Hello, how are you?",
    "stream": false
  }'

7. Systemd 서비스 설정

sudo nano /etc/systemd/system/llama.service

[Unit]
Description=Local LLM Server
After=network.target

[Service]
Type=simple
User=your_user
WorkingDirectory=/home/your_user/llama.cpp
ExecStart=/home/your_user/llama.cpp/server -m /home/your_user/models/qwen-1.5b-chat.Q4_K_M.gguf -c 2048 --port 8080
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable llama.service
sudo systemctl start llama.service

8. 모니터링 및 성능 조정

성능 측정:

# 추론 시간 측정
time curl -X POST http://localhost:8080/completion \
  -H "Content-Type: application/json" \
  -d '{"prompt":"Hello world", "n_predict":50}'

메모리 사용량 모니터링:

# 실시간 CPU/MEM 사용량
htop

# 로그 확인
journalctl -u llama.service -f

9. 실제 사용 예제

1. 개발 환경에서의 빠른 테스트:

# 100MB 모델로 간단한 테스트
ollama run qwen:1.5b

2. API 통합:

import requests

def call_local_llm(prompt):
    response = requests.post('http://localhost:8080/completion', 
                          json={'prompt': prompt, 'n_predict': 100})
    return response.json()['content']

# 사용 예제
result = call_local_llm("Python으로 간단한 웹 서버를 만들어줘")
print(result)

3. 고급 성능 설정:

# CPU 코어 제한
taskset -c 0-3 ./server -m models/qwen-1.5b-chat.Q4_K_M.gguf -c 2048 --port 8080

# 메모리 최적화
export CUDA_LAUNCH_BLOCKING=1
export CUDA_DEVICE_MAX_CONNECTIONS=1

10. 추천 최적화 팁

모델 선택: 정확도가 중요한 작업은 Q5_K_M 이상 사용
GPU 메모리: 16GB RAM 기준 8GB 모델 추천
API 설정: 8080 포트로 기본 설정 유지
로그 관리: Systemd 로그 자동 회전 설정

11. 문제 해결

일반적인 오류:

# 메모리 부족
error: failed to allocate memory

# 해결: 메모리 제한 해제
ulimit -v unlimited

# 포트 충돌
bind: Address already in use

# 해결: 포트 변경
./server --port 8081

이 가이드를 따라하면 20분 안에 로컬 LLM 환경을 구축하고, API를 통해 사용할 수 있습니다. 다양한 모델과 성능 조정 방법을 익혀두면 실제 프로젝트에 바로 적용할 수 있습니다.

📥 Get the full guide on Gumroad: https://gumroad.com/l/auto ($7)

LangGraph 워크플로우 템플릿 (v36)

matias yoon — Mon, 25 May 2026 17:40:42 +0000

LangGraph 워크플로우 템플릿 (v36)

1. LangGraph 아키텍처 개요

LangGraph는 상태 기반 워크플로우 엔진으로, 다음과 같은 핵심 구성 요소로 작동합니다:

Nodes (노드): 각 단계의 처리 로직
Edges (엣지): 노드 간의 전이 조건
State (상태): 워크플로우 내 모든 정보 저장
Checkpointing (체크포인팅): 상태 저장/복원 기능

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
import operator

class GraphState(TypedDict):
    messages: Annotated[list, operator.add]

# 기본 워크플로우 구조
workflow = StateGraph(GraphState)

2. 템플릿 1: 간단한 RAG 에이전트 (검색 → 생성 → 검증)

실제 문서 검색 후 답변 생성 시스템:

from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser

class RAGAgent:
    def __init__(self, vectorstore, llm):
        self.vectorstore = vectorstore
        self.llm = llm

    def retrieve(self, state):
        question = state["messages"][-1]["content"]
        docs = self.vectorstore.similarity_search(question, k=3)
        return {"retrieved_docs": docs}

    def generate(self, state):
        context = "\n".join([doc.page_content for doc in state["retrieved_docs"]])
        prompt = PromptTemplate.from_template(
            "다음 문맥을 기반으로 질문에 답변하세요:\n{context}\n질문: {question}"
        )
        chain = prompt | self.llm | StrOutputParser()
        answer = chain.invoke({
            "context": context,
            "question": state["messages"][-1]["content"]
        })
        return {"answer": answer}

    def validate(self, state):
        # 답변 검증 로직
        return {"validated": True}

    def build_graph(self):
        workflow = StateGraph(GraphState)

        workflow.add_node("retrieve", self.retrieve)
        workflow.add_node("generate", self.generate)
        workflow.add_node("validate", self.validate)

        workflow.set_entry_point("retrieve")
        workflow.add_edge("retrieve", "generate")
        workflow.add_edge("generate", "validate")
        workflow.add_edge("validate", END)

        return workflow.compile()

# 사용 예시
rag_agent = RAGAgent(vectorstore, ChatOpenAI(model="gpt-4"))
graph = rag_agent.build_graph()

3. 템플릿 2: 다중 도구 에이전트 (계획 → 실행 → 관찰 → 결정)

다양한 도구를 활용한 에이전트 시스템:

from langchain.tools import Tool
from langchain_core.tools import tool
import json

class ToolAgent:
    def __init__(self, tools):
        self.tools = tools

    def plan(self, state):
        # 계획 생성
        return {"plan": ["tool1", "tool2"]}

    def execute(self, state):
        # 도구 실행
        executed_tools = []
        for tool_name in state["plan"]:
            tool = next((t for t in self.tools if t.name == tool_name), None)
            if tool:
                result = tool.run(state["messages"][-1]["content"])
                executed_tools.append({"tool": tool_name, "result": result})
        return {"executed_tools": executed_tools}

    def observe(self, state):
        # 실행 결과 관찰
        return {"observations": "실행 완료"}

    def decide(self, state):
        # 결정 로직
        return {"final_decision": "continue"}

    def build_graph(self):
        workflow = StateGraph(GraphState)

        workflow.add_node("plan", self.plan)
        workflow.add_node("execute", self.execute)
        workflow.add_node("observe", self.observe)
        workflow.add_node("decide", self.decide)

        workflow.set_entry_point("plan")
        workflow.add_edge("plan", "execute")
        workflow.add_edge("execute", "observe")
        workflow.add_edge("observe", "decide")
        workflow.add_edge("decide", END)

        return workflow.compile()

# 도구 정의 예시
tools = [
    Tool(
        name="search",
        func=lambda x: f"검색 결과: {x}",
        description="검색 도구"
    ),
    Tool(
        name="calculate",
        func=lambda x: f"계산 결과: {eval(x)}",
        description="계산 도구"
    )
]

4. 템플릿 3: 인간-중개 워크플로우 (중지 → 검토 → 계속)

사용자 검토를 포함한 작업 흐름:

import asyncio
from datetime import datetime

class HumanInLoopAgent:
    def __init__(self, llm):
        self.llm = llm

    def process(self, state):
        # 기본 처리
        return {"processed_data": "처리된 데이터"}

    def pause_for_review(self, state):
        # 중지 및 검토
        return {"status": "awaiting_review"}

    def review(self, state):
        # 검토 상태
        return {"review_status": "approved"}

    def continue_processing(self, state):
        # 계속 진행
        return {"status": "processing"}

    def build_graph(self):
        workflow = StateGraph(GraphState)

        workflow.add_node("process", self.process)
        workflow.add_node("pause", self.pause_for_review)
        workflow.add_node("review", self.review)
        workflow.add_node("continue", self.continue_processing)

        workflow.set_entry_point("process")
        workflow.add_edge("process", "pause")
        workflow.add_edge("pause", "review")
        workflow.add_edge("review", "continue")
        workflow.add_edge("continue", END)

        return workflow.compile()

# 인간 검토를 위한 특별 처리
def human_review_callback(state):
    # 외부 시스템과 통신
    print("사용자 검토 요청:", state["messages"][-1]["content"])
    # 실제 구현에서는 API 호출이나 이메일 알림 필요
    return {"review_requested": True}

5. 템플릿 5: 병렬 실행 에이전트 (분기 → 처리 → 집계)

병렬 작업 처리 및 결과 집계:

class ParallelAgent:
    def __init__(self, llm):
        self.llm = llm

    def fan_out(self, state):
        # 작업 분기
        return {"tasks": ["task1", "task2", "task3"]}

    def process_parallel(self, state):
        # 병렬 처리
        tasks = state["tasks"]
        results = []
        for task in tasks:
            # 각 작업 병렬 처리
            result = f"결과: {task}"
            results.append({"task": task, "result": result})
        return {"results": results}

    def aggregate(self, state):
        # 결과 집계
        aggregated = "\n".join([r["result"] for r in state["results"]])
        return {"aggregated_result": aggregated}

    def build_graph(self):
        workflow = StateGraph(GraphState)

        workflow.add_node("fan_out", self.fan_out)
        workflow.add_node("process_parallel", self.process_parallel)
        workflow.add_node("aggregate", self.aggregate)

        workflow.set_entry_point("fan_out")
        workflow.add_edge("fan_out", "process_parallel")
        workflow.add_edge("process_parallel", "aggregate")
        workflow.add_edge("aggregate", END)

        return workflow.compile()

6. 상태 관리 패턴

상태를 효율적으로 관리하고 저장하는 방법:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.runnables import RunnableConfig

class StateManager:
    def __init__(self, checkpoint_backend=None):
        self.checkpoint_backend = checkpoint_backend or MemorySaver()

    def create_workflow_with_checkpoint(self, workflow_builder):
        return workflow_builder.with_config(
            configurable={
                "thread_id": "12345"
            }
        )

    def get_state(self, graph, thread_id):
        # 현재 상태 조회
        return graph.get_state({"configurable": {"thread_id": thread_id}})

    def update_state(self, graph, thread_id, update_data):
        # 상태 업데이트
        return graph.update_state(
            {"configurable": {"thread_id": thread_id}},
            update_data
        )

# 사용 예시
state_manager = StateManager(SqliteSaver("checkpoints.db"))

7. 스트리밍 및 실시간 업데이트

실시간 데이터 스트리밍 처리:


python
import asyncio
from typing import AsyncGenerator

class StreamingAgent:
    async def stream_process(self, state):
        # 스트리밍 처리
        messages = state

---

📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)

터미널 AI 에이전트 구축 (v35)

matias yoon — Mon, 25 May 2026 17:25:54 +0000

터미널 AI 에이전트 구축 (v35)

터미널에서 작동하는 AI 에이전트를 직접 구축하여 개발 생산성을 높이는 방법을 안내합니다. 이 가이드는 로컬에서 실행 가능한 고성능 AI 에이전트를 구축하는 실용적인 접근법을 제공합니다.

1. CLI AI 에이전트 생태계

현재 터미널 AI 에이전트 시장은 다음과 같은 주요 플랫폼으로 구성되어 있습니다:

주요 도구 비교

Aider:

pip install aider
aider --help

Continue.dev:

npm install -g @continue_dev/cli

OpenCode:

git clone https://github.com/open-code/open-code.git
cd open-code && pip install -r requirements.txt

이 중 OpenCode가 가장 간단하고 커스터마이징이 용이합니다.

2. 로컬 LLM API 엔드포인트 설정

로컬 LLM을 위한 API 엔드포인트를 설정합니다:

# llama.cpp 설치
git clone https://github.com/ggerganov/llama.cpp.git
cd llama.cpp
make

# 모델 다운로드 및 변환
wget https://huggingface.co/TheBloke/Mistral-7B-v0.1-GGUF/resolve/main/mistral-7b-v0.1.Q4_K_M.gguf
./llama.cpp/llama-server --model mistral-7b-v0.1.Q4_K_M.gguf --port 8080

API 테스트:

curl -X POST http://localhost:8080/completion \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "Write a Python function to calculate Fibonacci numbers",
    "max_tokens": 100
  }'

3. 간단한 Python CLI 에이전트 구축

# ai_agent.py
import openai
import json
import subprocess
import os
from typing import Dict, List, Any

class TerminalAgent:
    def __init__(self, model="http://localhost:8080"):
        self.model = model
        self.functions = {
            "run_command": self.run_command,
            "read_file": self.read_file,
            "write_file": self.write_file,
            "search_code": self.search_code
        }

    def run_command(self, command: str) -> str:
        """명령어 실행"""
        try:
            result = subprocess.run(
                command, shell=True, capture_output=True, text=True
            )
            return result.stdout + result.stderr
        except Exception as e:
            return f"Error: {str(e)}"

    def read_file(self, filepath: str) -> str:
        """파일 읽기"""
        try:
            with open(filepath, 'r') as f:
                return f.read()
        except Exception as e:
            return f"Error reading file: {str(e)}"

    def write_file(self, filepath: str, content: str) -> str:
        """파일 쓰기"""
        try:
            with open(filepath, 'w') as f:
                f.write(content)
            return f"File written to {filepath}"
        except Exception as e:
            return f"Error writing file: {str(e)}"

    def search_code(self, pattern: str, directory: str = ".") -> str:
        """코드 검색"""
        try:
            result = subprocess.run(
                f"find {directory} -type f -name '*.py' -exec grep -l '{pattern}' {{}} \\;",
                shell=True, capture_output=True, text=True
            )
            return result.stdout
        except Exception as e:
            return f"Search error: {str(e)}"

    def call_function(self, name: str, arguments: Dict[str, Any]) -> str:
        """함수 호출"""
        if name in self.functions:
            return self.functions[name](**arguments)
        return f"Unknown function: {name}"

    def process_request(self, prompt: str, tools: List[str] = None) -> str:
        """요청 처리"""
        # 툴 정보 생성
        tool_descriptions = []
        for tool in tools or []:
            if tool in self.functions:
                tool_descriptions.append({
                    "name": tool,
                    "description": f"Use {tool} for specific tasks",
                    "parameters": {}
                })

        # 실제 API 호출 (llama.cpp API)
        response = self._call_llm(prompt, tool_descriptions)
        return response

    def _call_llm(self, prompt: str, tools: List[Dict]) -> str:
        """LLM 호출"""
        # 실제 구현은 외부 API 호출
        return f"LLM response to: {prompt}"

if __name__ == "__main__":
    agent = TerminalAgent()
    print("Terminal AI Agent ready. Type 'quit' to exit.")

    while True:
        user_input = input(">>> ")
        if user_input.lower() == 'quit':
            break

        response = agent.process_request(user_input, ["run_command", "read_file"])
        print(response)

4. tmux와 통합

터미널 세션 관리를 위해 tmux 통합:

# tmux 세션 생성
tmux new-session -d -s ai_agent

# 세션에 명령어 실행
tmux send-keys -t ai_agent "python ai_agent.py" Enter

# 세션 정보 확인
tmux list-sessions

Python 스크립트에서 tmux 사용:

# tmux_integration.py
import subprocess
import os

class TmuxManager:
    def __init__(self, session_name: str = "ai_agent"):
        self.session_name = session_name

    def create_session(self):
        """세션 생성"""
        subprocess.run(["tmux", "new-session", "-d", "-s", self.session_name])

    def send_command(self, command: str):
        """세션에 명령어 전송"""
        subprocess.run([
            "tmux", "send-keys", "-t", self.session_name, command, "Enter"
        ])

    def get_output(self) -> str:
        """출력 가져오기"""
        result = subprocess.run([
            "tmux", "capture-pane", "-p", "-t", self.session_name
        ], capture_output=True, text=True)
        return result.stdout

# 사용 예제
tmux = TmuxManager()
tmux.create_session()
tmux.send_command("python ai_agent.py")

5. 커스텀 툴 개발

코드 검색 툴

# tools/code_search.py
import os
import re
from typing import List, Dict

class CodeSearchTool:
    def __init__(self, project_root: str = "."):
        self.project_root = project_root

    def search_patterns(self, patterns: List[str], file_extensions: List[str] = None) -> Dict[str, List[str]]:
        """패턴 검색"""
        if file_extensions is None:
            file_extensions = ['.py', '.js', '.cpp', '.java']

        results = {}
        for pattern in patterns:
            results[pattern] = []
            for root, dirs, files in os.walk(self.project_root):
                for file in files:
                    if any(file.endswith(ext) for ext in file_extensions):
                        filepath = os.path.join(root, file)
                        try:
                            with open(filepath, 'r') as f:
                                content = f.read()
                                matches = re.finditer(pattern, content)
                                for match in matches:
                                    results[pattern].append({
                                        'file': filepath,
                                        'line': content[:match.start()].count('\n') + 1,
                                        'context': self._get_context(content, match.start())
                                    })
                        except Exception:
                            continue
        return results

    def _get_context(self, content: str, position: int, context_lines: int = 3) -> str:
        """문맥 추출"""
        lines = content.split('\n')
        line_num = content[:position].count('\n')
        start = max(0, line_num - context_lines)
        end = min(len(lines), line_num + context_lines + 1)
        return '\n'.join(lines[start:end])

# 사용 예제
search_tool = CodeSearchTool("/path/to/project")
results = search_tool.search_patterns([r"def.*function"])

Git 통합 툴


python
# tools/git_tool.py
import subprocess
import json
from datetime import datetime

class GitTool:
    def __init__(self, repo_path: str = "."):
        self.repo_path = repo_path

    def get_commits(self, limit: int = 10) -> List[Dict]:
        """최근 커밋 목록"""
        cmd = [
            "git", "-C", self.repo_path,
            "log", "--oneline", f"--max-count={limit}",
            "--format=%H|%an|%ad|%s"
        ]
        result = subprocess.run(cmd, capture_output=True, text=True)

        commits = []
        for line in result.stdout.strip().split('\n'):
            if line:
                hash_, author, date, message = line.split

---

📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)

RAG 시스템 실전 구축 (v35)

matias yoon — Mon, 25 May 2026 17:11:07 +0000

RAG 시스템 실전 구축 (v35)

개요

RAG(Retrieval-Augmented Generation) 시스템은 대규모 언어 모델(LLM)의 정보 제한을 극복하기 위한 핵심 기술입니다. 이 가이드에서는 실제 개발 환경에서 사용할 수 있는 RAG 시스템을 구축하는 방법을 실전 중심으로 설명합니다.

1. RAG 기본 개념

RAG는 세 가지 핵심 단계로 구성됩니다:

검색(Retrieval): 사용자 쿼리와 유사한 문서를 검색
보강(Augmentation): 검색된 문서를 프롬프트에 포함
생성(Generation): LLM이 검색된 정보를 기반으로 답변 생성

# 기본 RAG 구조
class BasicRAG:
    def __init__(self, embedding_model, vector_db, llm):
        self.embedding_model = embedding_model
        self.vector_db = vector_db
        self.llm = llm

    def query(self, user_query):
        # 1. 임베딩 생성
        query_embedding = self.embedding_model.encode(user_query)

        # 2. 검색
        retrieved_docs = self.vector_db.search(query_embedding, k=5)

        # 3. 프롬프트 구성
        context = "\n".join([doc.content for doc in retrieved_docs])
        prompt = f"질문: {user_query}\n문맥: {context}"

        # 4. 생성
        response = self.llm.generate(prompt)
        return response

2. Chunking 전략

문서를 적절한 단위로 분할하는 것이 중요합니다:

2.1 의미 기반 Chunking

from sentence_transformers import SentenceTransformer
import numpy as np

class SemanticChunker:
    def __init__(self, model_name="all-MiniLM-L6-v2"):
        self.model = SentenceTransformer(model_name)

    def chunk_by_semantic(self, text, max_tokens=512):
        sentences = text.split('. ')
        embeddings = self.model.encode(sentences)

        chunks = []
        current_chunk = []
        current_length = 0

        for sentence, embedding in zip(sentences, embeddings):
            if current_length + len(sentence.split()) > max_tokens:
                chunks.append(' '.join(current_chunk))
                current_chunk = [sentence]
                current_length = len(sentence.split())
            else:
                current_chunk.append(sentence)
                current_length += len(sentence.split())

        if current_chunk:
            chunks.append(' '.join(current_chunk))
        return chunks

# 사용 예시
chunker = SemanticChunker()
chunks = chunker.chunk_by_semantic("대규모 언어 모델은... 매우 강력한 기술입니다.")

2.2 Recursive Chunking

class RecursiveChunker:
    def __init__(self, chunk_size=1024, chunk_overlap=128):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

    def chunk_recursive(self, text, separators=["\n\n", "\n", " ", ""]):
        chunks = []
        current_chunk = ""

        for separator in separators:
            if separator in text:
                parts = text.split(separator)
                for i, part in enumerate(parts):
                    if len(part.strip()) == 0:
                        continue
                    if len(current_chunk) + len(part) <= self.chunk_size:
                        current_chunk += part + separator
                    else:
                        if current_chunk:
                            chunks.append(current_chunk.strip())
                        current_chunk = part + separator
                break
        else:
            # 마지막 단계: 기본 분할
            for i in range(0, len(text), self.chunk_size):
                chunk = text[i:i + self.chunk_size]
                if len(chunk) >= self.chunk_size * 0.5:
                    chunks.append(chunk)

        if current_chunk and len(current_chunk.strip()) > 0:
            chunks.append(current_chunk.strip())

        return chunks

# 사용 예시
recursive_chunker = RecursiveChunker()
chunks = recursive_chunker.chunk_recursive("내용을 분할하는 데 사용됩니다.")

3. Embedding 모델 선택과 비교

from sentence_transformers import SentenceTransformer
import torch
import time

class EmbeddingBenchmark:
    def __init__(self):
        self.models = {
            "all-MiniLM-L6-v2": SentenceTransformer("all-MiniLM-L6-v2"),
            "all-mpnet-base-v2": SentenceTransformer("all-mpnet-base-v2"),
            "sentence-t5-xxl": SentenceTransformer("sentence-t5-xxl"),
            "bge-large-en": SentenceTransformer("BAAI/bge-large-en")
        }

    def benchmark_embeddings(self, texts):
        results = {}
        for name, model in self.models.items():
            start_time = time.time()
            embeddings = model.encode(texts)
            end_time = time.time()

            results[name] = {
                "latency": end_time - start_time,
                "size": len(embeddings),
                "dimension": len(embeddings[0]) if len(embeddings) > 0 else 0
            }
        return results

# 사용 예시
benchmark = EmbeddingBenchmark()
texts = ["예시 문장 1", "예시 문장 2", "예시 문장 3"]
results = benchmark.benchmark_embeddings(texts)
for model, metrics in results.items():
    print(f"{model}: {metrics['latency']:.2f}s, {metrics['dimension']}차원")

4. Vector Database 비교

4.1 Chroma

import chromadb
from chromadb.config import Settings

class ChromaVectorDB:
    def __init__(self, collection_name="rag_collection"):
        self.client = chromadb.Client(Settings(chroma_db_impl="duckdb"))
        self.collection = self.client.get_or_create_collection(collection_name)

    def add_documents(self, documents, embeddings, ids):
        self.collection.add(
            documents=documents,
            embeddings=embeddings,
            ids=ids
        )

    def search(self, query_embedding, k=5):
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=k
        )
        return results["documents"][0], results["ids"][0]

4.2 Qdrant

from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue
import numpy as np

class QdrantVectorDB:
    def __init__(self, host="localhost", port=6333, collection_name="rag_collection"):
        self.client = QdrantClient(host=host, port=port)
        self.collection_name = collection_name

        # 컬렉션 생성
        self.client.recreate_collection(
            collection_name=collection_name,
            vectors_config={"size": 384, "distance": "Cosine"}
        )

    def add_documents(self, documents, embeddings, ids):
        points = [
            {
                "id": id,
                "vector": embedding,
                "payload": {"content": doc}
            }
            for id, doc, embedding in zip(ids, documents, embeddings)
        ]
        self.client.upsert(
            collection_name=self.collection_name,
            points=points
        )

    def search(self, query_embedding, k=5):
        results = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_embedding,
            limit=k
        )
        return [point.payload["content"] for point in results], [point.id for point in results]

4.3 pgvector

import psycopg2
from psycopg2.extras import Json
import numpy as np

class PGVectorDB:
    def __init__(self, connection_string):
        self.conn = psycopg2.connect(connection_string)
        self.create_table()

    def create_table(self):
        with self.conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS rag_documents (
                    id UUID PRIMARY KEY,
                    content TEXT,
                    embedding VECTOR(384),
                    metadata JSONB
                )
            """)
            cur.execute("""
                CREATE INDEX IF NOT EXISTS idx_embedding ON rag_documents USING ivfflat (embedding vector_cosine_ops)
            """)
        self.conn.commit()

    def add_documents(self, documents, embeddings, ids):
        with self.conn.cursor() as cur:
            for id, content, embedding in zip(ids, documents, embeddings):
                cur.execute(
                    "INSERT INTO rag_documents (id, content, embedding) VALUES (%s, %s, %s)",
                    (id, content, embedding.tolist())
                )
        self.conn.commit()

    def search(self, query_embedding, k=5):
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT content FROM rag_documents 
                ORDER BY embedding <-> %s 
                LIMIT %s
            """, (query_embedding.tolist(), k))
            results = cur.fetchall()
        return [result[0] for result in results], None  # ID는 반환하지 않음

5. 완전한 RAG 파이프라인 구현


python
import os
import uuid
from typing import List, Tuple
from sentence_transformers import SentenceTransformer
import chromadb
from chromadb.config import Settings
from langchain_open

---

📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($7)

로컬 LLM 셋업 가이드 (v35)

matias yoon — Mon, 25 May 2026 16:56:20 +0000

로컬 LLM 셋업 가이드 (v35)

1. 개요 및 사전 준비

로컬 LLM (Large Language Model) 실행은 AI 개발자들에게 광범위한 유연성과 데이터 프라이버시를 제공합니다. 이 가이드는 Linux 환경에서 LLM을 실행하고 관리하는 실용적인 방법을 다룹니다.

하드웨어 요구사항:

CPU: 최소 8코어 이상, 권장 16코어 이상
RAM: 최소 32GB, 권장 64GB 이상
GPU: NVIDIA GPU (CUDA 지원), 8GB 이상 VRAM (권장 16GB 이상)
저장공간: 최소 100GB (모델 다운로드 및 캐시 포함)

사전 설치:

# Ubuntu/Debian
sudo apt update
sudo apt install -y git cmake build-essential python3 python3-pip

# CentOS/RHEL
sudo yum install -y git cmake gcc gcc-c++ python3 python3-pip

2. 프레임워크 비교

프레임워크	장점	단점	적합한 경우
llama.cpp	최소 종속성, 직접 실행 가능	기능 제한, 메모리 사용량 높음	간단한 테스트, 단일 모델 실행
Ollama	사용자 친화적, 빠른 설치	추가 런타임 필요	개발 및 테스트 환경
vLLM	높은 성능, 멀티모델 처리	복잡한 설치 과정	프로덕션 환경
LocalAI	다양한 API 호환성, 확장성	리소스 소모 많음	기업 환경

3. 추천 설정: llama.cpp 기반 설치

llama.cpp는 최소 종속성과 높은 호환성을 제공하여 최적의 선택입니다.

# 소스 코드 클론
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp

# 빌드
make clean
make

# 기본 실행 확인
./main -h

4. 모델 선택 가이드

모델 유형별 사용 사례:

LLaMA 2 7B: 일반적인 자연어 처리, 개발 테스트
LLaMA 2 13B: 고성능 요구 사항, 엔터프라이즈 환경
Mistral 7B: 높은 효율성과 빠른 응답 시간
Mixtral 8x7B: 복잡한 다국어 처리

# 예제 모델 다운로드 (LLaMA 2 7B)
wget https://huggingface.co/llama/llama-2-7b/resolve/main/llama-2-7b.bin -O models/llama-2-7b.bin

5. 양자화 유형 설명

Q4_K_M: 4비트 양자화, 최적의 성능/정확도 균형
Q5_K_M: 5비트 양자화, 정확도 향상
Q8_0: 8비트 양자화, 높은 정확도

# Q4_K_M 양자화
./quantize models/llama-2-7b.bin models/llama-2-7b-q4_k_m.bin Q4_K_M

6. API 설정 및 기존 도구 통합

llama.cpp는 기본적으로 REST API를 제공하지 않지만, 다음 방법으로 API를 활성화합니다:

# 기본 서버 실행
./main -m models/llama-2-7b-q4_k_m.bin -p "Hello, how are you?" --port 8080

# 서버 실행 (API 모드)
./main -m models/llama-2-7b-q4_k_m.bin -a "Hello, how are you?" --port 8080 --host 0.0.0.0

사용 가능한 API 엔드포인트:

POST /completion - 텍스트 생성
GET /health - 상태 확인
POST /generate - 스트리밍 생성

7. Systemd 서비스 설정

24/7 운영을 위한 systemd 서비스 설정:

# 서비스 파일 생성
sudo nano /etc/systemd/system/llama.service

[Unit]
Description=Local LLM Server
After=network.target

[Service]
Type=simple
User=developer
WorkingDirectory=/home/developer/llama.cpp
ExecStart=/home/developer/llama.cpp/main -m /home/developer/models/llama-2-7b-q4_k_m.bin --port 8080 --host 0.0.0.0
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

# 서비스 활성화 및 시작
sudo systemctl daemon-reload
sudo systemctl enable llama.service
sudo systemctl start llama.service
sudo systemctl status llama.service

8. 모니터링 및 성능 조정

성능 모니터링:

# GPU 사용량 확인
nvidia-smi

# 시스템 리소스 모니터링
htop
iotop

최적화 설정:

# 메모리 최적화
./main -m models/llama-2-7b-q4_k_m.bin --threads 8 --ctx-size 2048 --batch-size 512

# 성능 테스트
./main -m models/llama-2-7b-q4_k_m.bin -p "This is a test prompt" --repeat-penalty 1.1

9. 실용적 명령어 예제

모델 다운로드 및 준비:

# 모델 디렉토리 생성
mkdir -p models

# 모델 다운로드
wget https://huggingface.co/llama/llama-2-7b/resolve/main/llama-2-7b.bin -O models/llama-2-7b.bin

# 양자화
./quantize models/llama-2-7b.bin models/llama-2-7b-q4_k_m.bin Q4_K_M

서버 실행 및 테스트:

# 서버 실행
./main -m models/llama-2-7b-q4_k_m.bin --port 8080 --host 0.0.0.0

# API 테스트
curl -X POST http://localhost:8080/completion \
  -H "Content-Type: application/json" \
  -d '{"prompt":"Explain quantum computing in simple terms","max_tokens":100}'

성능 벤치마크:

# 응답 시간 측정
time curl -s -X POST http://localhost:8080/completion \
  -H "Content-Type: application/json" \
  -d '{"prompt":"What is machine learning?","max_tokens":50}' > /dev/null

# Throughput 테스트
ab -n 100 -c 10 http://localhost:8080/completion

자동 재시작 설정:

# 자동 재시작 스크립트
cat > ~/llama-restart.sh << 'EOF'
#!/bin/bash
if ! pgrep -f "main.*llama.*" > /dev/null; then
    echo "$(date): LLM server not running. Restarting..."
    systemctl restart llama.service
fi
EOF

chmod +x ~/llama-restart.sh

이 가이드는 실질적인 로컬 LLM 셋업을 위한 완전한 지침을 제공합니다. 이 설정은 개발 및 테스트 환경에 적합하며, 필요한 경우 확장하여 프로덕션 환경에서도 사용할 수 있습니다.

📥 Get the full guide on Gumroad: https://gumroad.com/l/auto ($7)