🦙 Llama 완전 가이드

import asyncio
from openai import OpenAI, AsyncOpenAI

# ── 동기 클라이언트 ───────────────────────────────
client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")

# 기본 호출
resp = client.chat.completions.create(
    model="llama3.1",
    messages=[
        {"role": "system", "content": "당신은 시니어 Python 개발자입니다."},
        {"role": "user",   "content": "asyncio.gather와 asyncio.wait의 차이를 설명해줘"},
    ],
    temperature=0.7,
    max_tokens=1024,
)
print(resp.choices[0].message.content)

# ── 스트리밍 ─────────────────────────────────────
async def stream_response(prompt: str):
    aclient = AsyncOpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
    async with aclient.chat.completions.stream(
        model="llama3.1",
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                print(delta, end="", flush=True)

asyncio.run(stream_response("Python 비동기 패턴을 설명해줘"))

import json
from openai import OpenAI

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")

# ── 도구 함수 정의 ────────────────────────────────
def get_weather(city: str, unit: str = "celsius") -> str:
    """도시 날씨 조회 (실제 구현 시 API 호출)"""
    return json.dumps({"city": city, "temp": 23, "unit": unit, "condition": "맑음"})

def search_docs(query: str, top_k: int = 3) -> str:
    """내부 문서 검색"""
    return json.dumps({"results": [f"문서: {query} 관련 내용 {i}" for i in range(top_k)]})

TOOLS = {
    "get_weather": get_weather,
    "search_docs": search_docs,
}

TOOL_SCHEMAS = [
    {"type": "function", "function": {
        "name": "get_weather",
        "description": "특정 도시의 현재 날씨를 조회합니다",
        "parameters": {"type": "object",
            "properties": {
                "city": {"type": "string", "description": "도시명 (예: 서울)"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
            }, "required": ["city"]},
    }},
    {"type": "function", "function": {
        "name": "search_docs",
        "description": "내부 지식베이스에서 관련 문서를 검색합니다",
        "parameters": {"type": "object",
            "properties": {
                "query": {"type": "string", "description": "검색 질의"},
                "top_k": {"type": "integer", "description": "반환할 결과 수", "default": 3},
            }, "required": ["query"]},
    }},
]

# ── ReAct 루프 ────────────────────────────────────
def run_agent(user_query: str) -> str:
    messages = [
        {"role": "system", "content": "당신은 도움이 되는 AI 어시스턴트입니다. 필요할 때 도구를 사용하세요."},
        {"role": "user", "content": user_query},
    ]

    for step in range(5):
        resp = client.chat.completions.create(
            model="llama3.1",
            messages=messages,
            tools=TOOL_SCHEMAS,
            tool_choice="auto",
        )
        msg = resp.choices[0].message
        messages.append(msg)

        if not msg.tool_calls:          # 최종 답변
            return msg.content

        for tc in msg.tool_calls:       # 도구 실행
            args   = json.loads(tc.function.arguments)
            result = TOOLS[tc.function.name](**args)
            print(f"[Tool] {tc.function.name}({args})")
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": result,
            })
    return "최대 스텝 초과"

print(run_agent("서울 날씨를 알려주고, 우산이 필요한지 판단해줘"))

import json
from openai import OpenAI
from pydantic import BaseModel, Field

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")

class SentimentResult(BaseModel):
    sentiment: str     = Field(description="positive|negative|neutral")
    score:     float   = Field(description="0.0~1.0 신뢰도")
    reason:    str     = Field(description="판단 근거 한 문장")
    keywords:  list[str] = Field(description="핵심 키워드 목록")

def analyze_sentiment(text: str) -> SentimentResult:
    schema = SentimentResult.model_json_schema()
    resp = client.chat.completions.create(
        model="llama3.1",
        messages=[
            {"role": "system",
             "content": f"감정 분석 결과를 반드시 다음 JSON 스키마로 반환하세요:\n{json.dumps(schema, ensure_ascii=False)}"},
            {"role": "user", "content": f"분석할 텍스트: {text}"},
        ],
        response_format={"type": "json_object"},  # JSON 강제
    )
    raw = json.loads(resp.choices[0].message.content)
    return SentimentResult(**raw)   # Pydantic 검증

result = analyze_sentiment("이 제품은 정말 실망스럽네요. 배송도 늦고 품질도 별로예요.")
print(f"감정: {result.sentiment} (신뢰도: {result.score:.0%})")
print(f"근거: {result.reason}")
print(f"키워드: {', '.join(result.keywords)}")

from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.chat_models import ChatOllama
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain.text_splitter import RecursiveCharacterTextSplitter

# ── 임베딩 모델 (로컬 실행) ───────────────────────
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-m3",          # 한국어 포함 다국어 지원
    model_kwargs={"device": "cuda"},
    encode_kwargs={"normalize_embeddings": True},
)

# ── 문서 인덱싱 ──────────────────────────────────
documents = [
    "AI DevOps Korea는 AI · Cloud · DevOps 학습 플랫폼입니다.",
    "k6는 JavaScript로 작성하는 현대적인 부하 테스트 도구입니다.",
    "JMeter는 Java 기반의 GUI 지원 부하 테스트 도구입니다.",
]
splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=50)
docs     = splitter.create_documents(documents)
vectordb = Chroma.from_documents(docs, embeddings, persist_directory="./chroma_db")

# ── RAG 체인 (LCEL) ──────────────────────────────
llm = ChatOllama(model="llama3.1", temperature=0.1)

prompt = ChatPromptTemplate.from_template("""다음 컨텍스트를 바탕으로 질문에 답하세요.
컨텍스트에 없는 내용은 모른다고 하세요.

컨텍스트: {context}

질문: {question}""")

rag_chain = (
    {"context": vectordb.as_retriever(search_kwargs={"k": 3}), "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

answer = rag_chain.invoke("k6와 JMeter의 차이점은?")
print(answer)

import asyncio
from openai import AsyncOpenAI

# vLLM은 OpenAI SDK를 그대로 사용
client = AsyncOpenAI(base_url="http://localhost:8000/v1", api_key="none")

# ── 배치 추론 (높은 처리량) ──────────────────────
async def batch_inference(prompts: list[str]) -> list[str]:
    tasks = [
        client.chat.completions.create(
            model="meta-llama/Meta-Llama-3.1-8B-Instruct",
            messages=[{"role": "user", "content": p}],
            max_tokens=512,
        )
        for p in prompts
    ]
    responses = await asyncio.gather(*tasks)
    return [r.choices[0].message.content for r in responses]

# ── 서버 성능 모니터링 ────────────────────────────
import httpx

async def check_metrics():
    async with httpx.AsyncClient() as http:
        metrics = await http.get("http://localhost:8000/metrics")
        # Prometheus 형식으로 TPS, 지연시간, 큐 길이 확인
        print(metrics.text[:500])

asyncio.run(batch_inference(["Python이란?", "Go란?", "Rust란?"]))

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig, TrainingArguments
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from trl import SFTTrainer
from datasets import load_dataset

MODEL_ID = "meta-llama/Meta-Llama-3.1-8B-Instruct"

# ── QLoRA: 4-bit 양자화 로딩 ─────────────────────
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True,
)
model = AutoModelForCausalLM.from_pretrained(
    MODEL_ID, quantization_config=bnb_config, device_map="auto"
)
model = prepare_model_for_kbit_training(model)

# ── LoRA 설정 ────────────────────────────────────
lora_config = LoraConfig(
    r=16,                    # 랭크 (클수록 표현력↑, 메모리↑)
    lora_alpha=32,           # 스케일링 = alpha/r
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj",
                    "gate_proj", "up_proj", "down_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM",
)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# trainable: 41,943,040 || all: 8,072,204,288 || 0.52%

# ── 학습 ─────────────────────────────────────────
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
dataset   = load_dataset("json", data_files="train.jsonl", split="train")

trainer = SFTTrainer(
    model=model,
    train_dataset=dataset,
    tokenizer=tokenizer,
    args=TrainingArguments(
        output_dir="./llama-finetuned",
        num_train_epochs=3,
        per_device_train_batch_size=4,
        gradient_accumulation_steps=4,
        learning_rate=2e-4,
        bf16=True,
        save_steps=100,
        logging_steps=10,
    ),
)
trainer.train()

# LoRA 어댑터만 저장 (수십 MB)
model.save_pretrained("./lora-adapter")

# ── 추론 시 어댑터 로드 ───────────────────────────
from peft import PeftModel
base_model = AutoModelForCausalLM.from_pretrained(MODEL_ID, torch_dtype=torch.bfloat16)
model      = PeftModel.from_pretrained(base_model, "./lora-adapter")