🔍 DeepSeek 완전 가이드

import re
from dataclasses import dataclass
from openai import OpenAI

client = OpenAI(
    api_key="YOUR_DEEPSEEK_KEY",
    base_url="https://api.deepseek.com",
)

@dataclass
class R1Response:
    thinking:   str          # <think> 내부 추론 과정
    answer:     str          # 최종 답변
    think_len:  int          # 추론 토큰 수 (복잡도 지표)

def call_r1(prompt: str, system: str = "") -> R1Response:
    messages = []
    if system:
        messages.append({"role": "system", "content": system})
    messages.append({"role": "user", "content": prompt})

    resp = client.chat.completions.create(
        model="deepseek-reasoner",
        messages=messages,
    )
    msg = resp.choices[0].message

    # DeepSeek API는 reasoning_content를 별도 필드로 제공
    thinking = getattr(msg, "reasoning_content", "") or ""

    # Ollama 등 로컬 모델은 <think> 태그로 반환
    if not thinking:
        match = re.search(r"<think>(.*?)</think>", msg.content, re.DOTALL)
        thinking = match.group(1).strip() if match else ""

    answer = re.sub(r"<think>.*?</think>", "", msg.content, flags=re.DOTALL).strip()
    return R1Response(thinking=thinking, answer=answer, think_len=len(thinking.split()))

# ── 사용 예시 ─────────────────────────────────────
result = call_r1(
    prompt="피보나치 수열의 40번째 항을 동적 프로그래밍으로 계산하는 Python 코드를 작성해줘",
    system="당신은 알고리즘 전문가입니다. 시간·공간 복잡도를 반드시 분석하세요.",
)
print(f"추론 길이: {result.think_len}개 단어")
print(f"추론 과정 일부: {result.thinking[:300]}...")
print(f"\n최종 답변:\n{result.answer}")

from openai import OpenAI

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

def ask_r1(question: str) -> tuple[str, str]:
    import re
    resp = client.chat.completions.create(
        model="deepseek-r1:7b",
        messages=[{"role": "user", "content": question}],
        temperature=0.6,   # R1 추천 온도
    )
    content = resp.choices[0].message.content
    think_m = re.search(r"<think>(.*?)</think>", content, re.DOTALL)
    thinking = think_m.group(1).strip() if think_m else ""
    answer   = re.sub(r"<think>.*?</think>", "", content, flags=re.DOTALL).strip()
    return thinking, answer

thinking, answer = ask_r1("하노이 탑 문제를 재귀와 반복 두 가지 방법으로 풀어줘")
print(f"[추론] {thinking[:200]}...")
print(f"[답변] {answer}")

import json
from openai import OpenAI

# V3 — Function Calling (코드 실행, 도구 호출)
client = OpenAI(api_key="YOUR_KEY", base_url="https://api.deepseek.com")

TOOLS = [
    {"type": "function", "function": {
        "name": "run_python",
        "description": "Python 코드를 실행하고 결과를 반환합니다",
        "parameters": {"type": "object",
            "properties": {
                "code": {"type": "string", "description": "실행할 Python 코드"},
                "timeout": {"type": "integer", "default": 10},
            }, "required": ["code"]},
    }},
    {"type": "function", "function": {
        "name": "search_arxiv",
        "description": "arXiv에서 논문을 검색합니다",
        "parameters": {"type": "object",
            "properties": {
                "query": {"type": "string"},
                "max_results": {"type": "integer", "default": 5},
            }, "required": ["query"]},
    }},
]

def run_python(code: str, timeout: int = 10) -> str:
    # 실제 구현 시 subprocess + 샌드박스 사용
    return f"실행 완료:\n{code}\n→ [결과]"

def search_arxiv(query: str, max_results: int = 5) -> str:
    return f"arXiv 검색 '{query}': {max_results}개 논문 발견..."

TOOL_FNS = {"run_python": run_python, "search_arxiv": search_arxiv}

def chat_with_tools(user_msg: str) -> str:
    messages = [{"role": "user", "content": user_msg}]
    for _ in range(5):
        resp = client.chat.completions.create(
            model="deepseek-chat",   # V3 — Function Calling
            messages=messages, tools=TOOLS, 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 = TOOL_FNS[tc.function.name](**args)
            messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})
    return "스텝 초과"

print(chat_with_tools("피보나치 수열 20번째 항을 Python으로 계산하고 실행 결과를 보여줘"))

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

# R1: 계획 수립 (CoT 추론)
r1_client = OpenAI(api_key="YOUR_KEY", base_url="https://api.deepseek.com")

# V3: 도구 실행 (Function Calling)
v3_client = OpenAI(api_key="YOUR_KEY", base_url="https://api.deepseek.com")

class Plan(BaseModel):
    steps:      list[str]   # 단계별 실행 계획
    tools_needed: list[str] # 필요한 도구 목록
    reasoning:  str         # 계획 수립 근거

def make_plan(user_goal: str) -> Plan:
    """R1으로 실행 계획 수립"""
    resp = r1_client.chat.completions.create(
        model="deepseek-reasoner",
        messages=[{
            "role": "user",
            "content": f"""다음 목표를 달성하기 위한 단계별 계획을 JSON으로 작성해줘.
목표: {user_goal}

반드시 JSON 형식으로:
{{"steps": [...], "tools_needed": [...], "reasoning": "..."}}"""
        }],
    )
    content = resp.choices[0].message.content
    # <think> 태그 제거 후 JSON 추출
    clean = re.sub(r"<think>.*?</think>", "", content, flags=re.DOTALL).strip()
    json_m = re.search(r"\{.*\}", clean, re.DOTALL)
    return Plan(**json.loads(json_m.group())) if json_m else Plan(
        steps=[user_goal], tools_needed=[], reasoning="단순 작업"
    )

def execute_step(step: str, context: str) -> str:
    """V3로 단계 실행 (도구 호출 가능)"""
    resp = v3_client.chat.completions.create(
        model="deepseek-chat",
        messages=[
            {"role": "system", "content": f"이전 컨텍스트: {context}"},
            {"role": "user",   "content": f"다음 단계를 실행해줘: {step}"},
        ],
    )
    return resp.choices[0].message.content

def run_planner_executor(goal: str) -> str:
    # 1. R1으로 계획 수립
    plan = make_plan(goal)
    print(f"계획 수립 완료: {len(plan.steps)}단계")
    print(f"근거: {plan.reasoning[:200]}")

    # 2. V3로 단계별 실행
    context, results = "", []
    for i, step in enumerate(plan.steps, 1):
        print(f"[{i}/{len(plan.steps)}] {step}")
        result = execute_step(step, context)
        results.append(result)
        context = f"단계 {i} 완료: {result[:200]}"

    return "\n".join(results)

output = run_planner_executor("Python으로 REST API 서버를 만들고 Docker로 배포하는 전체 과정을 안내해줘")

from openai import OpenAI
import sys

client = OpenAI(api_key="YOUR_KEY", base_url="https://api.deepseek.com")

def stream_r1(prompt: str):
    """추론 과정과 최종 답변을 실시간 스트리밍"""
    stream = client.chat.completions.create(
        model="deepseek-reasoner",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )

    in_think = False
    for chunk in stream:
        delta = chunk.choices[0].delta

        # reasoning_content: 추론 과정 (R1 전용 필드)
        if hasattr(delta, "reasoning_content") and delta.reasoning_content:
            if not in_think:
                print("\n💭 [추론 중...]", flush=True)
                in_think = True
            print(delta.reasoning_content, end="", flush=True)

        # content: 최종 답변
        if delta.content:
            if in_think:
                print("\n\n✅ [최종 답변]", flush=True)
                in_think = False
            print(delta.content, end="", flush=True)

    print()  # 줄바꿈

stream_r1("피타고라스 정리를 증명하고 Python으로 직각삼각형 판별기를 만들어줘")

import asyncio, re
from openai import AsyncOpenAI

client = AsyncOpenAI(base_url="http://localhost:8000/v1", api_key="none")

async def batch_reason(problems: list[str]) -> list[dict]:
    """여러 추론 문제를 동시에 처리"""
    async def solve(problem: str) -> dict:
        resp = await client.chat.completions.create(
            model="deepseek-ai/DeepSeek-R1-Distill-Qwen-7B",
            messages=[{"role": "user", "content": problem}],
            max_tokens=4096,
            temperature=0.6,
        )
        content = resp.choices[0].message.content
        think_m = re.search(r"<think>(.*?)</think>", content, re.DOTALL)
        return {
            "problem": problem,
            "thinking": think_m.group(1)[:200] if think_m else "",
            "answer":   re.sub(r"<think>.*?</think>", "", content, flags=re.DOTALL).strip(),
        }

    return await asyncio.gather(*[solve(p) for p in problems])

problems = [
    "피보나치 40번째 항을 메모이제이션으로 계산하는 Python 코드",
    "이진 탐색 트리의 삽입과 검색을 구현해줘",
    "동적 프로그래밍으로 최장 공통 부분 수열(LCS)을 구현해줘",
]
results = asyncio.run(batch_reason(problems))
for r in results:
    print(f"Q: {r['problem'][:50]}...")
    print(f"A: {r['answer'][:200]}\n")