0환경 설정
1
Ollama + DeepSeek-R1 —
ollama pull deepseek-r1:7b (7B Distill, ~5GB)2
패키지 —
pip install fastapi uvicorn ollama pydanticDeepSeek-R1의 특징: 응답 전에
<think>...</think> 블록으로 사고 과정을 출력합니다. 이 과정을 파싱해 투명한 추론을 UI에 표시할 수 있습니다.1Think 태그 파서 (핵심 유틸)
모든 DeepSeek-R1 예제에서 공통으로 사용하는 사고 과정 파서입니다.
think_parser.py — R1 응답 파싱 유틸리티
import re
from dataclasses import dataclass
from ollama import Client
ollama = Client()
@dataclass
class R1Response:
thinking: str # 블록 내용
answer: str # 최종 답변
full_response: str # 전체 원문
def parse_r1_response(raw: str) -> R1Response:
"""DeepSeek-R1 응답에서 사고 과정과 답변을 분리"""
think_pattern = re.compile(r'(.*?) ', re.DOTALL)
match = think_pattern.search(raw)
thinking = match.group(1).strip() if match else ""
answer = think_pattern.sub('', raw).strip()
return R1Response(thinking=thinking, answer=answer, full_response=raw)
def ask_r1(
question: str,
system: str = "당신은 논리적으로 사고하는 AI 어시스턴트입니다.",
model: str = "deepseek-r1:7b",
show_thinking: bool = False,
) -> R1Response:
"""DeepSeek-R1에 질문하고 파싱된 응답 반환"""
resp = ollama.chat(
model=model,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": question},
],
options={"temperature": 0.6, "top_p": 0.95},
)
result = parse_r1_response(resp["message"]["content"])
if show_thinking and result.thinking:
print("💭 사고 과정:")
print(result.thinking[:500] + ("..." if len(result.thinking) > 500 else ""))
print()
return result
# 스트리밍 버전
def ask_r1_stream(question: str, model: str = "deepseek-r1:7b"):
"""스트리밍으로 Think + 답변 실시간 출력"""
stream = ollama.chat(
model=model,
messages=[{"role": "user", "content": question}],
stream=True,
options={"temperature": 0.6},
)
full = ""
in_think = False
for chunk in stream:
token = chunk["message"]["content"]
full += token
if "" in token:
in_think = True
print("\n💭 [사고 중...]", end="", flush=True)
elif " " in token:
in_think = False
print("\n\n✅ [답변]", end="", flush=True)
else:
color = "\033[90m" if in_think else "\033[0m"
print(f"{color}{token}\033[0m", end="", flush=True)
print()
return parse_r1_response(full) python2수학 문제 풀이 시스템
DeepSeek-R1의 수학 특화 능력을 활용한 단계별 풀이 시스템입니다.
math_solver.py — 단계별 수학 풀이
from think_parser import ask_r1, R1Response
from pydantic import BaseModel
import json
MATH_SYSTEM = """당신은 수학 전문가입니다.
문제를 단계별로 풀고 최종 답을 명확히 표시하세요.
최종 답은 반드시 '**정답: [값]**' 형식으로 끝내세요."""
class MathSolution(BaseModel):
problem: str
thinking_steps: list[str]
solution_steps: list[str]
final_answer: str
confidence: float
def solve_math(problem: str) -> MathSolution:
result = ask_r1(problem, system=MATH_SYSTEM)
# 사고 과정에서 단계 추출
thinking_steps = [
line.strip() for line in result.thinking.split('\n')
if line.strip() and len(line.strip()) > 10
][:8]
# 답변에서 풀이 단계 추출
solution_steps = [
line.strip() for line in result.answer.split('\n')
if line.strip()
]
# 최종 답 추출
import re
answer_match = re.search(r'\*\*정답[:\s]+([^\*]+)\*\*', result.answer)
final_answer = answer_match.group(1).strip() if answer_match else result.answer.split('\n')[-1]
return MathSolution(
problem=problem,
thinking_steps=thinking_steps,
solution_steps=solution_steps,
final_answer=final_answer,
confidence=0.95 if len(thinking_steps) > 3 else 0.7,
)
# 다양한 수학 문제 테스트
problems = [
"피보나치 수열의 10번째 값은?",
"반지름이 5인 원의 넓이는? (π≈3.14)",
"2^10 + 3^5 - 100 = ?",
"1부터 100까지 짝수의 합은?",
]
for problem in problems:
print(f"📝 문제: {problem}")
sol = solve_math(problem)
print(f"✅ 정답: {sol.final_answer}")
print(f"🔍 신뢰도: {sol.confidence:.0%}")
print(f"📊 추론 단계: {len(sol.thinking_steps)}단계\n")python3코드 디버깅 에이전트
debug_agent.py — 자동 디버깅 & 수정
에러 메시지와 코드를 입력하면 R1이 원인을 추론하고 수정된 코드를 제안합니다.
from think_parser import ask_r1
from pydantic import BaseModel
import re
DEBUG_SYSTEM = """당신은 시니어 소프트웨어 엔지니어입니다.
코드와 에러를 분석해 다음 형식으로 응답하세요:
## 에러 원인
[명확한 원인 설명]
## 수정된 코드
```python
[완전히 수정된 코드]
```
## 방지 방법
[재발 방지를 위한 팁]"""
class DebugResult(BaseModel):
original_code: str
error_message: str
root_cause: str
fixed_code: str
prevention_tips: list[str]
thinking_summary: str
def debug_code(code: str, error: str, language: str = "python") -> DebugResult:
prompt = f"""다음 {language} 코드에서 에러가 발생했습니다.
에러 메시지:
```
{error}
```
문제 코드:
```{language}
{code}
```
원인을 분석하고 수정해주세요."""
result = ask_r1(prompt, system=DEBUG_SYSTEM)
# 섹션 파싱
def extract_section(text: str, header: str) -> str:
pattern = rf"## {header}\n(.*?)(?=## |\Z)"
match = re.search(pattern, text, re.DOTALL)
return match.group(1).strip() if match else ""
# 코드 블록 추출
code_match = re.search(r"```(?:python)?\n(.*?)```", result.answer, re.DOTALL)
fixed_code = code_match.group(1).strip() if code_match else ""
root_cause = extract_section(result.answer, "에러 원인")
prevention_raw = extract_section(result.answer, "방지 방법")
prevention_tips = [
line.lstrip("•-* ").strip()
for line in prevention_raw.split('\n')
if line.strip()
]
# 사고 과정 요약
thinking_lines = result.thinking.split('\n')
thinking_summary = ' '.join(thinking_lines[:3]) if thinking_lines else ""
return DebugResult(
original_code=code,
error_message=error,
root_cause=root_cause,
fixed_code=fixed_code,
prevention_tips=prevention_tips[:3],
thinking_summary=thinking_summary[:200],
)
# 테스트
buggy_code = """
def calculate_average(numbers):
total = sum(numbers)
return total / len(numbers)
result = calculate_average([]) # 빈 리스트
print(result)
"""
error_msg = "ZeroDivisionError: division by zero"
debug_result = debug_code(buggy_code, error_msg)
print("🔍 원인:", debug_result.root_cause)
print("\n✅ 수정된 코드:")
print(debug_result.fixed_code)
print("\n💡 방지 방법:")
for tip in debug_result.prevention_tips:
print(f" • {tip}")python4알고리즘 설계 & 복잡도 분석
algo_analyzer.py — 알고리즘 설계 & Big-O 분석
from think_parser import ask_r1
from pydantic import BaseModel
import json
ALGO_SYSTEM = """당신은 알고리즘 전문가입니다.
문제를 분석하고 최적의 알고리즘을 설계한 후 다음 JSON으로 반환하세요:
{
"algorithm_name": "알고리즘 이름",
"approach": "접근 방식 설명",
"time_complexity": "O(...)",
"space_complexity": "O(...)",
"implementation": "Python 구현 코드",
"test_cases": [{"input": ..., "expected": ...}],
"optimizations": ["추가 최적화 가능성"]
}"""
class AlgoDesign(BaseModel):
algorithm_name: str
approach: str
time_complexity: str
space_complexity: str
implementation: str
test_cases: list[dict]
optimizations: list[str]
thinking_insight: str # R1의 핵심 통찰
def design_algorithm(problem: str) -> AlgoDesign:
result = ask_r1(problem, system=ALGO_SYSTEM)
# JSON 파싱 시도
try:
import re
json_match = re.search(r'\{[\s\S]*\}', result.answer)
if json_match:
data = json.loads(json_match.group())
else:
raise ValueError("JSON not found")
except:
# 파싱 실패 시 기본값
data = {
"algorithm_name": "미파싱",
"approach": result.answer[:200],
"time_complexity": "분석 중",
"space_complexity": "분석 중",
"implementation": "",
"test_cases": [],
"optimizations": [],
}
# 사고 과정에서 핵심 통찰 추출
thinking_lines = [l for l in result.thinking.split('\n') if l.strip() and len(l) > 20]
insight = thinking_lines[0] if thinking_lines else ""
return AlgoDesign(**data, thinking_insight=insight)
# 다양한 알고리즘 문제
problems = [
"정렬된 배열에서 특정 값을 찾는 가장 효율적인 알고리즘을 설계하세요.",
"주어진 문자열에서 가장 긴 팰린드롬 부분 문자열을 찾는 알고리즘을 설계하세요.",
"N개의 동전으로 특정 금액을 만드는 최소 동전 개수를 구하는 알고리즘을 설계하세요.",
]
for problem in problems[:1]: # 데모용으로 1개만
print(f"📝 문제: {problem}\n")
design = design_algorithm(problem)
print(f"🔧 알고리즘: {design.algorithm_name}")
print(f"⏱ 시간복잡도: {design.time_complexity}")
print(f"💾 공간복잡도: {design.space_complexity}")
if design.implementation:
print(f"\n📝 구현:\n{design.implementation}")
print(f"\n💭 R1 통찰: {design.thinking_insight}")python5추론 API 서버
reasoning_api.py — Think 과정 포함 API
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from ollama import Client, AsyncClient
from think_parser import parse_r1_response
import json, asyncio
app = FastAPI(title="DeepSeek-R1 Reasoning API")
sync_ollama = Client()
async_ollama = AsyncClient()
class ReasoningRequest(BaseModel):
question: str
model: str = "deepseek-r1:7b"
show_thinking: bool = True
temperature: float = 0.6
class ReasoningResponse(BaseModel):
question: str
thinking: str
answer: str
thinking_length: int
@app.post("/reason", response_model=ReasoningResponse)
async def reason(req: ReasoningRequest):
resp = await async_ollama.chat(
model=req.model,
messages=[{"role": "user", "content": req.question}],
options={"temperature": req.temperature},
)
parsed = parse_r1_response(resp["message"]["content"])
return ReasoningResponse(
question=req.question,
thinking=parsed.thinking if req.show_thinking else "",
answer=parsed.answer,
thinking_length=len(parsed.thinking),
)
@app.post("/reason/stream")
async def reason_stream(req: ReasoningRequest):
"""사고 과정과 답변을 실시간으로 스트리밍"""
async def generate():
stream = await async_ollama.chat(
model=req.model,
messages=[{"role": "user", "content": req.question}],
stream=True,
options={"temperature": req.temperature},
)
buffer = ""
phase = "thinking" # thinking → answer
async for chunk in stream:
token = chunk["message"]["content"]
buffer += token
if "" in buffer and phase == "waiting":
phase = "thinking"
if " " in buffer and phase == "thinking":
phase = "answer"
yield f"data: {json.dumps({'type': 'phase_change', 'phase': 'answer'})}\n\n"
event_type = "thinking" if phase == "thinking" else "answer"
if token:
yield f"data: {json.dumps({'type': event_type, 'token': token})}\n\n"
yield f"data: {json.dumps({'type': 'done'})}\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
# 전문 분야별 추론 엔드포인트
@app.post("/math")
async def math_reason(problem: str):
resp = await async_ollama.chat(
model="deepseek-r1:7b",
messages=[{
"role": "system",
"content": "수학 문제를 단계별로 풀고 최종 답을 명확히 제시하세요."
}, {
"role": "user",
"content": problem
}],
)
parsed = parse_r1_response(resp["message"]["content"])
return {"problem": problem, "steps": parsed.thinking, "answer": parsed.answer}
@app.post("/logic")
async def logic_reason(statement: str):
resp = await async_ollama.chat(
model="deepseek-r1:7b",
messages=[{
"role": "system",
"content": "논리적 추론으로 주어진 진술을 분석하고 결론을 도출하세요."
}, {
"role": "user",
"content": statement
}],
)
parsed = parse_r1_response(resp["message"]["content"])
return {"statement": statement, "reasoning": parsed.thinking, "conclusion": parsed.answer}
# uvicorn reasoning_api:app --reloadpython