🐍 Python AI 가이드

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

# ── 기본 비동기 호출 ──────────────────────────────
async def call_llm(prompt: str, model: str = "gpt-4o-mini") -> str:
    resp = await client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
    )
    return resp.choices[0].message.content

# ── 스트리밍 (토큰 단위 실시간 출력) ─────────────
async def stream_llm(prompt: str):
    async with client.chat.completions.stream(
        model="gpt-4o",
        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)

# ── 여러 프롬프트 동시 병렬 호출 ─────────────────
async def batch_call(prompts: list[str]) -> list[str]:
    return await asyncio.gather(*[call_llm(p) for p in prompts])

async def main():
    prompts = ["Python이란?", "Go란?", "Rust란?"]
    # 직렬: 순차 실행 (~3초)  vs  병렬: 동시 실행 (~1초)
    serial   = [await call_llm(p) for p in prompts]
    parallel = await batch_call(prompts)          # 약 3배 빠름

asyncio.run(main())

import instructor
from openai import OpenAI
from pydantic import BaseModel, Field
from enum import Enum

client = instructor.from_openai(OpenAI())

class Priority(str, Enum):
    HIGH   = "high"
    MEDIUM = "medium"
    LOW    = "low"

class Task(BaseModel):
    title:    str        = Field(description="작업 제목")
    priority: Priority   = Field(description="우선순위")
    deadline: str | None = Field(description="마감일 YYYY-MM-DD")
    subtasks: list[str]  = Field(default_factory=list)

class ProjectPlan(BaseModel):
    goal:        str        = Field(description="프로젝트 목표")
    tasks:       list[Task] = Field(description="작업 목록")
    total_weeks: int        = Field(description="예상 소요 기간(주)")

# LLM이 자유 텍스트 대신 검증된 Pydantic 객체를 반환
plan = client.chat.completions.create(
    model="gpt-4o",
    response_model=ProjectPlan,
    messages=[{"role": "user",
               "content": "AI 챗봇 서비스를 3개월 안에 출시하는 계획을 짜줘"}]
)

for task in plan.tasks:
    print(f"[{task.priority.upper()}] {task.title} — 마감: {task.deadline}")
    for sub in task.subtasks:
        print(f"  • {sub}")

import json, math
from openai import OpenAI
from typing import Callable

client = OpenAI()

# ── 도구 정의 ─────────────────────────────────────
def search_web(query: str) -> str:
    """웹에서 최신 정보를 검색합니다."""
    return f"[검색 결과] {query}: 관련 최신 정보..."   # 실제: Tavily API

def calculate(expression: str) -> str:
    """수식을 안전하게 계산합니다."""
    try:
        allowed = {k: v for k, v in math.__dict__.items() if not k.startswith("_")}
        return str(eval(expression, {"__builtins__": {}}, allowed))
    except Exception as e:
        return f"오류: {e}"

TOOLS: dict[str, Callable] = {"search_web": search_web, "calculate": calculate}
SCHEMAS = [
    {"type": "function", "function": {
        "name": "search_web", "description": "웹 검색",
        "parameters": {"type": "object",
                       "properties": {"query": {"type": "string"}},
                       "required": ["query"]}}},
    {"type": "function", "function": {
        "name": "calculate", "description": "수식 계산",
        "parameters": {"type": "object",
                       "properties": {"expression": {"type": "string"}},
                       "required": ["expression"]}}},
]

# ── ReAct 루프 ────────────────────────────────────
def run_agent(query: str, max_steps: int = 5) -> str:
    messages = [{"role": "user", "content": query}]
    for _ in range(max_steps):
        resp = client.chat.completions.create(
            model="gpt-4o", messages=messages, tools=SCHEMAS
        )
        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}) → {result[:60]}")
            messages.append({"role": "tool", "tool_call_id": tc.id,
                              "content": result})
    return "최대 스텝 초과"

print(run_agent("삼성전자 현재 주가에 2.5를 곱하면?"))

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def summarize_one(sem: asyncio.Semaphore, doc: str, idx: int) -> dict:
    async with sem:   # 동시 최대 N개 호출 제한
        try:
            resp = await client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": f"요약:\n{doc}"}],
                max_tokens=200,
            )
            return {"idx": idx, "summary": resp.choices[0].message.content,
                    "ok": True}
        except Exception as e:
            return {"idx": idx, "error": str(e), "ok": False}

async def batch_summarize(docs: list[str], concurrency: int = 10) -> list[dict]:
    sem     = asyncio.Semaphore(concurrency)
    results = await asyncio.gather(
        *[summarize_one(sem, doc, i) for i, doc in enumerate(docs)]
    )
    return sorted(results, key=lambda x: x["idx"])

# 1,000개 문서를 직렬 대비 ~10배 빠르게 처리
docs    = [f"문서 {i} 내용..." for i in range(100)]
results = asyncio.run(batch_summarize(docs))
print(f"완료: {sum(r['ok'] for r in results)}, "
      f"실패: {sum(not r['ok'] for r in results)}")

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from openai import AsyncOpenAI
import json

app    = FastAPI(title="AI Agent API")
client = AsyncOpenAI()

class ChatRequest(BaseModel):
    message:    str
    model:      str = "gpt-4o"
    session_id: str = "default"

# ── SSE 스트리밍 엔드포인트 ──────────────────────
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    async def generate():
        async with client.chat.completions.stream(
            model=req.model,
            messages=[{"role": "user", "content": req.message}],
        ) as stream:
            async for chunk in stream:
                delta = chunk.choices[0].delta.content
                if delta:
                    yield f"data: {json.dumps({'token': delta})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache"})

# ── 구조화 출력 엔드포인트 ────────────────────────
class AnalysisResult(BaseModel):
    sentiment:  str
    key_points: list[str]
    confidence: float

@app.post("/analyze", response_model=AnalysisResult)
async def analyze(req: ChatRequest):
    import instructor
    inst = instructor.from_openai(AsyncOpenAI(), mode=instructor.Mode.JSON)
    return await inst.chat.completions.create(
        model=req.model, response_model=AnalysisResult,
        messages=[{"role": "user", "content": f"분석:\n{req.message}"}]
    )

# 실행: uvicorn agent_server:app --reload --port 8000

import pytest
from unittest.mock import patch, MagicMock
from pydantic import BaseModel, ValidationError

from agent_loop import calculate

# ── 도구 단위 테스트 ──────────────────────────────
def test_calculate_basic():
    assert calculate("2 + 3") == "5"
    assert calculate("10 ** 2") == "100"
    assert float(calculate("math.sqrt(2)")) == pytest.approx(1.414, rel=1e-3)

def test_calculate_blocks_dangerous():
    assert "오류" in calculate("__import__('os').system('ls')")

# ── LLM 호출 Mock 테스트 ──────────────────────────
def test_agent_calls_tool_then_answers():
    tool_call = MagicMock()
    tool_call.function.name      = "calculate"
    tool_call.function.arguments = '{"expression": "2 + 2"}'
    tool_call.id                 = "call_1"

    msg1 = MagicMock(); msg1.tool_calls = [tool_call]; msg1.content = None
    msg2 = MagicMock(); msg2.tool_calls = None;        msg2.content = "4입니다"

    r1 = MagicMock(); r1.choices[0].message = msg1
    r2 = MagicMock(); r2.choices[0].message = msg2

    with patch("agent_loop.client.chat.completions.create",
               side_effect=[r1, r2]):
        from agent_loop import run_agent
        assert "4" in run_agent("2 더하기 2는?")

# ── 구조화 출력 스키마 검증 ───────────────────────
class AgentOut(BaseModel):
    answer:     str
    sources:    list[str]
    confidence: float

def test_valid_schema():
    out = AgentOut(answer="결과", sources=["doc.pdf"], confidence=0.9)
    assert 0.0 <= out.confidence <= 1.0

def test_rejects_out_of_range():
    with pytest.raises(ValidationError):
        AgentOut(answer="", sources=[], confidence=1.5)