LLM Agent 설계 패턴 완전 가이드: ReAct부터 Multi-Agent까지

Perception (입력 받기)
    → Reasoning (LLM이 생각)
    → Action (툴 사용)
    → Observation (결과 확인)
    → back to Reasoning (다시 생각)

Thought: 무엇을 해야 하는지 추론
Action: tool_name(params)
Observation: [툴 실행 결과]
Thought: 결과를 보고 다음 행동 결정
... (반복)
Final Answer: 최종 답변

system_prompt = """
You have access to these tools: [search, calculator, code_executor]

Always follow this exact format:
Thought: I need to...
Action: tool_name(params)
Observation: [tool result]
... (repeat as needed)
Final Answer: ...

Never skip the Thought step. Never fabricate Observations.
"""

# 실제 트레이스 예시:
# Question: "2023년 한국 GDP를 인구로 나누면 얼마인가?"
# Thought: 한국의 GDP와 인구를 찾아야 한다
# Action: search("South Korea GDP 2023")
# Observation: "South Korea GDP 2023: $1.71 trillion USD"
# Thought: 이제 인구가 필요하다
# Action: search("South Korea population 2024")
# Observation: "51.7 million"
# Thought: 이제 계산
# Action: calculator("1710000000000 / 51700000")
# Observation: "33075"
# Final Answer: 약 $33,075 (1인당 GDP)

import re
from typing import Callable

class ReActAgent:
    def __init__(self, llm, tools: dict[str, Callable], max_iterations=10):
        self.llm = llm
        self.tools = tools
        self.max_iterations = max_iterations

    def run(self, question: str) -> str:
        messages = [
            {"role": "system", "content": self._build_system_prompt()},
            {"role": "user", "content": question}
        ]

        for i in range(self.max_iterations):
            response = self.llm.invoke(messages)
            content = response.content

            # Final Answer가 나오면 종료
            if "Final Answer:" in content:
                return content.split("Final Answer:")[-1].strip()

            # Action 파싱
            action_match = re.search(r"Action: (\w+)\((.*?)\)", content)
            if not action_match:
                return content  # 포맷 실패 시 그대로 반환

            tool_name = action_match.group(1)
            tool_args = action_match.group(2)

            # 툴 실행
            if tool_name not in self.tools:
                observation = f"Error: tool '{tool_name}' not found"
            else:
                try:
                    observation = self.tools[tool_name](tool_args)
                except Exception as e:
                    observation = f"Error executing tool: {str(e)}"

            # Observation을 메시지에 추가
            messages.append({"role": "assistant", "content": content})
            messages.append({"role": "user", "content": f"Observation: {observation}"})

        return "Max iterations reached without final answer"

    def _build_system_prompt(self):
        tool_names = list(self.tools.keys())
        return f"""You have access to these tools: {tool_names}
Always use this format:
Thought: [your reasoning]
Action: tool_name(params)
Observation: [will be filled in]
Final Answer: [when done]"""

response = llm.invoke(
    "Q: A bat and a ball cost $1.10 total. The bat costs $1.00 more than the ball. "
    "How much does the ball cost?\n\nLet's think step by step."
)
# LLM이 단계별 추론을 하게 됨

few_shot_examples = """
Q: If 5 machines take 5 minutes to make 5 widgets, how long does it take 100 machines to make 100 widgets?
A:
Step 1: One machine makes 1 widget in 5 minutes.
Step 2: 100 machines each making 1 widget simultaneously = 5 minutes.
Answer: 5 minutes.

Q: [new question here]
A:
"""

class PlanAndExecuteAgent:
    def __init__(self, planner_llm, executor_llm, tools):
        self.planner = planner_llm
        self.executor = executor_llm
        self.tools = tools

    def run(self, goal: str) -> str:
        # 1단계: 계획 수립 (강력한 LLM 사용)
        plan_prompt = f"""
        Goal: {goal}

        Create a step-by-step plan to achieve this goal.
        Return a numbered list of concrete, executable steps.
        Each step should be specific enough to execute independently.
        """
        plan_response = self.planner.invoke(plan_prompt)
        steps = self._parse_plan(plan_response)

        print(f"Plan created: {len(steps)} steps")

        # 2단계: 각 단계 실행 (저렴한 LLM 사용 가능)
        results = []
        for i, step in enumerate(steps):
            print(f"Executing step {i+1}: {step}")

            result = self.executor.invoke(
                f"Execute this step: {step}\n\n"
                f"Context from previous steps: {results}\n\n"
                f"Available tools: {list(self.tools.keys())}"
            )
            results.append({"step": step, "result": result})

        # 3단계: 최종 정리
        summary = self.planner.invoke(
            f"Goal: {goal}\n\nResults: {results}\n\nSummarize the final outcome."
        )
        return summary

class ReflectionAgent:
    def __init__(self, llm):
        self.llm = llm

    def run(self, question: str, num_reflections: int = 2) -> str:
        # 초기 답변 생성
        answer = self.llm.invoke(f"Answer this question: {question}")

        for i in range(num_reflections):
            # 자기 비판
            critique = self.llm.invoke(f"""
            Question: {question}
            Answer: {answer}

            Critically evaluate this answer:
            1. Is it factually correct?
            2. Is it complete? What's missing?
            3. Is it clearly explained?
            4. Are there any logical errors?

            Be specific and constructive.
            """)

            print(f"Critique {i+1}: {critique[:200]}...")

            # 비판을 반영해서 개선
            answer = self.llm.invoke(f"""
            Original question: {question}
            Previous answer: {answer}
            Critique: {critique}

            Now provide an improved answer that addresses the critique.
            """)

        return answer

from typing import List
import asyncio

class TreeOfThoughts:
    def __init__(self, llm, branching_factor=3, max_depth=4):
        self.llm = llm
        self.branching_factor = branching_factor
        self.max_depth = max_depth

    async def solve(self, problem: str) -> str:
        root = {"thought": problem, "score": 1.0, "children": []}

        # BFS로 탐색
        queue = [root]
        best_leaf = None
        best_score = 0

        for depth in range(self.max_depth):
            next_queue = []

            for node in queue:
                # 각 노드에서 여러 다음 생각 생성
                children = await self._generate_thoughts(node["thought"])

                for child_thought in children:
                    # 각 경로 평가
                    score = await self._evaluate_thought(child_thought, problem)
                    child_node = {
                        "thought": child_thought,
                        "score": score,
                        "children": []
                    }
                    node["children"].append(child_node)
                    next_queue.append(child_node)

                    if score > best_score:
                        best_score = score
                        best_leaf = child_node

            # 상위 k개 경로만 계속 탐색 (빔 서치)
            queue = sorted(next_queue, key=lambda x: x["score"], reverse=True)[:self.branching_factor]

        # 최고 경로로 최종 답변 생성
        return await self._generate_final_answer(best_leaf["thought"], problem)

    async def _generate_thoughts(self, current_thought: str) -> List[str]:
        response = await self.llm.ainvoke(f"""
        Current reasoning: {current_thought}
        Generate {self.branching_factor} different ways to continue this reasoning.
        Each should be a distinct approach.
        Return as numbered list.
        """)
        return self._parse_numbered_list(response)

    async def _evaluate_thought(self, thought: str, problem: str) -> float:
        response = await self.llm.ainvoke(f"""
        Problem: {problem}
        Reasoning path: {thought}
        Rate this reasoning path from 0.0 to 1.0.
        Return only the number.
        """)
        try:
            return float(response.strip())
        except:
            return 0.5

# 반드시 max_iterations와 loop detection 추가
MAX_ITERATIONS = 15
seen_actions = set()

for i in range(MAX_ITERATIONS):
    action = agent.get_next_action()
    action_key = f"{action.tool}:{action.args}"

    if action_key in seen_actions:
        return "Error: Agent stuck in loop. Breaking out."
    seen_actions.add(action_key)

def trim_messages_to_fit(messages, max_tokens=100000):
    """오래된 Observation부터 제거"""
    while count_tokens(messages) > max_tokens:
        # 첫 번째 Observation 쌍 제거 (system 메시지는 유지)
        for i, msg in enumerate(messages[1:], 1):
            if "Observation:" in msg.get("content", ""):
                messages.pop(i)
                messages.pop(i-1)  # 해당 Action도 제거
                break
    return messages

# 나쁜 예
result = tool.run(args)

# 좋은 예: 에러를 LLM에게 알려줘서 대응하게 함
try:
    result = tool.run(args)
except Exception as e:
    result = f"Tool error: {str(e)}. Try a different approach."
    # LLM이 이 에러를 보고 다른 방법을 시도함

import asyncio

async def run_tool_with_timeout(tool, args, timeout=30):
    try:
        return await asyncio.wait_for(tool.arun(args), timeout=timeout)
    except asyncio.TimeoutError:
        return f"Tool timed out after {timeout}s. Try a different approach."