DEV Community: matias yoon

RAG 시스템 실전 구축 (v6)

matias yoon — Sun, 24 May 2026 12:27:51 +0000

RAG 시스템 실전 구축 (v6)

1. RAG 기초 개념

Retrieval-Augmented Generation (RAG)은 대규모 언어 모델(LLM)의 지식 범위를 확장하는 효율적인 방법입니다. RAG는 다음 세 가지 주요 단계로 구성됩니다:

검색 (Retrieval): 질문과 관련된 문서 조각을 벡터 데이터베이스에서 검색합니다.
증강 (Augmentation): 검색된 문서를 프롬프트에 포함하여 LLM의 입력을 향상시킵니다.
생성 (Generation): 증강된 입력을 기반으로 답변을 생성합니다.

# 간단한 RAG 루프 구현
class SimpleRAG:
    def __init__(self, embedding_model, vector_db):
        self.embedding_model = embedding_model
        self.vector_db = vector_db

    def query(self, question):
        # 1. 질문 임베딩
        query_embedding = self.embedding_model.encode(question)

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

        # 3. 증강 및 생성
        context = "\n".join([doc.content for doc in relevant_docs])
        prompt = f"질문: {question}\n문맥: {context}"

        return self.generate_answer(prompt)

2. 청킹 전략

2.1 의미적 청킹 (Semantic 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_semantic(self, text, max_tokens=512):
        sentences = text.split('. ')
        embeddings = self.model.encode(sentences)

        # 문장 간 유사도 기반 그룹화
        chunks = []
        current_chunk = []
        current_embedding = None

        for i, (sentence, embedding) in enumerate(zip(sentences, embeddings)):
            if not current_chunk:
                current_chunk.append(sentence)
                current_embedding = embedding
            else:
                # 유사도 계산
                similarity = np.dot(current_embedding, embedding) / (
                    np.linalg.norm(current_embedding) * np.linalg.norm(embedding)
                )

                if similarity > 0.8 and len(current_chunk) < 10:
                    current_chunk.append(sentence)
                else:
                    chunks.append('. '.join(current_chunk))
                    current_chunk = [sentence]
                    current_embedding = embedding

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

        return chunks

2.2 재귀적 청킹 (Recursive Chunking)

문서를 여러 레벨로 분할하여 다양한 청킹 수준을 제공합니다.

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

    def chunk_recursive(self, text):
        if len(text) <= self.chunk_size:
            return [text]

        chunks = []
        start = 0

        while start < len(text):
            end = min(start + self.chunk_size, len(text))

            # 오버랩 처리
            if start > 0:
                overlap_start = max(0, start - self.overlap)
                chunk_content = text[overlap_start:end]
            else:
                chunk_content = text[start:end]

            chunks.append(chunk_content)
            start = end - self.overlap

        return chunks

2.3 에이전트 기반 청킹 (Agentic Chunking)

문서의 구조적 특징을 고려한 청킹입니다.

import re

class AgenticChunker:
    def __init__(self):
        self.section_patterns = [
            r'##\s+(.+)',
            r'#\s+(.+)',
            r'\*\*\s+(.+)\s+\*\*'
        ]

    def chunk_by_structure(self, text):
        # 섹션 기준으로 분할
        sections = re.split(r'(\n## |\n# )', text)

        chunks = []
        current_chunk = ""

        for i, part in enumerate(sections):
            if i % 2 == 1:  # 섹션 헤더
                if current_chunk:
                    chunks.append(current_chunk.strip())
                current_chunk = part
            else:  # 섹션 내용
                current_chunk += part

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

        return chunks

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

3.1 모델 비교

from sentence_transformers import SentenceTransformer
from transformers import AutoTokenizer, AutoModel
import torch

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")
        }

    def compare_models(self, texts, benchmark_dataset):
        results = {}

        for model_name, model in self.models.items():
            # 성능 측정
            start_time = time.time()
            embeddings = model.encode(texts)
            end_time = time.time()

            # 정확도 평가 (가상 데이터셋 사용)
            accuracy = self.evaluate_accuracy(embeddings, benchmark_dataset)

            results[model_name] = {
                "latency": end_time - start_time,
                "accuracy": accuracy,
                "size": model.get_sentence_embedding_dimension()
            }

        return results

# 사용 예시
benchmark = EmbeddingBenchmark()
texts = ["Python은 강력한 프로그래밍 언어입니다.", "Machine Learning은 AI의 한 분야입니다."]
results = benchmark.compare_models(texts, "benchmark_dataset")

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

4.1 Chroma vs Qdrant vs pgvector vs Milvus

# Chroma 구현
import chromadb
from chromadb import Client

class ChromaVectorDB:
    def __init__(self):
        self.client = Client()
        self.collection = self.client.get_or_create_collection("rag_docs")

    def add_documents(self, documents, embeddings):
        self.collection.add(
            documents=documents,
            embeddings=embeddings,
            ids=[f"doc_{i}" for i in range(len(documents))]
        )

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

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

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

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

    def search(self, query_embedding, k=5):
        results = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_embedding,
            limit=k
        )
        return [hit.payload["text"] for hit in 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 rag_documents (
                    id SERIAL PRIMARY KEY,
                    content TEXT,
                    embedding VECTOR(768)
                )
            """)
            self.conn.commit()

    def add_documents(self, documents, embeddings):
        with self.conn.cursor() as cur:
            for doc, embedding in zip(documents, embeddings):
                cur.execute(
                    "INSERT INTO rag_documents (content, embedding) VALUES (%s, %s)",
                    (doc, embedding)
                )
        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, k))
            return [row[0] for row in cur.fetchall()]

4.2 성능 비교


python
import time

def benchmark_vector_dbs(documents, embeddings, queries):
    dbs = {
        "Chroma": ChromaVectorDB(),
        "Qdrant": QdrantVectorDB(),
        "pgvector": PgVectorDB("postgresql://user:pass@

---

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

로컬 LLM 셋업 가이드 (v6)

matias yoon — Sun, 24 May 2026 12:12:57 +0000

로컬 LLM 셋업 가이드 (v6)

개요 및 사전 요구사항

로컬 LLM(대규모 언어 모델) 실행은 데이터 프라이버시와 성능을 동시에 고려할 수 있는 최적의 솔루션입니다. 이 가이드는 최소한의 하드웨어로도 실행 가능한 실용적인 로컬 LLM 셋업을 제공합니다.

요구사항:

OS: Linux (Ubuntu 22.04 이상 권장)
GPU: NVIDIA RTX 3060 이상 (필수 아님, CPU도 가능)
RAM: 16GB 이상 (추천 32GB)
저장소: SSD 50GB 이상 여유 공간

기초 명령어:

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

# 필요한 패키지 설치
sudo apt install git cmake build-essential python3-pip -y

# 디렉토리 생성
mkdir ~/llm && cd ~/llm

프레임워크 비교

프레임워크	장점	단점	추천 사용 사례
llama.cpp	최소 의존성, 직접 컴파일 가능	API 제한, 복잡한 설정	빠른 프로토타이핑
Ollama	쉬운 설치, 자동 모델 다운로드	API 제한, CPU 성능 저하	개발용, 빠른 테스트
vLLM	높은 성능, 멀티 GPU 지원	복잡한 설치, 높은 메모리 사용	실시간 API 서비스
LocalAI	다중 프레임워크 지원, 완전한 API 호환	CPU 성능 저하, 메모리 요구량 높음	엔터프라이즈 환경

추천 설정: llama.cpp + Ollama

우리는 llama.cpp를 기반으로 한 Ollama를 추천합니다. 이유는 다음과 같습니다:

빠른 설치와 실행
다양한 모델 지원
간단한 API 인터페이스
CPU/가속기 모두 지원

설치 단계:

# Ollama 설치
curl -fsSL https://ollama.com/install.sh | sh

# Ollama 서비스 시작
sudo systemctl start ollama
sudo systemctl enable ollama

# 서비스 상태 확인
systemctl status ollama

모델 선택 가이드

모델	용도	추천 사양	메모리 요구
llama3:8b	일반 텍스트 생성	CPU/GPU 8GB RAM	4-6GB
mistral:7b	코드 생성	CPU/GPU 16GB RAM	6-8GB
phi3:3b	빠른 응답	CPU 8GB RAM	2-3GB
gemma:2b	경량 작업	CPU 4GB RAM	1-2GB

모델 다운로드 예제:

# Ollama에서 모델 다운로드
ollama pull llama3:8b
ollama pull mistral:7b
ollama pull phi3:3b

# 모델 목록 확인
ollama list

양자화 유형 설명

Q4_K_M vs Q5_K_M

Q4_K_M: 4비트 양자화 (최대 15% 손실)
Q5_K_M: 5비트 양자화 (최대 10% 손실)

명령어:

# 모델을 Q4_K_M로 변환
ollama create mymodel -f Modelfile

# Modelfile 예제
FROM llama3:8b
PARAMETER temperature 0.7
PARAMETER top_p 0.9

API 설정 및 통합

기본 API 테스트:

# API 테스트
curl http://localhost:11434/api/generate \
  -d '{
    "model": "llama3:8b",
    "prompt": "Hello, how are you?",
    "stream": false
  }'

Python 연동 예제:

import requests
import json

def call_llm(prompt, model="llama3:8b"):
    response = requests.post(
        'http://localhost:11434/api/generate',
        json={
            'model': model,
            'prompt': prompt,
            'stream': False
        }
    )
    return response.json()['response']

# 사용 예시
result = call_llm("Python으로 1부터 10까지의 숫자를 출력하는 코드를 작성해줘")
print(result)

Systemd 서비스 설정

서비스 파일 생성:

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

서비스 파일 내용:

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

[Service]
Type=simple
User=your_user
WorkingDirectory=/home/your_user/llm
ExecStart=/usr/bin/ollama serve
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

서비스 시작:

# 서비스 등록
sudo systemctl daemon-reload
sudo systemctl enable llm-api
sudo systemctl start llm-api

# 상태 확인
sudo systemctl status llm-api

성능 모니터링

시스템 모니터링:

# CPU/메모리 사용량 확인
htop

# GPU 사용량 확인 (NVIDIA)
nvidia-smi

# 로그 확인
journalctl -u llm-api -f

벤치마크 테스트:

# 간단한 성능 테스트
ollama run llama3:8b "Write a short poem about technology"

실전 예제

웹 애플리케이션 연동:

# Flask 웹 서버 예제
pip install flask requests

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

app = Flask(__name__)

@app.route('/chat', methods=['POST'])
def chat():
    prompt = request.json.get('prompt')
    response = requests.post(
        'http://localhost:11434/api/generate',
        json={
            'model': 'llama3:8b',
            'prompt': prompt
        }
    )
    return jsonify(response.json())

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

자동화 스크립트:

#!/bin/bash
# auto-llm.sh

# 모델 업데이트 체크
ollama list

# 백그라운드 실행
nohup ollama serve > /dev/null 2>&1 &

# 로그 확인
tail -f /var/log/ollama.log

최적화 팁

모델 선택: 필요에 따라 적절한 크기의 모델 선택
메모리 관리: 불필요한 모델은 삭제
API 캐싱: 반복적인 요청을 캐싱하여 성능 향상

결론

이 가이드는 16GB RAM과 기본 GPU 환경에서 로컬 LLM을 쉽게 설정할 수 있도록 합니다. Ollama를 기반으로 한 설정은 빠르고 안정적인 환경을 제공하며, 다양한 모델을 쉽게 관리할 수 있습니다.

추천 구매 제품: Local LLM Orchestrator

가격: $3.99

기능: 모델 관리, API 통합, 성능 모니터링, 자동 업데이트

이 가이드를 따라하면 누구나 로컬 LLM을 쉽게 설정하고 운영할 수 있습니다.

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

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

matias yoon — Sun, 24 May 2026 11:58:03 +0000

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

LangGraph는 현대 AI 에이전트 개발을 위한 강력한 프레임워크로, 상태 기반 워크플로우를 구현하는 데 최적화되어 있습니다. 이 가이드에서는 실제 개발자들이 자주 직면하는 문제를 해결하는 5가지 핵심 템플릿을 제공합니다.

1. LangGraph 아키텍처 개요

LangGraph는 다음과 같은 핵심 구성 요소로 구성됩니다:

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

# 상태 정의
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    # 기타 상태 필드

# 노드 정의
def retrieve_node(state):
    # 문서 검색 로직
    pass

def generate_node(state):
    # 생성 로직
    pass

# 엣지 정의
def route_to_next(state):
    # 다음 노드 결정 로직
    pass

# 워크플로우 정의
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)
workflow.add_edge("retrieve", "generate")
workflow.add_conditional_edges("generate", route_to_next)
workflow.set_entry_point("retrieve")
workflow.set_finish_point(END)

# 체크포인팅 설정
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

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

이 템플릿은 문서 검색과 생성을 결합한 기본적인 RAG 워크플로우입니다.

from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
import operator
from typing import Annotated, List
from langgraph.graph import StateGraph, END

class RAGState(TypedDict):
    query: str
    retrieved_docs: List[str]
    generated_response: str
    validation_result: bool

# 검색 노드
def retrieve_node(state: RAGState):
    # Chroma에서 문서 검색
    vectorstore = Chroma(
        persist_directory="./chroma_db",
        embedding_function=OpenAIEmbeddings()
    )

    docs = vectorstore.similarity_search(state["query"], k=3)
    return {"retrieved_docs": [doc.page_content for doc in docs]}

# 생성 노드
def generate_node(state: RAGState):
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a helpful assistant that answers questions based on provided context."),
        ("human", """Context: {context}
Question: {question}
Answer:""")
    ])

    model = ChatOpenAI(model="gpt-4")
    chain = prompt | model

    response = chain.invoke({
        "context": "\n".join(state["retrieved_docs"]),
        "question": state["query"]
    })

    return {"generated_response": response.content}

# 검증 노드
def validate_node(state: RAGState):
    # 생성된 응답 검증
    if len(state["generated_response"]) < 10:
        return {"validation_result": False}

    # 간단한 키워드 기반 검증
    keywords = ["answer", "question", "context", "response"]
    valid = any(keyword in state["generated_response"].lower() for keyword in keywords)
    return {"validation_result": valid}

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

    workflow.add_node("retrieve", retrieve_node)
    workflow.add_node("generate", generate_node)
    workflow.add_node("validate", validate_node)

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

    # 검증 결과에 따라 처리
    def route_validation(state: RAGState):
        if not state["validation_result"]:
            return "retry"
        return "end"

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

    workflow.set_entry_point("retrieve")
    return workflow.compile()

# 사용 예시
# app = create_rag_workflow()
# result = app.invoke({"query": "Python 디자인 패턴에 대해 설명해주세요"})

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

이 템플릿은 여러 도구를 활용할 수 있는 복잡한 에이전트를 위한 워크플로우입니다.

from typing import Literal
from langgraph.graph import StateGraph, END
from langchain_core.messages import ToolMessage

class ToolAgentState(TypedDict):
    messages: Annotated[list, operator.add]
    plan: List[str]
    current_tool: str
    tool_result: str

# 계획 노드
def plan_node(state: ToolAgentState):
    # 도구 사용 계획 생성
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a task planner. Analyze the following request and create a plan of actions."),
        ("human", "Request: {request}\nPlan:")

    ])

    model = ChatOpenAI(model="gpt-4")
    chain = prompt | model

    response = chain.invoke({"request": state["messages"][-1].content})

    return {"plan": response.content.split('\n')}

# 실행 노드
def execute_node(state: ToolAgentState):
    # 도구 실행
    tool_name = state["plan"][0]  # 첫 번째 계획 항목

    # 실제 도구 실행 로직 (예: 파일 시스템 조작)
    if tool_name == "file_search":
        # 파일 검색 로직
        result = "Found files matching criteria"
    elif tool_name == "code_analysis":
        # 코드 분석 로직
        result = "Code analysis complete"
    else:
        result = "Tool not recognized"

    return {"current_tool": tool_name, "tool_result": result}

# 관찰 노드
def observe_node(state: ToolAgentState):
    # 도구 결과 관찰
    messages = state["messages"]
    tool_message = ToolMessage(
        content=state["tool_result"],
        tool_call_id=state["current_tool"]
    )

    return {"messages": [tool_message]}

# 결정 노드
def decide_node(state: ToolAgentState):
    # 다음 단계 결정
    if len(state["plan"]) > 1:
        return "continue"
    return "end"

# 워크플로우 정의
def create_tool_workflow():
    workflow = StateGraph(ToolAgentState)

    workflow.add_node("plan", plan_node)
    workflow.add_node("execute", execute_node)
    workflow.add_node("observe", observe_node)
    workflow.add_node("decide", decide_node)

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

    def route_decision(state: ToolAgentState):
        if state["plan"] and len(state["plan"]) > 1:
            return "plan"
        return "end"

    workflow.add_conditional_edges(
        "decide",
        route_decision,
        {
            "plan": "plan",
            "end": END
        }
    )

    workflow.set_entry_point("plan")
    return workflow.compile()

# 사용 예시
# app = create_tool_workflow()
# result = app.invoke({"messages": [HumanMessage(content="파일 시스템에서 모든 Python 파일을 검색해주세요")]})

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

이 템플릿은 인간의 개입이 필요한 작업을 처리하기 위한 워크플로우입니다.


python
from langgraph.graph import StateGraph, END, START
from langchain_core.messages import AIMessage

class HumanInLoopState(TypedDict):
    messages: Annotated[list, operator.add]
    action_needed: bool
    user_feedback: str
    processed_result: str

# 작업 생성 노드
def create_task_node(state: HumanInLoopState):
    # 작업 생성 및 인간 개입 필요성 판단
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Generate task based on user request and determine if human review is needed."),
        ("human", "Request: {request}\nGenerate task:")
    ])

    model = ChatOpenAI(model="gpt-4")
    chain = prompt | model

    response = chain.invoke({"request": state["messages"][-1].content})

    # 간단한 규칙으로 인간 개입 필요성 판단
    needs_review = "review" in response.content.lower() or "approve" in response.content.lower()

    return {
        "messages": [AIMessage(content=response.content)],
        "action_needed":

---

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

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

matias yoon — Sun, 24 May 2026 11:43:08 +0000

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

터미널 기반 AI 에이전트는 개발자에게 매우 실용적인 도구로 자리 잡았습니다. 다양한 CLI 기반 AI 도구들 중에서 가장 효율적인 방식으로 개발자 워크플로우를 개선할 수 있는 방법을 소개합니다.

1. CLI AI 에이전트 생태계

현재 CLI AI 에이전트 시장은 다음과 같은 주요 도구들로 구성되어 있습니다:

Aider

pip install aider
aider --help

Continue.dev

npm install -g continue

OpenCode

git clone https://github.com/opencode-dev/opencode
cd opencode && pip install -e .

이들 도구들은 모두 자체적으로 파일 편집, 코드 생성, git 통합 기능을 제공하지만, 커스터마이징과 성능 최적화에 있어 한계가 있습니다.

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

로컬 LLM을 사용하면 보안성과 성능을 높일 수 있습니다:

Ollama 설치

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

# macOS
brew install ollama

# 시작
ollama serve

모델 다운로드 및 실행

# llama3.2 모델 다운로드
ollama pull llama3.2

# 로컬 API 시작
ollama run llama3.2

API 테스트

# curl로 테스트
curl http://localhost:11434/api/generate \
  -H "Content-Type: application/json" \
  -d '{
    "model": "llama3.2",
    "prompt": "Hello, how are you?",
    "stream": false
  }'

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

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

class TerminalAgent:
    def __init__(self, model="llama3.2"):
        self.model = model
        self.client = openai.OpenAI(
            base_url="http://localhost:11434/v1",
            api_key="ollama"
        )
        self.context = []

    def call_llm(self, prompt: str) -> str:
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7
        )
        return response.choices[0].message.content

    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 get_file_content(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 save_file(self, filepath: str, content: str) -> bool:
        try:
            with open(filepath, 'w') as f:
                f.write(content)
            return True
        except Exception as e:
            print(f"Error saving file: {str(e)}")
            return False

# 사용 예시
if __name__ == "__main__":
    agent = TerminalAgent()

    # 코드 생성 예시
    prompt = """
    Create a Python function that calculates the fibonacci sequence up to n terms.
    Return the list of numbers.
    """

    response = agent.call_llm(prompt)
    print("Generated code:")
    print(response)

4. tmux와의 통합

터미널 에이전트는 tmux와 함께 사용될 때 더욱 강력합니다:

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

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

# 세션에 파일 생성
tmux send-keys -t ai_agent "touch test.py" Enter

# 세션 정보 확인
tmux list-sessions

tmux 자동화 스크립트

#!/bin/bash
# setup_tmux.sh

SESSION_NAME="ai_dev"
tmux new-session -s $SESSION_NAME -d

# 메인 창 생성
tmux new-window -t $SESSION_NAME:1 -n "main"
tmux send-keys -t $SESSION_NAME:1 "cd ~/projects/myproject" Enter

# 터미널 창 생성
tmux new-window -t $SESSION_NAME:2 -n "terminal"
tmux send-keys -t $SESSION_NAME:2 "python aider_agent.py" Enter

# 에이전트 창 생성
tmux new-window -t $SESSION_NAME:3 -n "agent"
tmux send-keys -t $SESSION_NAME:3 "watch -n 1 'ls -la'" Enter

# 세션에 연결
tmux attach -t $SESSION_NAME

5. 커스텀 도구 개발

코드 검색 도구

# tools/code_search.py
import os
import re

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

    def search_in_files(self, pattern: str, file_extensions: list = None) -> List[Dict]:
        results = []
        if file_extensions is None:
            file_extensions = ['.py', '.js', '.ts', '.java', '.cpp']

        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.append({
                                    'file': filepath,
                                    'line': content[:match.start()].count('\n') + 1,
                                    'context': self.get_context(content, match.start())
                                })
                    except Exception as e:
                        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])

# 사용 예시
searcher = CodeSearchTool("/path/to/project")
results = searcher.search_in_files(r"def.*function_name")

Git 통합 도구

# tools/git_tool.py
import subprocess
import json

class GitTool:
    def __init__(self):
        pass

    def get_status(self) -> Dict:
        result = subprocess.run(['git', 'status', '--porcelain'], 
                               capture_output=True, text=True)
        return {"status": result.stdout.strip()}

    def get_diff(self) -> str:
        result = subprocess.run(['git', 'diff'], 
                               capture_output=True, text=True)
        return result.stdout

    def commit_changes(self, message: str) -> bool:
        try:
            subprocess.run(['git', 'add', '.'], check=True)
            subprocess.run(['git', 'commit', '-m', message], check=True)
            return True
        except subprocess.CalledProcessError:
            return False

    def get_branch_info(self) -> Dict:
        result = subprocess.run(['git', 'branch', '--show-current'], 
                               capture_output=True, text=True)
        return {"branch": result.stdout.strip()}

# 사용 예시
git_tool = GitTool()
status = git_tool.get_status()
print(json.dumps(status, indent=2))

6. 컨텍스트 윈도우 관리

대규모 코드베이스에서는 컨텍스트 윈도우 관리가 중요합니다:


python
# context_manager.py
import tiktoken
from typing import List, Dict

class ContextManager:
    def __init__(self, model_name: str = "gpt-4"):
        self.encoding = tiktoken.encoding_for_model(model_name)
        self.max_tokens = 8192  # 8K 토큰 제한

    def count_tokens(self, text: str) -> int:
        return len(self.encoding.encode(text))

    def truncate_context(self, context: List[Dict], max_tokens: int = None) -> List[Dict]:
        if max_tokens is None:
            max_tokens = self.max_tokens

        truncated = []
        current_tokens = 0

        # 최근 항목부터 처리
        for item in reversed(context):
            item_tokens = self.count_tokens(str(item))
            if current_tokens + item_tokens <= max_tokens:
                truncated.append(item)
                current_tokens += item_tokens
            else:
                break

        return list(reversed(truncated))

    def add_context

---

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

RAG 시스템 실전 구축 (v5)

matias yoon — Sun, 24 May 2026 11:28:13 +0000

RAG 시스템 실전 구축 (v5)

Practical Guide for ML Engineers & Backend Developers

Product: $5 RAG Pipeline Starter Kit with 500+ Code Examples

1. RAG Fundamentals: The Retrieval-Augmentation-Generation Loop

Retriever-Augmenter-Generator (RAG) systems work in three phases:

# Basic RAG workflow
def rag_pipeline(query, vector_db, llm):
    # 1. RETRIEVAL: Find relevant documents
    relevant_docs = vector_db.search(query, top_k=5)

    # 2. AUGMENTATION: Combine context with query
    augmented_prompt = format_prompt(query, relevant_docs)

    # 3. GENERATION: Generate response
    response = llm.generate(augmented_prompt)
    return response

Key components:

Retriever: Finds relevant documents (vector similarity search)
Augmenter: Formats context into prompt
Generator: LLM produces final answer

2. Chunking Strategies: Breaking Down Documents Efficiently

Semantic Chunking (Best for code understanding):

import tiktoken
from sentence_transformers import SentenceTransformer

def semantic_chunking(text, model, max_tokens=512):
    sentences = text.split('. ')
    chunks = []
    current_chunk = ""

    for sentence in sentences:
        # Estimate token count
        if len(tokenize(current_chunk + sentence)) > max_tokens:
            chunks.append(current_chunk.strip())
            current_chunk = sentence + ". "
        else:
            current_chunk += sentence + ". "

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

Recursive Chunking (For structured data):

def recursive_chunking(text, chunk_size=1000):
    """Split text recursively by delimiters"""
    delimiters = ['\n\n', '\n', '. ', ' ', '']
    chunks = [text]

    for delim in delimiters:
        new_chunks = []
        for chunk in chunks:
            if len(chunk) > chunk_size:
                new_chunks.extend(chunk.split(delim))
            else:
                new_chunks.append(chunk)
        chunks = new_chunks
        if all(len(chunk) <= chunk_size for chunk in chunks):
            break

    return [chunk for chunk in chunks if len(chunk) > 10]

3. Embedding Model Selection and Comparison

Top 5 Models for Code RAG:

# Model comparison benchmark
models = {
    "all-MiniLM-L6-v2": {
        "dimensions": 384,
        "size_mb": 100,
        "speed": "fast",
        "code_sensitivity": "high"
    },
    "Sentence-BERT": {
        "dimensions": 768,
        "size_mb": 400,
        "speed": "medium",
        "code_sensitivity": "very_high"
    },
    "CodeBERT": {
        "dimensions": 768,
        "size_mb": 1000,
        "speed": "medium",
        "code_sensitivity": "excellent"
    },
    "MiniLM": {
        "dimensions": 384,
        "size_mb": 50,
        "speed": "fast",
        "code_sensitivity": "medium"
    },
    "MPNet": {
        "dimensions": 768,
        "size_mb": 400,
        "speed": "slow",
        "code_sensitivity": "high"
    }
}

# Implementation example
class EmbeddingService:
    def __init__(self, model_name="all-MiniLM-L6-v2"):
        self.model = SentenceTransformer(model_name)

    def encode(self, texts):
        return self.model.encode(texts)

    def get_dimensions(self):
        return self.model.get_sentence_embedding_dimension()

# Benchmark function
def benchmark_embeddings():
    service = EmbeddingService()
    texts = ["Sample code snippet"] * 1000
    import time

    start = time.time()
    embeddings = service.encode(texts)
    end = time.time()

    print(f"Time for 1000 embeddings: {end-start:.2f}s")
    return embeddings

4. Vector Database Comparison

Database	Speed	Memory	Cost	Best For
Chroma	Fast	Low	Free	Development
Qdrant	Fast	Medium	Free	Production
pgvector	Medium	High	Free	PostgreSQL users
Milvus	Fast	High	Free	Large-scale

# Chroma implementation
import chromadb

class ChromaVectorStore:
    def __init__(self, collection_name="code_docs"):
        self.client = chromadb.Client()
        self.collection = self.client.get_or_create_collection(collection_name)

    def add_documents(self, documents, embeddings):
        self.collection.add(
            documents=documents,
            embeddings=embeddings,
            ids=[str(i) for i in range(len(documents))]
        )

    def search(self, query, top_k=5):
        result = self.collection.query(
            query_texts=[query],
            n_results=top_k
        )
        return result['documents'][0]

# Qdrant implementation
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition

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

    def search(self, query_vector, top_k=5):
        results = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_vector,
            limit=top_k
        )
        return [hit.payload['text'] for hit in results]

5. Full RAG Pipeline from Scratch

import os
import json
from pathlib import Path
from typing import List, Dict
from sentence_transformers import SentenceTransformer
import chromadb
from chromadb import Client

class CodeRAGPipeline:
    def __init__(self, embedding_model="all-MiniLM-L6-v2", db_path="./chroma_db"):
        self.embedding_model = SentenceTransformer(embedding_model)
        self.client = chromadb.PersistentClient(path=db_path)
        self.collection = self.client.get_or_create_collection("code_documents")
        self.chunk_size = 1000

    def preprocess_code_files(self, file_path: str) -> List[str]:
        """Extract code chunks from file"""
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()

        # Simple chunking by function/class boundaries
        lines = content.split('\n')
        chunks = []
        current_chunk = []

        for line in lines:
            if line.strip().startswith(('def ', 'class ')) and current_chunk:
                chunks.append('\n'.join(current_chunk))
                current_chunk = [line]
            else:
                current_chunk.append(line)

        if current_chunk:
            chunks.append('\n'.join(current_chunk))

        return chunks

    def embed_and_store(self, documents: List[str], file_path: str):
        """Generate embeddings and store in vector DB"""
        embeddings = self.embedding_model.encode(documents)

        self.collection.add(
            documents=documents,
            embeddings=embeddings.tolist(),
            ids=[f"{file_path}_{i}" for i in range(len(documents))]
        )

    def query(self, query: str, top_k: int = 5) -> List[Dict]:
        """Query the RAG system"""
        query_embedding = self.embedding_model.encode([query])

        results = self.collection.query(
            query_embeddings=query_embedding.tolist(),
            n_results=top_k
        )

        return [
            {
                'text': doc,
                'distance': dist
            }
            for doc, dist in zip(results['documents'][0], results['distances'][0])
        ]

    def generate_response(self, query: str, retrieved_docs: List[Dict]) -> str:
        """Generate final response using LLM"""
        context = "\n\n".join([doc['text'] for doc in retrieved_docs[:3]])
        prompt = f"""
        Context: {context}

        Question: {query}

        Answer:
        """
        # This would integrate with your LLM (e.g., llama.cpp)
        return f"Based on {len(retrieved_docs)} documents, the answer is..."

6. Advanced Techniques: Query Transformation & Hybrid Search

Query Transformation:


python
def transform_query(original_query: str) -> List[str]:
    """Generate multiple query variations"""
    transformations = [
        original_query,
        f"Explain {original_query}",
        f"How to implement {original_query}",
        f"Code example for {original_query}",
        f"Best practices for {original_query}"
    ]
    return transformations

class HybridSearchRAG:
    def __init__(self):
        self.vector_store = ChromaVectorStore()
        self.text_store = TextSearchEngine()

    def hybrid_search(self, query: str, top_k: int = 5):
        """Combine vector and keyword search"""
        # Vector search
        vector_results = self.vector_store.search(query, top_k)

        # Keyword search (simplified)
        keyword_results = self.text_store.search(query, top_k)

        # Combine results with weighted scoring

---

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

로컬 LLM 셋업 가이드 (v5)

matias yoon — Sun, 24 May 2026 11:13:19 +0000

Local LLM Setup Guide (v5)

Practical Installation & Optimization Guide for Developers

1. Overview & Prerequisites

Running LLMs locally requires minimal hardware but significant attention to memory management and system configuration.

Hardware Requirements:

GPU: NVIDIA RTX 30xx/40xx series recommended (8GB+ VRAM)
CPU: Intel i5-12600K or AMD Ryzen 7 5800X
RAM: Minimum 16GB, 32GB recommended
Storage: 500GB+ SSD for model storage

OS: Ubuntu 22.04 LTS or Debian 12

Prerequisites Installation:

sudo apt update
sudo apt install -y git cmake build-essential python3-pip
pip3 install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118

2. Framework Comparison

Framework	Pros	Cons	Use Case
llama.cpp	Native C++, no dependencies, minimal overhead	No GPU acceleration in CPU mode	Lightweight inference
Ollama	Simple CLI, automatic model downloads	Less control, dependency on Docker	Quick prototyping
vLLM	High throughput, optimized for serving	Complex setup, requires Python expertise	Production inference
LocalAI	API-first, supports multiple backends	Heavy, requires container setup	Enterprise deployments

Recommendation: Use llama.cpp for development, Ollama for quick testing.

3. Step-by-Step Installation (llama.cpp)

# Clone and build llama.cpp
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp
make clean
make

# Download a model (example: Mistral 7B)
wget https://huggingface.co/TheBloke/Mistral-7B-v0.1-GGUF/resolve/main/mistral-7b-v0.1.Q4_K_M.gguf -P models/

# Basic inference test
./main -m models/mistral-7b-v0.1.Q4_K_M.gguf -p "Hello world" --temp 0.2

4. Model Selection Guide

For Chat Applications:

Mistral 7B Q4_K_M (balanced quality vs size)
Llama 2 7B Q4_K_M (best commercial support)

For Code Generation:

CodeLlama 7B Q4_K_M (best for programming tasks)
Phi-2 Q4_K_M (smaller, fast)

For Research/Development:

Llama 3 8B Q4_K_M (latest architecture)
Mixtral 8x7B Q4_K_M (sparse mixture of experts)

Example usage:

# Chat model
./main -m models/mistral-7b-v0.1.Q4_K_M.gguf \
  -p "You are a helpful AI assistant. User: What is 2+2? Assistant:" \
  --temp 0.1 --repeat_penalty 1.1

# Code model
./main -m models/codellama-7b.Q4_K_M.gguf \
  -p "def fibonacci(n):" \
  --temp 0.0 --repeat_penalty 1.0

5. Quantization Types Explained

Quantization	Size	Quality	Use Case
Q4_K_M	4.2GB	High	General use
Q5_K_M	5.1GB	Very High	Research
Q6_K	6.4GB	Excellent	Premium quality
Q8_0	8.2GB	Full	Maximum quality
F16	16GB	Full	Development/testing

Benchmark comparison (1000 tokens, RTX 4090):

Q4_K_M: 150 tokens/sec
Q5_K_M: 130 tokens/sec
Q8_0: 100 tokens/sec

Quantization command:

# Convert GGUF model with specific quantization
python3 convert-hf-to-gguf.py models/Mistral-7B-v0.1/ --outtype q4_k_m --outfile mistral-7b-q4k.gguf

6. API Setup and Integration

Simple HTTP API with Python:

# server.py
from flask import Flask, request, jsonify
import subprocess
import json

app = Flask(__name__)

@app.route('/generate', methods=['POST'])
def generate():
    data = request.json
    prompt = data['prompt']
    model_path = data.get('model', 'models/mistral-7b-v0.1.Q4_K_M.gguf')

    result = subprocess.run([
        './main', '-m', model_path, '-p', prompt, 
        '--temp', '0.2', '--repeat_penalty', '1.1'
    ], capture_output=True, text=True)

    return jsonify({'response': result.stdout.strip()})

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

Integration with existing tools:

# Test API
curl -X POST http://localhost:8000/generate \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Explain quantum computing in simple terms:"}'

7. Systemd Service for 24/7 Operation

Create /etc/systemd/system/local-llm.service:

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

[Service]
Type=simple
User=developer
WorkingDirectory=/home/developer/llama.cpp
ExecStart=/home/developer/llama.cpp/main -m /home/developer/models/mistral-7b-v0.1.Q4_K_M.gguf -p "System ready" --port 8080 --host 0.0.0.0
Restart=always
RestartSec=10
Environment=LD_LIBRARY_PATH=/usr/local/cuda/lib64

[Install]
WantedBy=multi-user.target

Enable and start:

sudo systemctl daemon-reload
sudo systemctl enable local-llm
sudo systemctl start local-llm
sudo systemctl status local-llm

8. Monitoring and Performance Tuning

Memory monitoring script:

#!/bin/bash
# monitor.sh
while true; do
    echo "Timestamp: $(date)"
    echo "GPU Memory:"
    nvidia-smi --query-gpu=memory.used,memory.total --format=csv,noheader,nounits
    echo "System Memory:"
    free -h
    echo "---"
    sleep 30
done

Performance optimization flags:

# For high throughput (1000+ tokens/sec)
./main -m model.gguf \
  --threads 16 \
  --ctx-size 4096 \
  --batch-size 512 \
  --temp 0.0 \
  --repeat_penalty 1.0 \
  --n-predict 1000

# For low latency (sub-100ms response)
./main -m model.gguf \
  --threads 8 \
  --ctx-size 1024 \
  --batch-size 64 \
  --temp 0.2 \
  --repeat_penalty 1.1

9. Real Command Examples

Complete workflow example:

# 1. Setup directory
mkdir -p ~/llm-dev/models
cd ~/llm-dev

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

# 3. Download model
wget https://huggingface.co/TheBloke/Mistral-7B-v0.1-GGUF/resolve/main/mistral-7b-v0.1.Q4_K_M.gguf -P models/

# 4. Test inference
./main -m models/mistral-7b-v0.1.Q4_K_M.gguf \
  -p "Write a bash script that checks disk space:" \
  --temp 0.1 --repeat_penalty 1.0

# 5. Monitor performance
watch -n 1 nvidia-smi

Benchmark script:

#!/bin/bash
# benchmark.sh
MODEL_PATH="models/mistral-7b-v0.1.Q4_K_M.gguf"
PROMPT="The quick brown fox jumps over the lazy dog. This is a test prompt for benchmarking."

echo "Starting benchmark..."
time ./main -m $MODEL_PATH -p "$PROMPT" --temp 0.0 --repeat_penalty 1.0 --n-predict 100
echo "Benchmark complete"

Quick deployment command:


bash
# One-liner setup and test
git clone https://github.com/ggerganov/llama.cpp && cd llama.cpp && make && \
wget https://huggingface.co/TheBloke/Mistral-7B-v0.1-GGUF/

---

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

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

matias yoon — Sun, 24 May 2026 10:58:25 +0000

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

개요: LangGraph 아키텍처 기초

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

노드 (Nodes): 각 작업 단계를 나타냄
엣지 (Edges): 노드 간의 전이 조건을 정의
상태 (State): 모든 노드가 공유하는 데이터 구조
체크포인트 (Checkpointing): 상태 저장 및 복구 기능

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

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

템플릿 1: 단순 RAG 에이전트

이 템플릿은 문서 검색 → 생성 → 검증의 단순한 RAG 워크플로우를 구현합니다. 특히 문서 검색과 생성을 분리하여 효율성을 극대화합니다.

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

class RAGAgentState(TypedDict):
    query: str
    retrieved_docs: list
    generated_response: str
    validation: bool

# 노드 정의
def retrieve_node(state):
    # 문서 검색
    vectorstore = FAISS.load_local("vectorstore", OpenAIEmbeddings())
    retriever = vectorstore.as_retriever()
    docs = retriever.invoke(state["query"])
    return {"retrieved_docs": docs}

def generate_node(state):
    # 문서 기반 생성
    template = PromptTemplate.from_template(
        "다음 문서를 기반으로 질문에 답변하세요:\n\n{context}\n\n질문: {question}"
    )
    llm = ChatOpenAI(model="gpt-4")
    chain = template | llm | StrOutputParser()

    context = "\n\n".join([doc.page_content for doc in state["retrieved_docs"]])
    response = chain.invoke({"context": context, "question": state["query"]})

    return {"generated_response": response}

def validate_node(state):
    # 응답 검증 (단순 예시)
    # 실제 구현에서는 더 복잡한 검증 로직이 필요
    return {"validation": len(state["generated_response"]) > 10}

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

    workflow.add_node("retrieve", retrieve_node)
    workflow.add_node("generate", generate_node)
    workflow.add_node("validate", validate_node)

    workflow.set_entry_point("retrieve")

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

    # 검증 실패 시 재시도
    workflow.add_conditional_edges(
        "validate",
        lambda x: "retry" if not x["validation"] else "end",
        {"retry": "retrieve", "end": END}
    )

    return workflow.compile()

템플릿 2: 멀티-도구 에이전트

이 템플릿은 계획 → 실행 → 관찰 → 결정의 반복적인 워크플로우를 구현하여 복잡한 작업을 수행합니다. 다양한 도구를 사용할 수 있도록 설계되었습니다.

from langchain.tools import Tool
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

class ToolAgentState(TypedDict):
    tasks: list[str]
    current_task: str
    execution_result: str
    decision: str

# 도구 정의
@tool
def web_search(query: str):
    """웹 검색 도구"""
    # 실제 검색 로직 구현
    return f"검색 결과: {query}"

@tool
def file_operations(operation: str, path: str):
    """파일 조작 도구"""
    return f"파일 {operation} 완료: {path}"

# 노드 정의
def plan_node(state):
    # 작업 계획
    tasks = ["웹 검색", "데이터 수집", "분석"]
    return {"tasks": tasks}

def execute_node(state):
    # 현재 작업 실행
    current_task = state["tasks"][0] if state["tasks"] else ""

    if "웹 검색" in current_task:
        result = web_search.run(current_task)
    elif "파일" in current_task:
        result = file_operations.run(current_task)
    else:
        result = f"작업 실행: {current_task}"

    return {"execution_result": result, "current_task": current_task}

def observe_node(state):
    # 실행 결과 관찰
    # 결과 분석 및 다음 단계 결정
    return {"decision": "continue"}

def decide_node(state):
    # 결정: 다음 작업 선택
    if state["tasks"]:
        next_task = state["tasks"][1:]  # 다음 작업
        return {"tasks": next_task}
    return {"tasks": []}

# 워크플로우 정의
def create_tool_workflow():
    workflow = StateGraph(ToolAgentState)

    workflow.add_node("plan", plan_node)
    workflow.add_node("execute", execute_node)
    workflow.add_node("observe", observe_node)
    workflow.add_node("decide", decide_node)

    workflow.set_entry_point("plan")

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

    workflow.add_conditional_edges(
        "decide",
        lambda x: "end" if not x["tasks"] else "execute",
        {"end": END, "execute": "execute"}
    )

    return workflow.compile()

템플릿 3: 인간-중개 워크플로우

이 템플릿은 인간의 개입이 필요한 상황에서 사용되며, 실행 일시정지 → 검토 → 계속의 흐름을 제공합니다.

from langchain_core.messages import HumanMessage
import time

class HumanInLoopState(TypedDict):
    messages: list[BaseMessage]
    pending_review: str
    human_feedback: str

def pause_node(state):
    # 실행 일시정지 및 인간 검토 요청
    return {"pending_review": "필요한 검토가 있습니다"}

def review_node(state):
    # 인간 검토 노드 (예시: 가상 검토)
    # 실제 구현에서는 웹훅 또는 메시지 큐를 통해 검토 요청
    print("사용자 검토 요청:", state["pending_review"])
    time.sleep(1)  # 시뮬레이션
    return {"human_feedback": "검토 완료"}

def continue_node(state):
    # 검토 완료 후 계속 진행
    return {"pending_review": None}

def create_human_loop_workflow():
    workflow = StateGraph(HumanInLoopState)

    workflow.add_node("pause", pause_node)
    workflow.add_node("review", review_node)
    workflow.add_node("continue", continue_node)

    workflow.set_entry_point("pause")

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

    return workflow.compile()

템플릿 5: 병렬 실행 에이전트

이 템플릿은 하나의 입력을 여러 작업으로 분할 → 병렬 처리 → 결과 집계의 패턴을 제공합니다.


python
from concurrent.futures import ThreadPoolExecutor
import asyncio

class ParallelAgentState(TypedDict):
    inputs: list[str]
    parallel_results: list[str]
    aggregated_result: str

def fan_out_node(state):
    # 입력 분할
    return {"parallel_results": []}

def process_node(state):
    # 각각의 입력 처리 (병렬 실행)
    def process_single(input_item):
        # 실제 처리 로직
        return f"처리 결과: {input_item}"

    inputs = state["inputs"]
    # 병렬 처리 (실제 구현에서는 ThreadPoolExecutor 사용)
    with ThreadPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(process_single, inputs))

    return {"parallel_results": results}

def aggregate_node(state):
    # 결과 집계
    aggregated = ", ".join(state["parallel_results"])
    return {"aggregated_result": aggregated}

def create_parallel_workflow():
    workflow = StateGraph(ParallelAgentState)

    workflow.add_node("fan_out", fan_out_node)
    workflow.add_node("process", process_node)
    workflow.add_node("aggregate", aggregate_node)

    workflow.set_entry_point("fan_out")

    workflow.add_edge("fan_out", "process")
    workflow.add_edge("process", "aggregate")
    workflow.add_edge("aggregate", END)


---

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

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

matias yoon — Sun, 24 May 2026 10:43:32 +0000

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

개발자를 위한 경량 로컬 AI 코딩 어시스턴트 구축 가이드

1. CLI AI 에이전트 생태계 개요

터미널 기반 AI 에이전트는 개발자들이 코드를 작성하고 디버깅할 때 실시간으로 도움을 받을 수 있도록 해주는 도구입니다. 현재 주류로는 다음과 같은 솔루션들이 있습니다:

Aider

# 설치
pip install aider

# 사용 예시
aider --model gpt-4 --yes

Continue.dev

# VSCode 확장 설치
# 설치 후 터미널에서 실행
continue

OpenCode

# GitHub에서 클론
git clone https://github.com/opencode/opencode.git
cd opencode
npm install

커스텀 스크립트

가장 유연한 접근 방식으로, 직접 만들 수 있는 도구는 다음처럼 구성할 수 있습니다:

# aider-like 기능을 가진 간단한 에이전트
import subprocess
import json

class TerminalAgent:
    def __init__(self):
        self.model = "llama3:8b"

    def query(self, prompt):
        # 로컬 LLM 호출
        result = subprocess.run([
            "ollama", "run", self.model, prompt
        ], capture_output=True, text=True)
        return result.stdout

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

로컬 모델을 터미널에서 사용하기 위해 Ollama를 사용해 API 엔드포인트를 설정합니다:

# Ollama 설치
curl -fsSL https://ollama.com/install.sh | sh

# 모델 다운로드
ollama pull llama3:8b

# API 엔드포인트 시작 (기본 포트 11434)
ollama serve &

API 엔드포인트를 통해 다음과 같은 방식으로 호출할 수 있습니다:

import requests
import json

def call_local_llm(prompt):
    response = requests.post(
        'http://localhost:11434/api/generate',
        json={
            'model': 'llama3:8b',
            'prompt': prompt,
            'stream': False
        }
    )
    return response.json()['response']

# 사용 예시
result = call_local_llm("Python에서 파일을 읽는 코드를 작성해주세요")
print(result)

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

다음은 간단한 Python 기반 CLI 에이전트입니다. 이 에이전트는 코드를 생성하고 실행하며, 함수 호출 기능을 포함합니다:

#!/usr/bin/env python3
import subprocess
import sys
import requests
import json
import os

class LocalAIAgent:
    def __init__(self, model="llama3:8b"):
        self.model = model
        self.base_url = "http://localhost:11434"

    def execute_command(self, command):
        """터미널 명령 실행"""
        try:
            result = subprocess.run(
                command, 
                shell=True, 
                capture_output=True, 
                text=True,
                timeout=30
            )
            return result.stdout, result.stderr
        except subprocess.TimeoutExpired:
            return "", "Command timed out"

    def generate_code(self, instruction, context=""):
        """LLM을 통한 코드 생성"""
        prompt = f"""
        You are a senior Python developer.
        Context: {context}
        Instruction: {instruction}
        Provide only the code, no explanations.
        """

        response = requests.post(
            f'{self.base_url}/api/generate',
            json={
                'model': self.model,
                'prompt': prompt,
                'stream': False
            }
        )

        if response.status_code == 200:
            return response.json()['response']
        return ""

    def save_and_run(self, code, filename):
        """코드 저장 및 실행"""
        with open(filename, 'w') as f:
            f.write(code)

        # 실행
        stdout, stderr = self.execute_command(f'python {filename}')
        return stdout, stderr

# 사용 예시
if __name__ == "__main__":
    agent = LocalAIAgent()

    # 코드 생성
    code = agent.generate_code(
        "Create a Python function that calculates Fibonacci numbers with memoization"
    )

    print("Generated code:")
    print(code)

    # 저장 및 실행
    stdout, stderr = agent.save_and_run(code, "fibonacci.py")
    print("Output:", stdout)

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

tmux를 활용하여 에이전트를 터미널 세션에 통합할 수 있습니다:

# tmux 설치
sudo apt install tmux  # Ubuntu/Debian
brew install tmux      # macOS

# 새로운 세션 생성
tmux new-session -s ai-agent -d

# 세션에 명령 실행
tmux send-keys -t ai-agent "python3 agent.py" Enter

# 세션 접속
tmux attach -t ai-agent

다음은 tmux 통합을 위한 Python 스크립트입니다:

import subprocess
import os

class TmuxIntegrator:
    def __init__(self, session_name="ai-agent"):
        self.session_name = session_name

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

    def run_command_in_session(self, command):
        """세션 내에서 명령 실행"""
        subprocess.run([
            "tmux", "send-keys", "-t", self.session_name, command, "Enter"
        ])

    def attach_session(self):
        """세션 접속"""
        subprocess.run(["tmux", "attach", "-t", self.session_name])

# 사용 예시
integrator = TmuxIntegrator()
integrator.create_session()
integrator.run_command_in_session("python3 agent.py")

5. 커스텀 도구 개발

다음은 코드 검색, Git 통합, 파일 조작을 위한 커스텀 도구들입니다:

코드 검색 도구

import subprocess
import os

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

    def search_in_files(self, pattern, file_extensions=None):
        """파일 내에서 패턴 검색"""
        if file_extensions:
            extensions = " ".join([f"-name '*.{ext}'" for ext in file_extensions])
            cmd = f"find {self.project_root} -type f {extensions} -exec grep -l '{pattern}' {{}} \\;"
        else:
            cmd = f"find {self.project_root} -type f -exec grep -l '{pattern}' {{}} \\;"

        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
        return result.stdout.strip().split('\n') if result.stdout else []

# 사용 예시
searcher = CodeSearchTool()
results = searcher.search_in_files("def main", ["py", "js"])
print("Found files:", results)

Git 통합 도구

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

    def get_git_status(self):
        """Git 상태 확인"""
        result = subprocess.run(
            ["git", "status", "--porcelain"],
            capture_output=True, text=True
        )
        return result.stdout.strip().split('\n') if result.stdout else []

    def create_branch(self, branch_name):
        """브랜치 생성"""
        subprocess.run(["git", "checkout", "-b", branch_name])

    def commit_changes(self, message):
        """변경사항 커밋"""
        subprocess.run(["git", "add", "."])
        subprocess.run(["git", "commit", "-m", message])

# 사용 예시
git_tool = GitTool()
status = git_tool.get_git_status()
print("Git status:", status)

파일 조작 도구


python
import os
import shutil
from pathlib import Path

class FileTool:
    def create_directory(self, path):
        """디렉토리 생성"""
        os.makedirs(path, exist_ok=True)

    def copy_file(self, src, dst):
        """파일 복사"""
        shutil.copy2(src, dst)

    def delete_file(self, path):
        """파일 삭제"""
        if os.path.exists(path):
            os.remove(path)

    def read_file(self, path):
        """파일 읽기"""
        with open(path, 'r') as f:
            return f.read()

    def write_file(self, path, content):
        """파일 쓰기"""
        with open(path, 'w') as f:
            f.write(content)

# 사용 예시
file_tool = FileTool()
file_tool.write_file("test.txt", "Hello, World!")
content = file_tool.read_file("test.txt")
print("

---

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

RAG 시스템 실전 구축 (v4)

matias yoon — Sun, 24 May 2026 10:28:37 +0000

RAG 시스템 실전 구축 (v4)

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

Retrieval-Augmented Generation (RAG)은 대규모 언어 모델(LLM)의 지식 범위를 확장하기 위한 아키텍처입니다. RAG는 세 가지 핵심 구성 요소로 구성됩니다:

검색 (Retrieval): 사용자의 질문과 관련된 문서 또는 텍스트 조각을 찾습니다
보완 (Augmentation): 검색된 정보를 질문과 함께 LLM 입력으로 제공합니다
생성 (Generation): LLM이 검색된 컨텍스트와 질문을 기반으로 답변을 생성합니다

# 간단한 RAG 루프 구현
class SimpleRAG:
    def __init__(self, embedding_model, vector_db):
        self.embedding_model = embedding_model
        self.vector_db = vector_db

    def retrieve(self, query, k=5):
        query_embedding = self.embedding_model.encode(query)
        return self.vector_db.search(query_embedding, k=k)

    def generate(self, query, retrieved_docs):
        context = "\n".join(retrieved_docs)
        prompt = f"질문: {query}\n컨텍스트: {context}\n답변:"
        return self.llm.generate(prompt)

    def run(self, query):
        docs = self.retrieve(query)
        return self.generate(query, docs)

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

청킹은 대용량 텍스트를 모델이 처리할 수 있는 조각으로 나누는 과정입니다:

의미론적 청킹 (Semantic Chunking)

문서의 의미적 단위를 기준으로 청킹합니다:

from sentence_transformers import SentenceTransformer
import numpy as np
from sklearn.cluster import KMeans

def semantic_chunking(text, model, threshold=0.75):
    sentences = text.split('. ')
    embeddings = model.encode(sentences)

    # 유사도 기반 클러스터링
    similarity_matrix = np.dot(embeddings, embeddings.T)
    chunks = []
    current_chunk = []

    for i, sentence in enumerate(sentences):
        if not current_chunk:
            current_chunk.append(sentence)
        else:
            # 가장 최근 문장과의 유사도 계산
            recent_embedding = embeddings[i-1]
            current_embedding = embeddings[i]
            similarity = np.dot(recent_embedding, current_embedding) / (
                np.linalg.norm(recent_embedding) * np.linalg.norm(current_embedding)
            )

            if similarity < threshold:
                chunks.append(' '.join(current_chunk))
                current_chunk = [sentence]
            else:
                current_chunk.append(sentence)

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

    return chunks

재귀적 청킹 (Recursive Chunking)

계층적으로 청킹을 수행합니다:

def recursive_chunking(text, max_chunk_size=512):
    if len(text) <= max_chunk_size:
        return [text]

    # 문장 단위로 분할
    sentences = text.split('. ')
    chunks = []
    current_chunk = []
    current_length = 0

    for sentence in sentences:
        sentence_length = len(sentence) + 1  # +1 for space
        if current_length + sentence_length <= max_chunk_size:
            current_chunk.append(sentence)
            current_length += sentence_length
        else:
            if current_chunk:
                chunks.append('. '.join(current_chunk))
            current_chunk = [sentence]
            current_length = sentence_length

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

    return chunks

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

다양한 임베딩 모델을 비교하여 최적의 선택을 도와줍니다:

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

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')
        }

    def benchmark_models(self, test_sentences):
        results = {}
        for name, model in self.models.items():
            embeddings = model.encode(test_sentences)
            avg_time = self._time_encoding(model, test_sentences)
            results[name] = {
                'avg_time': avg_time,
                'embedding_size': embeddings.shape[1],
                'memory_usage': embeddings.nbytes
            }
        return results

    def _time_encoding(self, model, sentences):
        import time
        start = time.time()
        model.encode(sentences)
        return time.time() - start

# 사용 예시
benchmark = EmbeddingBenchmark()
test_data = ["Hello world", "Machine learning is powerful", "Natural language processing"]
results = benchmark.benchmark_models(test_data)
print(results)

4. 벡터 데이터베이스 비교: Chroma vs Qdrant vs pgvector vs Milvus

각 벡터 데이터베이스의 장단점을 비교합니다:

# Chroma (간단한 로컬 벡터 DB)
import chromadb
from chromadb import Client

class ChromaDB:
    def __init__(self, path="chroma_db"):
        self.client = Client(path)
        self.collection = self.client.get_or_create_collection("documents")

    def add_documents(self, documents, embeddings):
        self.collection.add(
            documents=documents,
            embeddings=embeddings,
            ids=[str(i) for i in range(len(documents))]
        )

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

# Qdrant (고성능 분산 벡터 DB)
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue

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

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

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

    def search(self, query_embedding, top_k=5):
        results = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_embedding,
            limit=top_k
        )
        return [hit.payload['text'] for hit in results]

# pgvector (PostgreSQL 확장)
import psycopg2
import numpy as np

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

    def _setup_table(self):
        with self.conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS documents (
                    id SERIAL PRIMARY KEY,
                    text TEXT,
                    embedding VECTOR(384)
                )
            """)
            cur.execute("CREATE INDEX IF NOT EXISTS idx_embedding ON documents USING ivfflat (embedding vector_cosine_ops)")
        self.conn.commit()

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

    def search(self, query_embedding, top_k=5):
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT text FROM documents 
                ORDER BY embedding <-> %s 
                LIMIT %s
            """, (query_embedding.tolist(), top_k))
            return [row[0] for row in cur.fetchall()]

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

다음은 완전한 RAG 파이프라인 구현입니다:


python
import os
from sentence_transformers import SentenceTransformer
from chromadb import Client
import numpy as np

class FullRAGPipeline:
    def __init__(self, embedding_model_name="all-MiniLM-L6-v2"):
        self.embedding_model = SentenceTransformer(embedding_model_name)
        self.chroma_client = Client()
        self.collection = self.chroma_client.get_or_create_collection("documents")
        self.chunk_size = 512
        self.overlap = 50

    def chunk_document(self, text):
        """문서 청킹"""
        # 간단한 재귀적 청킹
        sentences = text.split('. ')
        chunks = []
        current_chunk = []
        current_length = 0



---

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

로컬 LLM 셋업 가이드 (v4)

matias yoon — Sun, 24 May 2026 10:13:43 +0000

로컬 LLM 셋업 가이드 (v4)

1. 개요 및 사전 준비

로컬 LLM은 인터넷 연결 없이도 AI 추론을 할 수 있는 방식입니다. 이 가이드는 Linux 기반 시스템에서 실행 가능한 최적의 로컬 LLM 설정 방법을 안내합니다.

사전 요구사항:

OS: Ubuntu 20.04 이상 (권장: Ubuntu 22.04)
CPU: 최소 8코어 이상 (16코어 이상 권장)
RAM: 최소 16GB (32GB 이상 권장)
GPU: NVIDIA RTX 30xx/40xx 시리즈 이상 (메모리 8GB 이상)
저장공간: 최소 50GB (모델별로 다름)

GPU 확인 명령:

nvidia-smi
lspci | grep -i nvidia

2. 프레임워크 비교

프레임워크	장점	단점	사용 사례
llama.cpp	가볍고 빠르며, 직접 컴파일 가능	GUI 없음, 모델 변환 필요	서버/스마트폰 앱용
Ollama	쉬운 설치 및 관리	GPU 가속 제한	개발 테스트
vLLM	높은 추론 속도	복잡한 설치	고성능 서버
LocalAI	REST API 제공, 다양한 모델 지원	리소스 소모 많음	API 기반 앱

권장: llama.cpp + llamafile (가장 간단하고 효율적)

3. 설치 단계

3.1 기본 패키지 설치:

sudo apt update
sudo apt install -y build-essential git cmake python3 python3-pip

3.2 llama.cpp 설치:

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

3.3 모델 다운로드:

mkdir -p models
cd models
wget https://huggingface.co/TheBloke/Llama-3.2-3B-Instruct-GGUF/resolve/main/llama-3.2-3b-instruct.Q4_K_M.gguf

3.4 실행:

cd ../
./llama.cpp/main -m models/llama-3.2-3b-instruct.Q4_K_M.gguf -p "Qwen: Hello World" -n 100

4. 모델 선택 가이드

4.1 일반 용도:

Llama-3.2-3B-Instruct: 빠르고 효율적, 문서 분석용
Phi-4: 가볍고 빠른 추론

4.2 복잡한 문서 처리:

LLaVA-1.5-7B: 이미지/텍스트 혼합 처리 (시각 모델)
Qwen-VL: 중국어 기반, 문서 이해력 뛰어남

4.3 고성능:

Mixtral-8x22B: 대용량 모델, 추론 속도 저하 시 Q4_K_M로 변환

5. 양자화 타입 설명

타입	메모리 사용량	정확도	사용 사례
Q4_K_M	~5GB	높음	일반 추론
Q5_K_M	~6GB	매우 높음	정밀 문서 분석
Q6_K	~7GB	고정	높은 정확도 필요
Q8_0	~10GB	최고	테스트 및 개발

# 양자화 예시 (llama.cpp 내장)
./llama.cpp/convert-hf-to-gguf.py --model-path TheBloke/Llama-3.2-3B-Instruct-GGUF --output-path llama-3.2-3b-instruct.q4_k_m.gguf

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

6.1 FastAPI 서버 설정:

pip install fastapi uvicorn

6.2 API 서버 코드 (`server.py`):

from fastapi import FastAPI
import subprocess
import json

app = FastAPI()

@app.post("/chat")
async def chat(prompt: str):
    result = subprocess.run([
        "./llama.cpp/main",
        "-m", "models/llama-3.2-3b-instruct.Q4_K_M.gguf",
        "--prompt", prompt,
        "-n", "100",
        "--no-display-prompt"
    ], capture_output=True, text=True)

    return {"response": result.stdout}

6.3 실행 명령:

uvicorn server:app --reload --host 0.0.0.0 --port 8000

6.4 기존 앱 통합 예시:

curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Explain quantum computing in simple terms"}'

7. 24/7 운영을 위한 Systemd 서비스

7.1 서비스 파일 생성:

sudo nano /etc/systemd/system/llm-server.service

7.2 서비스 설정:

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

[Service]
Type=simple
User=your_user
WorkingDirectory=/home/your_user/llama.cpp
ExecStart=/usr/bin/python3 /home/your_user/llama.cpp/server.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

7.3 서비스 시작:

sudo systemctl daemon-reload
sudo systemctl enable llm-server
sudo systemctl start llm-server
sudo systemctl status llm-server

8. 모니터링 및 성능 튜닝

8.1 성능 모니터링:

# GPU 메모리 사용량
watch -n 1 nvidia-smi

# CPU/메모리 사용량
htop

8.2 추론 성능 테스트:

# 테스트 명령
./llama.cpp/main -m models/llama-3.2-3b-instruct.Q4_K_M.gguf -p "Test prompt" -n 100 --timing

8.3 자동 확장 설정 (예시):

# CPU 사용량 초과 시 알림
sudo apt install sysstat

9. 실전 예제 명령들

9.1 문서 분석 예제:

# 이미지 포함 문서 분석 (LLaVA 모델)
./llama.cpp/main -m models/llava-1.5-7b.Q4_K_M.gguf -p "Describe the image" -i image.jpg

9.2 대화형 추론:

# 채팅 모드
./llama.cpp/main -m models/llama-3.2-3b-instruct.Q4_K_M.gguf -c 2048 --temp 0.7 --repeat_penalty 1.1

9.3 성능 벤치마크:

# 추론 시간 측정
time ./llama.cpp/main -m models/llama-3.2-3b-instruct.Q4_K_M.gguf -p "Benchmark test" -n 100

10. 주의사항 및 문제 해결

10.1 메모리 부족 해결:

# 스왑 공간 생성
sudo fallocate -l 8G /swapfile
sudo chmod 600 /swapfile
sudo mkswap /swapfile
sudo swapon /swapfile

10.2 GPU 드라이버 업데이트:

sudo apt install nvidia-driver-535
sudo reboot

이 가이드를 따르면 기본적으로 로컬 LLM을 설치하고 실행할 수 있습니다. 다양한 모델과 설정을 통해 실제 엔지니어링 환경에서 사용할 수 있는 로컬 AI 서비스를 구축할 수 있습니다.

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

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

matias yoon — Sun, 24 May 2026 09:58:48 +0000

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

소개

LangGraph는 LangChain과 함께 사용하는 AI 에이전트를 구축하기 위한 강력한 프레임워크입니다. 이 가이드에서는 실제 개발자가 쉽게 적용할 수 있는 4개의 핵심 워크플로우 템플릿을 제공합니다. 각 템플릿은 특정 문제 해결을 위한 구조화된 접근 방식을 제공하며, 로컬 LLM 실행, GPU 리소스 최적화, 멀티모델 통합 등의 실질적인 문제를 해결합니다.

1. LangGraph 아키텍처 개요

LangGraph는 다음과 같은 핵심 구성 요소로 구성됩니다:

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

# 상태 정의
class GraphState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    # 추가 상태 필드
    user_input: str
    context: str

# 노드 정의
def retrieve_node(state):
    # 검색 로직
    return {"messages": [message]}

def generate_node(state):
    # 생성 로직
    return {"messages": [message]}

# 그래프 생성
workflow = StateGraph(GraphState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)
workflow.add_edge("retrieve", "generate")
workflow.set_entry_point("retrieve")
workflow.set_finish_point("generate")

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

이 템플릿은 문서 검색과 생성, 검증을 포함한 완전한 RAG 파이프라인입니다.

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_core.runnables import RunnableLambda
import operator

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

# 검색 노드
def retrieve_document(state):
    embeddings = OpenAIEmbeddings()
    vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)
    docs = vectorstore.similarity_search(state["question"], k=3)
    context = "\n".join([doc.page_content for doc in docs])
    return {"context": context}

# 생성 노드
def generate_answer(state):
    prompt = PromptTemplate.from_template(
        "Context: {context}\n\nQuestion: {question}\n\nAnswer:"
    )
    llm = ChatOpenAI(model="gpt-4", temperature=0.1)
    chain = prompt | llm | StrOutputParser()
    answer = chain.invoke({"context": state["context"], "question": state["question"]})
    return {"answer": answer}

# 검증 노드
def validate_answer(state):
    prompt = PromptTemplate.from_template(
        "Validate if the answer is correct for the question.\n\nQuestion: {question}\n\nAnswer: {answer}\n\nAnswer (True/False):"
    )
    llm = ChatOpenAI(model="gpt-4", temperature=0.1)
    chain = prompt | llm | StrOutputParser()
    validation = chain.invoke({"question": state["question"], "answer": state["answer"]})
    return {"validation": validation.lower() == "true"}

# 전체 워크플로우
def create_rag_workflow():
    workflow = StateGraph(RagState)

    workflow.add_node("retrieve", retrieve_document)
    workflow.add_node("generate", generate_answer)
    workflow.add_node("validate", validate_answer)

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

    workflow.set_entry_point("retrieve")
    workflow.set_finish_point("validate")

    return workflow.compile()

# 사용 예시
rag_workflow = create_rag_workflow()
result = rag_workflow.invoke({"question": "Python의 가상환경 생성 방법은?"})

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

이 템플릿은 복잡한 작업을 계획하고 실행하는 멀티-도구 에이전트를 구현합니다.

from typing import List, Dict, Any
from langchain.tools import tool
from langchain_core.messages import ToolMessage

class MultiToolState(TypedDict):
    tasks: List[str]
    current_task: str
    execution_result: str
    feedback: str

# 도구 정의
@tool
def search_web(query: str) -> str:
    """웹에서 정보 검색"""
    # 실제 구현은 web search API 호출
    return f"Search results for '{query}'"

@tool
def calculate_math(expression: str) -> str:
    """수학 계산"""
    # 실제 구현은 계산기 로직
    return f"Result of {expression}"

@tool
def get_weather(city: str) -> str:
    """날씨 정보 가져오기"""
    # 실제 구현은 날씨 API 호출
    return f"Weather in {city}: Sunny, 25°C"

# 계획 노드
def plan_task(state):
    # 현재 작업 계획 생성
    tasks = [
        "Research technical specifications",
        "Perform mathematical calculations",
        "Check current weather conditions"
    ]
    return {"tasks": tasks, "current_task": tasks[0]}

# 실행 노드
def execute_task(state):
    # 현재 작업 실행
    tool_map = {
        "search": search_web,
        "calculate": calculate_math,
        "weather": get_weather
    }

    # 실제 실행 로직 (간소화됨)
    task = state["current_task"]
    result = f"Executed task: {task}"
    return {"execution_result": result}

# 관찰 노드
def observe_result(state):
    # 실행 결과 분석
    return {"feedback": f"Task '{state['current_task']}' completed successfully"}

# 결정 노드
def decide_next(state):
    # 다음 작업 결정
    current_index = state["tasks"].index(state["current_task"])
    if current_index < len(state["tasks"]) - 1:
        return {"current_task": state["tasks"][current_index + 1]}
    return {"current_task": None}

# 전체 워크플로우
def create_multitool_workflow():
    workflow = StateGraph(MultiToolState)

    workflow.add_node("plan", plan_task)
    workflow.add_node("execute", execute_task)
    workflow.add_node("observe", observe_result)
    workflow.add_node("decide", decide_next)

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

    workflow.set_entry_point("plan")
    workflow.set_finish_point("decide")

    return workflow.compile()

4. 템플릿 3: 인간-함께-작업 워크플로우 (일시정지 → 검토 → 진행)

이 템플릿은 인간 검토가 필요한 작업 흐름을 구현합니다.

from typing import Optional
import asyncio

class HumanInLoopState(TypedDict):
    task_data: str
    human_review: Optional[str]
    approval: Optional[bool]
    processed_data: str

# 작업 생성 노드
def create_task(state):
    return {"task_data": "Data processing required"}

# 일시정지 노드
def pause_for_review(state):
    # 작업을 일시정지하고 인간 검토를 요청
    print("Waiting for human review...")
    return {"human_review": "Please review task data"}

# 검토 노드
async def review_task(state):
    # 인간 검토 기다리기 (실제 구현에서는 API 또는 사용자 인터페이스)
    await asyncio.sleep(1)  # 실제 구현에서는 사용자 입력 대기
    return {"human_review": "Review completed"}

# 진행 노드
def continue_processing(state):
    # 인간 승인 여부에 따라 진행
    if state.get("approval") is True:
        return {"processed_data": f"Processed: {state['task_data']}"}
    return {"processed_data": "Failed: Not approved"}

# 인간-함께-작업 워크플로우
def create_human_loop_workflow():
    workflow = StateGraph(HumanInLoopState)

    workflow.add_node("create", create_task)
    workflow.add_node("pause", pause_for_review)
    workflow.add_node("review", review_task)
    workflow.add_node("continue", continue_processing)

    workflow.add_edge("create", "pause")
    workflow.add_edge("pause", "review")
    workflow.add_edge("review", "continue")

    workflow.set_entry_point("create")
    workflow.set_finish_point("continue")

    return workflow.compile()

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

이

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

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

matias yoon — Sun, 24 May 2026 09:43:52 +0000

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

터미널에서 작동하는 AI 에이전트는 현대 개발 워크플로우에 필수적인 도구입니다. 이 가이드는 개발자가 로컬 환경에서 효율적으로 작동하는 AI 에이전트를 구축하고 활용하는 방법을 실질적인 코드와 명령어로 설명합니다.

1. CLI AI 에이전트 생태계

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

Aider: GitHub Copilot을 기반으로 한 코드 생성 도구

pip install aider
aider --help

Continue.dev: VS Code 확장에서 파생된 터미널 기반 에이전트

npm install -g continue
continue --help

OpenCode: 오픈소스 CLI 에이전트 프레임워크

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

사용자 정의 스크립트: 최소한의 커스터마이징이 필요한 경우

#!/usr/bin/env python3
import subprocess
import sys

def run_command(cmd):
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    return result.stdout

if __name__ == "__main__":
    if len(sys.argv) > 1:
        output = run_command(' '.join(sys.argv[1:]))
        print(output)

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

로컬 LLM을 터미널에서 사용하기 위해 API 서버를 설정해야 합니다:

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

# 모델 다운로드
ollama pull llama3.2

# API 서버 실행
ollama serve

API 엔드포인트 확인:

curl http://localhost:11434/api/tags

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

함수 호출 기능을 포함한 기본 CLI 에이전트를 구축합니다:

#!/usr/bin/env python3
# ai_agent.py

import requests
import json
import subprocess
import sys
from typing import Dict, List, Any

class TerminalAI:
    def __init__(self, api_url="http://localhost:11434/api/generate"):
        self.api_url = api_url
        self.conversation_history = []

    def call_llm(self, prompt: str, system_prompt: str = "") -> str:
        """LLM 호출 함수"""
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": prompt}
        ]

        payload = {
            "model": "llama3.2",
            "messages": messages,
            "stream": False
        }

        response = requests.post(
            self.api_url,
            json=payload,
            timeout=30
        )

        if response.status_code == 200:
            result = response.json()
            return result.get("response", "")
        else:
            raise Exception(f"LLM API Error: {response.status_code}")

    def code_review(self, code: str) -> str:
        """코드 리뷰 수행"""
        prompt = f"""
        다음 코드를 리뷰해 주세요:
        ```
{% endraw %}
python
        {code}
{% raw %}

        ```
        품질 향상을 위한 개선 제안을 제공해 주세요.
        """
        return self.call_llm(prompt, "You are a senior Python developer who reviews code quality")

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

# 사용 예시
if __name__ == "__main__":
    agent = TerminalAI()

    if len(sys.argv) > 1:
        if sys.argv[1] == "review":
            # 현재 디렉토리의 Python 파일을 읽어와 리뷰
            with open("app.py", "r") as f:
                code = f.read()
            review = agent.code_review(code)
            print(review)
        else:
            # 일반 명령어 실행
            command = " ".join(sys.argv[1:])
            output = agent.run_command(command)
            print(output)

실행 방법:

chmod +x ai_agent.py
python ai_agent.py review
python ai_agent.py "ls -la"

4. tmux와의 통합

터미널 분할 환경에서 에이전트와 상호작용:

# tmux 세션 시작
tmux new-session -s ai_agent

# 새로운 창 생성
tmux new-window

# 창 1: 에이전트 실행
python ai_agent.py

# 창 2: 코드 편집
vim app.py

에이전트의 출력을 tmux에 자동으로 전달하는 스크립트:

#!/usr/bin/env python3
# tmux_agent.py

import subprocess
import sys
import time

def send_to_tmux(window_name: str, content: str):
    """tmux 창에 내용 전송"""
    try:
        cmd = f"tmux send-keys -t {window_name} '{content}' Enter"
        subprocess.run(cmd, shell=True, check=True)
    except subprocess.CalledProcessError as e:
        print(f"Error sending to tmux: {e}")

# 사용 예시
if __name__ == "__main__":
    agent = TerminalAI()
    result = agent.call_llm("Hello, how are you?")
    send_to_tmux("ai_agent:0", result)

5. 맞춤형 도구 개발

코드 검색 도구

#!/usr/bin/env python3
# code_search.py

import os
import re
from pathlib import Path

class CodeSearcher:
    def __init__(self, root_dir: str = "."):
        self.root_dir = Path(root_dir)
        self.ignore_patterns = [
            ".git", "__pycache__", ".venv", "node_modules"
        ]

    def search_function(self, func_name: str) -> List[Dict]:
        """함수 이름으로 검색"""
        results = []

        for file_path in self.root_dir.rglob("*.py"):
            if any(pattern in str(file_path) for pattern in self.ignore_patterns):
                continue

            try:
                with open(file_path, 'r') as f:
                    content = f.read()

                # 함수 정의 검색 (정규식)
                pattern = rf'def\s+{func_name}\s*\([^)]*\)'
                matches = re.finditer(pattern, content)

                for match in matches:
                    results.append({
                        'file': str(file_path),
                        'line': content[:match.start()].count('\n') + 1,
                        'context': self._get_context(content, match.start())
                    })
            except Exception as e:
                continue

        return results

    def _get_context(self, content: str, position: int, lines: int = 3) -> str:
        """주변 컨텍스트 추출"""
        lines_list = content.split('\n')
        start_line = max(0, content[:position].count('\n') - lines)
        end_line = min(len(lines_list), start_line + lines * 2 + 1)

        return '\n'.join(lines_list[start_line:end_line])

# 사용 예시
searcher = CodeSearcher()
results = searcher.search_function("main")
for result in results:
    print(f"Found in {result['file']}: line {result['line']}")

Git 통합


python
#!/usr/bin/env python3
# git_integration.py

import subprocess
from datetime import datetime

class GitAgent:
    def __init__(self):
        self.repo_path = "."

    def get_recent_commits(self, count: int = 5) -> List[Dict]:
        """최근 커밋 목록 가져오기"""
        try:
            cmd = [
                "git", "-C", self.repo_path,
                "log", "--oneline", f"--{count}"
            ]
            result = subprocess.run(cmd, capture_output=True, text=True, check=True)

            commits = []
            for line in result.stdout.strip().split('\n'):
                if line:
                    sha, message = line.split(' ', 1)
                    commits.append({
                        'sha': sha,
                        'message': message,
                        'timestamp': datetime.now().isoformat()
                    })

            return commits
        except subprocess.CalledProcessError:
            return []

    def get_diff(self, file_path: str) -> str:
        """파일 변경사항 확인"""
        try:
            cmd = ["git", "-C",

---

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