Llama 실전 예제 모음

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from ollama import Client
from collections import defaultdict
import uuid

app = FastAPI(title="Llama Chat API")
ollama = Client()

# 세션별 대화 히스토리
sessions: dict[str, list[dict]] = defaultdict(list)

class ChatRequest(BaseModel):
    message: str
    session_id: str | None = None
    system: str = "당신은 친절하고 유능한 AI 어시스턴트입니다."
    model: str = "llama3.1"

class ChatResponse(BaseModel):
    reply: str
    session_id: str

@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest):
    sid = req.session_id or str(uuid.uuid4())
    history = sessions[sid]

    # 첫 메시지면 시스템 프롬프트 추가
    if not history:
        history.append({"role": "system", "content": req.system})

    history.append({"role": "user", "content": req.message})

    resp = ollama.chat(model=req.model, messages=history)
    reply = resp["message"]["content"]

    history.append({"role": "assistant", "content": reply})
    return ChatResponse(reply=reply, session_id=sid)

@app.get("/history/{session_id}")
async def get_history(session_id: str):
    return sessions.get(session_id, [])

@app.delete("/history/{session_id}")
async def clear_history(session_id: str):
    sessions.pop(session_id, None)
    return {"cleared": session_id}

# 실행: uvicorn chat_server:app --reload --port 8000python

from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.llms import Ollama
from langchain.chains import RetrievalQA
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document

# ── 1. 로컬 임베딩 모델 (무료, 다운로드 자동)
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)

# ── 2. 문서 로드 및 청킹
def load_documents(texts: list[str]) -> list[Document]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        separators=["\n\n", "\n", ".", " "],
    )
    docs = [Document(page_content=t) for t in texts]
    return splitter.split_documents(docs)

# ── 3. 벡터 DB 구축
def build_vectorstore(docs: list[Document]) -> Chroma:
    return Chroma.from_documents(
        docs,
        embeddings,
        persist_directory="./chroma_db",
    )

# ── 4. QA 체인 생성
def create_qa_chain(vectorstore: Chroma) -> RetrievalQA:
    llm = Ollama(model="llama3.1", temperature=0)
    return RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
        return_source_documents=True,
    )

# ── 사용 예시
if __name__ == "__main__":
    # 문서 추가
    sample_docs = [
        "TestForge는 개발자를 위한 부하 테스트 플랫폼입니다. k6, JMeter, nGrinder를 지원합니다.",
        "k6는 JavaScript로 스크립트를 작성하며 CI/CD 통합이 뛰어납니다.",
        "JMeter는 GUI 기반 도구로 다양한 프로토콜을 지원합니다.",
        "nGrinder는 웹 UI 기반의 분산 부하 테스트 플랫폼입니다.",
    ]

    docs = load_documents(sample_docs)
    vs = build_vectorstore(docs)
    qa = create_qa_chain(vs)

    questions = [
        "TestForge가 지원하는 부하 테스트 도구는?",
        "k6의 특징은 무엇인가요?",
    ]
    for q in questions:
        result = qa.invoke({"query": q})
        print(f"Q: {q}")
        print(f"A: {result['result']}\n")python

from ollama import Client
from pydantic import BaseModel
import json

ollama = Client()

class ReviewResult(BaseModel):
    summary: str
    issues: list[str]
    suggestions: list[str]
    security_concerns: list[str]
    score: int  # 1-10

REVIEW_SYSTEM = """당신은 시니어 소프트웨어 엔지니어입니다.
코드를 검토하고 다음 JSON 형식으로만 응답하세요:
{
  "summary": "전체 코드 요약 (1-2문장)",
  "issues": ["버그나 문제점 목록"],
  "suggestions": ["개선 제안 목록"],
  "security_concerns": ["보안 취약점 목록"],
  "score": 1-10 (코드 품질 점수)
}"""

def review_code(code: str, language: str = "python") -> ReviewResult:
    prompt = f"다음 {language} 코드를 리뷰해주세요:\n\n```{language}\n{code}\n```"

    resp = ollama.chat(
        model="llama3.1",
        messages=[
            {"role": "system", "content": REVIEW_SYSTEM},
            {"role": "user", "content": prompt},
        ],
        format="json",
    )

    data = json.loads(resp["message"]["content"])
    return ReviewResult(**data)

# CLI 도구로 사용
if __name__ == "__main__":
    import sys
    if len(sys.argv) < 2:
        print("Usage: python code_review.py ")
        sys.exit(1)

    with open(sys.argv[1]) as f:
        code = f.read()

    result = review_code(code)
    print(f"\n📋 요약: {result.summary}")
    print(f"⭐ 점수: {result.score}/10")

    if result.issues:
        print("\n❌ 문제점:")
        for i in result.issues: print(f"  • {i}")

    if result.suggestions:
        print("\n💡 개선 제안:")
        for s in result.suggestions: print(f"  • {s}")

    if result.security_concerns:
        print("\n🔒 보안 취약점:")
        for sc in result.security_concerns: print(f"  • {sc}")python

from ollama import Client
from pydantic import BaseModel
import json

ollama = Client()

# ── 뉴스 기사 정보 추출
class NewsInfo(BaseModel):
    title: str
    summary: str
    category: str
    sentiment: str  # positive / negative / neutral
    key_entities: list[str]
    importance_score: int  # 1-5

def extract_news_info(article: str) -> NewsInfo:
    resp = ollama.chat(
        model="llama3.1",
        messages=[{
            "role": "system",
            "content": """텍스트에서 정보를 추출해 JSON으로 반환하세요:
{
  "title": "기사 제목",
  "summary": "한 문장 요약",
  "category": "기술/경제/사회/정치/스포츠 중 하나",
  "sentiment": "positive/negative/neutral",
  "key_entities": ["주요 인물/기관/장소 목록"],
  "importance_score": 1-5
}"""
        }, {
            "role": "user",
            "content": f"다음 기사를 분석해주세요:\n\n{article}"
        }],
        format="json",
    )
    return NewsInfo(**json.loads(resp["message"]["content"]))

# ── 이력서 파싱
class ResumeInfo(BaseModel):
    name: str
    skills: list[str]
    years_experience: int
    education: str
    summary: str

def parse_resume(resume_text: str) -> ResumeInfo:
    resp = ollama.chat(
        model="llama3.1",
        messages=[{
            "role": "system",
            "content": '이력서에서 정보를 추출해 JSON으로 반환: {"name","skills","years_experience","education","summary"}'
        }, {
            "role": "user",
            "content": resume_text
        }],
        format="json",
    )
    return ResumeInfo(**json.loads(resp["message"]["content"]))

# 테스트
article = """
OpenAI가 새로운 GPT-5 모델을 발표했다. 이 모델은 이전 버전보다 3배 빠르고
정확도가 크게 향상되었다. CEO 샘 올트만은 "AI 산업의 새로운 전환점"이라고 밝혔다.
"""
info = extract_news_info(article)
print(f"제목: {info.title}")
print(f"카테고리: {info.category}, 감성: {info.sentiment}")
print(f"핵심 엔티티: {info.key_entities}")python

from fastapi import FastAPI
from fastapi.responses import StreamingResponse, HTMLResponse
from pydantic import BaseModel
from ollama import Client
import json, asyncio

app = FastAPI()
ollama = Client()

class StreamRequest(BaseModel):
    prompt: str
    model: str = "llama3.1"
    system: str = "당신은 유능한 AI 어시스턴트입니다."

@app.post("/stream")
async def stream_chat(req: StreamRequest):
    async def generate():
        stream = ollama.chat(
            model=req.model,
            messages=[
                {"role": "system", "content": req.system},
                {"role": "user", "content": req.prompt},
            ],
            stream=True,
        )
        for chunk in stream:
            token = chunk["message"]["content"]
            if token:
                yield f"data: {json.dumps({'token': token})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

# 테스트용 UI
@app.get("/", response_class=HTMLResponse)
async def ui():
    return """
<!DOCTYPE html><html lang="ko"><body style="font-family:system-ui;max-width:800px;margin:2rem auto;padding:1rem">
<h2>🦙 Llama 스트리밍 데모</h2>
<textarea id="prompt" rows="3" style="width:100%;padding:.5rem;font-size:1rem" placeholder="질문을 입력하세요..."></textarea>
<button onclick="send()" style="margin-top:.5rem;padding:.5rem 1.5rem;background:#7c3aed;color:#fff;border:none;border-radius:6px;cursor:pointer">전송</button>
<div id="output" style="margin-top:1rem;padding:1rem;background:#f5f3ff;border-radius:8px;min-height:100px;white-space:pre-wrap;"></div>
<script>
async function send() {
  const prompt = document.getElementById('prompt').value;
  const out = document.getElementById('output');
  out.textContent = '';
  const res = await fetch('/stream', {
    method: 'POST',
    headers: {'Content-Type': 'application/json'},
    body: JSON.stringify({prompt})
  });
  const reader = res.body.getReader();
  const decoder = new TextDecoder();
  while (true) {
    const {done, value} = await reader.read();
    if (done) break;
    const lines = decoder.decode(value).split('\n');
    for (const line of lines) {
      if (line.startsWith('data: ') && line !== 'data: [DONE]') {
        const d = JSON.parse(line.slice(6));
        out.textContent += d.token;
      }
    }
  }
}
</script>
</body></html>"""

# 실행: uvicorn streaming_app:app --reloadpython

from ollama import Client

ollama = Client()

def analyze_long_document(document: str, task: str = "요약") -> str:
    """
    Llama 3.1의 128K 컨텍스트로 긴 문서 처리.
    task: "요약" | "핵심_추출" | "질의응답" | "번역"
    """
    task_prompts = {
        "요약": "다음 문서를 5개 핵심 항목으로 요약해주세요.",
        "핵심_추출": "다음 문서에서 가장 중요한 사실 10가지를 추출해주세요.",
        "번역": "다음 문서를 한국어로 번역해주세요.",
    }

    system = task_prompts.get(task, task)
    char_count = len(document)
    token_estimate = char_count // 4  # 대략적 토큰 수 추정

    print(f"📄 문서 길이: {char_count:,}자 (약 {token_estimate:,} 토큰)")
    if token_estimate > 100000:
        print("⚠️  매우 긴 문서입니다. 처리 시간이 걸릴 수 있습니다.")

    resp = ollama.chat(
        model="llama3.1",
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": document},
        ],
        options={
            "num_ctx": 131072,  # 128K 컨텍스트 명시
            "temperature": 0.3,
        },
    )
    return resp["message"]["content"]

# PDF에서 텍스트 추출 후 분석 (pymupdf 필요: pip install pymupdf)
def analyze_pdf(pdf_path: str) -> str:
    try:
        import fitz  # pymupdf
        doc = fitz.open(pdf_path)
        text = "\n".join(page.get_text() for page in doc)
        return analyze_long_document(text, "요약")
    except ImportError:
        return "pip install pymupdf 를 먼저 실행하세요."

# 여러 파일 합쳐서 분석
def analyze_codebase(file_paths: list[str]) -> str:
    combined = ""
    for path in file_paths:
        with open(path) as f:
            combined += f"# === {path} ===\n{f.read()}\n\n"
    return analyze_long_document(
        combined,
        "다음 코드베이스를 분석해 아키텍처, 주요 컴포넌트, 개선점을 설명해주세요."
    )python