LLM 프롬프트 엔지니어링 고급 기법: Chain-of-Thought·ReAct·Tree of Thoughts 실전 적용

from openai import OpenAI

client = OpenAI()

def few_shot_sentiment_analysis(text: str) -> str:
    """Few-shot 프롬프팅을 활용한 감성 분석"""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": "주어진 텍스트의 감성을 분석하여 긍정/부정/중립으로 분류하세요."
            },
            # Few-shot 예시 1
            {"role": "user", "content": "이 제품 정말 최고입니다! 배송도 빠르고 품질도 좋아요."},
            {"role": "assistant", "content": "감성: 긍정\n근거: '최고', '빠르고', '좋아요' 등 긍정적 표현이 다수 포함"},
            # Few-shot 예시 2
            {"role": "user", "content": "배송이 너무 늦었고 제품에 흠집이 있었습니다."},
            {"role": "assistant", "content": "감성: 부정\n근거: '너무 늦었고', '흠집이 있었습니다' 등 부정적 경험 서술"},
            # Few-shot 예시 3
            {"role": "user", "content": "보통 수준의 제품입니다. 가격 대비 적당합니다."},
            {"role": "assistant", "content": "감성: 중립\n근거: '보통 수준', '적당합니다' 등 균형적 평가"},
            # 실제 분석 대상
            {"role": "user", "content": text}
        ],
        temperature=0.0
    )
    return response.choices[0].message.content

def chain_of_thought_math(problem: str) -> str:
    """Chain-of-Thought를 활용한 수학 문제 해결"""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "당신은 수학 문제를 단계별로 풀어주는 튜터입니다. "
                    "각 단계를 명확히 설명하고, 최종 답을 제시하세요."
                )
            },
            # CoT 예시: 추론 과정 포함
            {
                "role": "user",
                "content": "카페에 손님이 23명 있었다. 오전에 12명이 더 왔고, 오후에 8명이 나갔다. 현재 카페에 몇 명이 있는가?"
            },
            {
                "role": "assistant",
                "content": (
                    "단계별로 풀어보겠습니다.\n\n"
                    "1단계: 초기 인원 파악\n"
                    "- 처음 카페에 있던 손님: 23명\n\n"
                    "2단계: 오전 변동 계산\n"
                    "- 12명이 추가로 입장: 23 + 12 = 35명\n\n"
                    "3단계: 오후 변동 계산\n"
                    "- 8명이 퇴장: 35 - 8 = 27명\n\n"
                    "최종 답: 현재 카페에는 27명이 있습니다."
                )
            },
            {"role": "user", "content": problem}
        ],
        temperature=0.0
    )
    return response.choices[0].message.content

import json
import requests
from openai import OpenAI

client = OpenAI()

# 도구 정의
def search_wikipedia(query: str) -> str:
    """위키피디아 검색 도구"""
    url = "https://ko.wikipedia.org/w/api.php"
    params = {
        "action": "query",
        "list": "search",
        "srsearch": query,
        "format": "json",
        "srlimit": 3
    }
    resp = requests.get(url, params=params)
    results = resp.json().get("query", {}).get("search", [])
    if not results:
        return "검색 결과가 없습니다."
    return "\n".join([r["title"] + ": " + r["snippet"] for r in results])


def calculator(expression: str) -> str:
    """안전한 수식 계산기"""
    allowed_chars = set("0123456789+-*/.(). ")
    if all(c in allowed_chars for c in expression):
        try:
            result = eval(expression)
            return str(result)
        except Exception as e:
            return f"계산 오류: {e}"
    return "허용되지 않는 문자가 포함되어 있습니다."


TOOLS = {
    "search_wikipedia": search_wikipedia,
    "calculator": calculator,
}


def react_agent(question: str, max_steps: int = 5) -> str:
    """ReAct 패턴을 구현한 에이전트"""
    system_prompt = """당신은 ReAct 에이전트입니다. 다음 형식으로 응답하세요:

Thought: 현재 상황을 분석하고, 다음에 무엇을 해야 할지 추론합니다.
Action: 사용할 도구와 입력을 JSON 형식으로 제공합니다.
Observation: 도구 실행 결과를 확인합니다.
... (필요한 만큼 반복)
Final Answer: 최종 답변을 제시합니다.

사용 가능한 도구:
- search_wikipedia: 위키피디아 검색 (입력: 검색어)
- calculator: 수식 계산 (입력: 수학 표현식)"""

    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": question}
    ]

    for step in range(max_steps):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            temperature=0.0
        )
        assistant_msg = response.choices[0].message.content

        # Final Answer가 포함되면 종료
        if "Final Answer:" in assistant_msg:
            return assistant_msg.split("Final Answer:")[-1].strip()

        # Action 파싱 및 실행
        if "Action:" in assistant_msg:
            action_line = assistant_msg.split("Action:")[-1].strip()
            try:
                action_data = json.loads(action_line.split("\n")[0])
                tool_name = action_data.get("tool", "")
                tool_input = action_data.get("input", "")

                if tool_name in TOOLS:
                    observation = TOOLS[tool_name](tool_input)
                else:
                    observation = f"알 수 없는 도구: {tool_name}"
            except json.JSONDecodeError:
                observation = "Action 파싱 실패"

            messages.append({"role": "assistant", "content": assistant_msg})
            messages.append({
                "role": "user",
                "content": f"Observation: {observation}"
            })

    return "최대 단계 수에 도달하여 답변을 생성하지 못했습니다."

from collections import Counter
from openai import OpenAI

client = OpenAI()


def self_consistency_solve(
    question: str,
    num_samples: int = 5,
    temperature: float = 0.7
) -> dict:
    """Self-Consistency 샘플링으로 정확도를 높이는 추론"""
    system_prompt = (
        "수학 문제를 단계별로 풀어주세요. "
        "마지막 줄에 '정답: [숫자]' 형식으로 답을 표기하세요."
    )
    answers = []
    reasoning_paths = []

    for i in range(num_samples):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": question}
            ],
            temperature=temperature  # 다양한 추론 경로 생성
        )
        content = response.choices[0].message.content
        reasoning_paths.append(content)

        # 정답 추출
        for line in content.split("\n"):
            if line.strip().startswith("정답:"):
                answer = line.split("정답:")[-1].strip()
                answers.append(answer)
                break

    # 다수결 투표
    if not answers:
        return {"final_answer": "답변 추출 실패", "confidence": 0.0}

    counter = Counter(answers)
    most_common = counter.most_common(1)[0]
    confidence = most_common[1] / len(answers)

    return {
        "final_answer": most_common[0],
        "confidence": confidence,
        "vote_distribution": dict(counter),
        "num_samples": num_samples,
        "reasoning_paths": reasoning_paths
    }


# 사용 예시
result = self_consistency_solve(
    "학교에 학생이 450명 있다. 남학생이 전체의 60%이고, "
    "남학생 중 25%가 안경을 착용한다. 안경을 착용한 남학생은 몇 명인가?",
    num_samples=7
)
print(f"최종 답: {result['final_answer']}")
print(f"신뢰도: {result['confidence']:.1%}")
print(f"투표 분포: {result['vote_distribution']}")

from openai import OpenAI
from typing import List

client = OpenAI()


def generate_thoughts(problem: str, current_state: str, n: int = 3) -> List[str]:
    """현재 상태에서 가능한 다음 사고를 생성"""
    prompt = f"""문제: {problem}
현재까지의 추론: {current_state}

이 상태에서 가능한 다음 추론 단계를 {n}개 제안하세요.
각 단계는 서로 다른 접근법이어야 합니다.
형식:
1. [첫 번째 접근]
2. [두 번째 접근]
3. [세 번째 접근]"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.8
    )
    content = response.choices[0].message.content
    thoughts = []
    for line in content.split("\n"):
        line = line.strip()
        if line and line[0].isdigit() and "." in line:
            thoughts.append(line.split(".", 1)[1].strip())
    return thoughts[:n]


def evaluate_thought(problem: str, thought_path: str) -> float:
    """사고 경로의 유망성을 0~1 사이로 평가"""
    prompt = f"""문제: {problem}
추론 경로: {thought_path}

이 추론 경로가 정답에 도달할 가능성을 평가하세요.
0.0 (전혀 유망하지 않음) ~ 1.0 (매우 유망함) 사이의 점수를 제시하세요.
점수만 숫자로 응답하세요."""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1
    )
    try:
        score = float(response.choices[0].message.content.strip())
        return min(max(score, 0.0), 1.0)
    except ValueError:
        return 0.5


def tree_of_thoughts_bfs(
    problem: str,
    max_depth: int = 3,
    beam_width: int = 2
) -> str:
    """BFS 기반 Tree of Thoughts 구현"""
    # 초기 상태
    current_states = [("", 1.0)]  # (추론 경로, 점수)

    for depth in range(max_depth):
        all_candidates = []

        for state, _ in current_states:
            # 각 상태에서 후보 사고 생성
            thoughts = generate_thoughts(problem, state)

            for thought in thoughts:
                new_path = f"{state}\n단계 {depth+1}: {thought}" if state else f"단계 1: {thought}"
                score = evaluate_thought(problem, new_path)
                all_candidates.append((new_path, score))

        # beam_width만큼 상위 후보 선택
        all_candidates.sort(key=lambda x: x[1], reverse=True)
        current_states = all_candidates[:beam_width]

        print(f"깊이 {depth+1}: 최고 점수 = {current_states[0][1]:.2f}")

    # 최종 답변 생성
    best_path = current_states[0][0]
    final_prompt = f"""문제: {problem}
추론 경로: {best_path}

위 추론을 바탕으로 최종 답변을 작성하세요."""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": final_prompt}],
        temperature=0.0
    )
    return response.choices[0].message.content

from openai import OpenAI
from pydantic import BaseModel, Field
from typing import List, Optional

client = OpenAI()


class CodeReview(BaseModel):
    """코드 리뷰 결과 스키마"""
    file_path: str = Field(description="리뷰 대상 파일 경로")
    severity: str = Field(description="심각도: critical, warning, info")
    category: str = Field(description="카테고리: security, performance, style, logic")
    line_number: Optional[int] = Field(description="해당 라인 번호")
    message: str = Field(description="리뷰 코멘트")
    suggestion: str = Field(description="개선 제안 코드")


class CodeReviewResult(BaseModel):
    """전체 코드 리뷰 결과"""
    reviews: List[CodeReview]
    summary: str = Field(description="리뷰 요약")
    overall_score: int = Field(description="전체 점수 (1-10)")


def structured_code_review(code: str, language: str = "python") -> CodeReviewResult:
    """구조화된 출력을 활용한 코드 리뷰"""
    response = client.beta.chat.completions.parse(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    f"당신은 시니어 {language} 개발자입니다. "
                    "주어진 코드를 리뷰하고 구조화된 형식으로 피드백을 제공하세요."
                )
            },
            {"role": "user", "content": f"다음 코드를 리뷰해주세요:\n\n{code}"}
        ],
        response_format=CodeReviewResult
    )
    return response.choices[0].message.parsed

import json
from openai import OpenAI

client = OpenAI()

# 도구 스키마 정의
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_stock_price",
            "description": "주식의 현재 가격을 조회합니다",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {
                        "type": "string",
                        "description": "주식 심볼 (예: AAPL, GOOGL)"
                    },
                    "currency": {
                        "type": "string",
                        "enum": ["USD", "KRW", "JPY"],
                        "description": "표시 통화"
                    }
                },
                "required": ["symbol"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "create_alert",
            "description": "주가 알림을 설정합니다",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {"type": "string"},
                    "target_price": {"type": "number"},
                    "direction": {
                        "type": "string",
                        "enum": ["above", "below"]
                    }
                },
                "required": ["symbol", "target_price", "direction"]
            }
        }
    }
]


def function_calling_agent(user_message: str) -> str:
    """Function Calling을 활용한 에이전트"""
    messages = [
        {
            "role": "system",
            "content": "당신은 주식 정보를 제공하는 금융 어시스턴트입니다."
        },
        {"role": "user", "content": user_message}
    ]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools,
        tool_choice="auto"
    )

    msg = response.choices[0].message

    # Function Call 처리
    if msg.tool_calls:
        for tool_call in msg.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            # 실제 함수 실행 (여기서는 모의 데이터)
            if func_name == "get_stock_price":
                result = json.dumps({
                    "symbol": func_args["symbol"],
                    "price": 185.50,
                    "change": "+2.3%"
                })
            elif func_name == "create_alert":
                result = json.dumps({
                    "status": "success",
                    "alert_id": "alert_12345"
                })
            else:
                result = json.dumps({"error": "Unknown function"})

            messages.append(msg)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

        # 최종 응답 생성
        final_response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages
        )
        return final_response.choices[0].message.content

    return msg.content

from dataclasses import dataclass, field
from typing import Dict, Any
from datetime import datetime


@dataclass
class PromptTemplate:
    """프롬프트 템플릿 관리 클래스"""
    name: str
    version: str
    template: str
    metadata: Dict[str, Any] = field(default_factory=dict)
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())

    def render(self, **kwargs) -> str:
        """변수를 치환하여 최종 프롬프트 생성"""
        result = self.template
        for key, value in kwargs.items():
            placeholder = f"__{key.upper()}__"
            result = result.replace(placeholder, str(value))
        return result


# 프롬프트 레지스트리
class PromptRegistry:
    def __init__(self):
        self._templates: Dict[str, PromptTemplate] = {}

    def register(self, template: PromptTemplate):
        key = f"{template.name}:{template.version}"
        self._templates[key] = template

    def get(self, name: str, version: str = "latest") -> PromptTemplate:
        if version == "latest":
            candidates = [
                v for k, v in self._templates.items()
                if k.startswith(f"{name}:")
            ]
            return max(candidates, key=lambda t: t.version)
        return self._templates[f"{name}:{version}"]


# 사용 예시
registry = PromptRegistry()
registry.register(PromptTemplate(
    name="code_review",
    version="2.1",
    template=(
        "당신은 시니어 __LANGUAGE__ 개발자입니다.\n"
        "다음 코드를 리뷰하세요.\n\n"
        "리뷰 기준:\n"
        "- 보안 취약점\n"
        "- 성능 이슈\n"
        "- 코드 스타일\n\n"
        "코드:\n__CODE__"
    ),
    metadata={"model": "gpt-4o", "temperature": 0.0}
))