Gemma 실전 예제 모음

from fastapi import FastAPI
from pydantic import BaseModel
from ollama import Client, AsyncClient
import asyncio
from functools import lru_cache
import hashlib

app = FastAPI(title="Gemma Lightweight Server")
sync_ollama = Client()
async_ollama = AsyncClient()

class InferenceRequest(BaseModel):
    prompt: str
    max_tokens: int = 256
    use_cache: bool = True
    model: str = "gemma2:2b"  # 2B 기본값

# 응답 캐시 (동일 프롬프트 재사용)
_cache: dict[str, str] = {}

def get_cache_key(prompt: str, model: str) -> str:
    return hashlib.md5(f"{model}:{prompt}".encode()).hexdigest()

@app.post("/infer")
async def infer(req: InferenceRequest):
    key = get_cache_key(req.prompt, req.model)

    if req.use_cache and key in _cache:
        return {"response": _cache[key], "cached": True}

    resp = await async_ollama.generate(
        model=req.model,
        prompt=req.prompt,
        options={"num_predict": req.max_tokens, "temperature": 0.7},
    )

    result = resp["response"]
    if req.use_cache:
        _cache[key] = result

    return {"response": result, "cached": False}

# 배치 처리 (여러 프롬프트 동시)
@app.post("/batch")
async def batch_infer(prompts: list[str], model: str = "gemma2:2b"):
    async def single(p: str) -> str:
        r = await async_ollama.generate(model=model, prompt=p)
        return r["response"]

    # 동시에 최대 4개 처리
    semaphore = asyncio.Semaphore(4)
    async def limited(p: str) -> str:
        async with semaphore:
            return await single(p)

    results = await asyncio.gather(*[limited(p) for p in prompts])
    return {"results": list(results), "count": len(results)}

# 헬스체크 + 모델 상태
@app.get("/health")
def health():
    models = sync_ollama.list()
    available = [m["name"] for m in models.get("models", [])]
    return {
        "status": "ok",
        "available_models": available,
        "cached_responses": len(_cache),
    }python

from langchain_community.llms import Ollama
from langchain.chains.summarize import load_summarize_chain
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
from langchain.prompts import PromptTemplate

llm = Ollama(model="gemma2", temperature=0.3)

# 요약 프롬프트
MAP_PROMPT = PromptTemplate(
    input_variables=["text"],
    template="다음 텍스트를 3~5문장으로 핵심만 요약하세요:\n\n{text}\n\n요약:"
)

REDUCE_PROMPT = PromptTemplate(
    input_variables=["text"],
    template="""다음은 긴 문서의 각 부분을 요약한 내용입니다.
이를 통합해 전체 문서의 완성된 요약을 작성하세요.

{text}

최종 요약 (불릿 포인트 형식으로):"""
)

def summarize_document(text: str, chunk_size: int = 1000) -> dict:
    """
    긴 문서를 MapReduce 방식으로 요약합니다.
    Returns: {"summary": str, "chunks": int, "method": str}
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=100,
    )

    docs = splitter.create_documents([text])

    if len(docs) == 1:
        # 짧은 문서는 직접 요약
        chain = load_summarize_chain(llm, chain_type="stuff", prompt=MAP_PROMPT)
        method = "direct"
    else:
        # 긴 문서는 MapReduce
        chain = load_summarize_chain(
            llm,
            chain_type="map_reduce",
            map_prompt=MAP_PROMPT,
            combine_prompt=REDUCE_PROMPT,
        )
        method = "map_reduce"

    result = chain.invoke({"input_documents": docs})
    return {
        "summary": result["output_text"],
        "chunks": len(docs),
        "method": method,
        "original_length": len(text),
    }

# FastAPI 통합
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class SummarizeRequest(BaseModel):
    text: str
    style: str = "bullet"  # bullet | paragraph | brief

@app.post("/summarize")
async def summarize(req: SummarizeRequest):
    style_instructions = {
        "bullet": "• 불릿 포인트 형식으로",
        "paragraph": "단락 형식으로",
        "brief": "3문장 이내로 간략하게",
    }
    result = summarize_document(req.text)
    return resultpython

from fastapi import FastAPI
from pydantic import BaseModel
from ollama import Client
import json

app = FastAPI(title="Sentiment Analysis API")
ollama = Client()

class SentimentRequest(BaseModel):
    text: str
    detailed: bool = True

class SentimentResponse(BaseModel):
    text: str
    sentiment: str        # positive / negative / neutral / mixed
    score: float          # -1.0 ~ 1.0
    emotions: list[str]   # 구체적 감정 (기쁨, 분노, 슬픔 등)
    aspects: dict         # 측면별 감성 (제품, 서비스 등)
    summary: str

@app.post("/analyze", response_model=SentimentResponse)
async def analyze_sentiment(req: SentimentRequest):
    resp = ollama.chat(
        model="gemma2:2b",
        messages=[{
            "role": "system",
            "content": """텍스트의 감성을 분석해 JSON으로 반환하세요:
{
  "sentiment": "positive/negative/neutral/mixed",
  "score": -1.0에서 1.0 사이 숫자,
  "emotions": ["구체적 감정 목록"],
  "aspects": {"측면": "감성"},
  "summary": "감성 분석 한 줄 요약"
}"""
        }, {
            "role": "user",
            "content": req.text,
        }],
        format="json",
    )

    data = json.loads(resp["message"]["content"])
    return SentimentResponse(text=req.text, **data)

# 배치 분석
@app.post("/batch-analyze")
async def batch_analyze(texts: list[str]):
    results = []
    for text in texts:
        r = await analyze_sentiment(SentimentRequest(text=text))
        results.append(r)
    return resultspython

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

app = FastAPI()
ollama = Client()

class CodeRequest(BaseModel):
    task: str        # "generate" | "explain" | "refactor" | "test"
    code: str = ""
    description: str = ""
    language: str = "python"

TASK_PROMPTS = {
    "generate": "다음 설명을 기반으로 {lang} 코드를 작성하세요:\n{desc}\n\n코드만 출력하세요:",
    "explain": "다음 {lang} 코드를 단계별로 설명하세요:\n```{lang}\n{code}\n```",
    "refactor": "다음 {lang} 코드를 더 효율적으로 리팩토링하세요. 개선 사유도 설명하세요:\n```{lang}\n{code}\n```",
    "test": "다음 {lang} 코드에 대한 단위 테스트를 작성하세요:\n```{lang}\n{code}\n```",
}

@app.post("/code/stream")
async def code_stream(req: CodeRequest):
    prompt = TASK_PROMPTS[req.task].format(
        lang=req.language,
        code=req.code,
        desc=req.description,
    )

    def generate():
        stream = ollama.generate(
            model="gemma2",
            prompt=prompt,
            stream=True,
            options={"temperature": 0.2 if req.task != "generate" else 0.7},
        )
        for chunk in stream:
            token = chunk.get("response", "")
            if token:
                yield f"data: {json.dumps({'token': token})}\n\n"
        yield "data: [DONE]\n\n"

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

@app.post("/code")
async def code_sync(req: CodeRequest):
    prompt = TASK_PROMPTS[req.task].format(
        lang=req.language, code=req.code, desc=req.description
    )
    resp = ollama.generate(model="gemma2:2b", prompt=prompt,
                           options={"temperature": 0.3})
    return {"result": resp["response"]}python

import os
os.environ["KERAS_BACKEND"] = "jax"  # 또는 "torch"

import keras
import keras_nlp

# ── 1. Gemma 2B 로드 (HuggingFace 토큰 필요)
gemma_lm = keras_nlp.models.GemmaCausalLM.from_preset("gemma2_instruct_2b_en")
gemma_lm.summary()

# ── 2. LoRA 활성화 (전체 파라미터의 ~0.3%만 학습)
gemma_lm.backbone.enable_lora(rank=4)
gemma_lm.preprocessor.sequence_length = 256

# ── 3. 학습 데이터 준비 (Q&A 형식)
train_data = [
    "user\nPython의 GIL이란?\n\nmodel\nGIL(Global Interpreter Lock)은 Python 인터프리터가 한 번에 하나의 스레드만 실행하도록 하는 뮤텍스입니다.",
    "user\nFastAPI의 장점은?\n\nmodel\nFastAPI는 자동 문서 생성, 높은 성능, 타입 힌팅 기반 검증이 강점입니다.",
    # 실제로는 수백~수천 개의 예제 필요
]

# ── 4. 컴파일 및 학습
gemma_lm.compile(
    loss=keras.losses.SparseCategoricalCrossentropy(from_logits=True),
    optimizer=keras.optimizers.Adam(learning_rate=5e-5),
    weighted_metrics=[keras.metrics.SparseCategoricalAccuracy()],
)

gemma_lm.fit(train_data, epochs=2, batch_size=1)

# ── 5. 추론 테스트
prompt = "user\nPython asyncio를 설명해주세요.\n\nmodel\n"
print(gemma_lm.generate(prompt, max_length=512))

# ── 6. 저장
gemma_lm.save_weights("gemma_finetuned.weights.h5")python