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

Perception (receive input)
    → Reasoning (LLM thinks)
    → Action (use a tool)
    → Observation (check result)
    → back to Reasoning

Thought: [reasoning about what to do]
Action: tool_name(params)
Observation: [result of tool call]
Thought: [reasoning about the result]
... (repeat)
Final Answer: [conclusion]

# System prompt that enforces ReAct format
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.
"""

# Example trace:
# Question: "What is South Korea's GDP in 2023 divided by its population?"
# Thought: I need to find South Korea's GDP and population
# Action: search("South Korea GDP 2023")
# Observation: "South Korea GDP 2023: $1.71 trillion USD"
# Thought: Now I need population
# Action: search("South Korea population 2024")
# Observation: "51.7 million"
# Thought: Now I can calculate
# Action: calculator("1710000000000 / 51700000")
# Observation: "33075"
# Final Answer: ~$33,075 GDP per capita

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

            # Done when we see Final Answer
            if "Final Answer:" in content:
                return content.split("Final Answer:")[-1].strip()

            # Parse the Action
            action_match = re.search(r"Action: (\w+)\((.*?)\)", content)
            if not action_match:
                return content  # Format failure — return as-is

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

            # Execute the tool
            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)}"

            # Append Observation to message history
            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."
)
# Forces the LLM to reason through the problem sequentially

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      # Expensive, powerful model
        self.executor = executor_llm    # Cheaper, faster model
        self.tools = tools

    def run(self, goal: str) -> str:
        # Phase 1: Planning — use your best 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")

        # Phase 2: Execution — can use a cheaper model
        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})

        # Phase 3: Summarize
        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:
        # Generate initial answer
        answer = self.llm.invoke(f"Answer this question thoroughly: {question}")

        for i in range(num_reflections):
            # Self-critique phase
            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 the explanation clear?
            4. Are there logical errors?

            Be specific and constructive.
            """)

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

            # Refinement phase
            answer = self.llm.invoke(f"""
            Original question: {question}
            Previous answer: {answer}
            Critique: {critique}

            Provide an improved answer that addresses all critique points.
            """)

        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 exploration
        queue = [root]
        best_leaf = None
        best_score = 0

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

            for node in queue:
                # Generate multiple next thoughts from this node
                children = await self._generate_thoughts(node["thought"])

                for child_thought in children:
                    # Score each reasoning path
                    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

            # Keep only top-k paths (beam search)
            queue = sorted(next_queue, key=lambda x: x["score"], reverse=True)[:self.branching_factor]

        # Generate final answer from best path
        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} distinct ways to continue this reasoning.
        Return as a 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 = 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):
    """Remove oldest Observations first, keep system message"""
    while count_tokens(messages) > max_tokens:
        for i, msg in enumerate(messages[1:], 1):
            if "Observation:" in msg.get("content", ""):
                messages.pop(i)
                messages.pop(i - 1)  # Remove the paired Action too
                break
    return messages

# Bad — silent failure
result = tool.run(args)

# Good — tell the LLM what went wrong so it can adapt
try:
    result = tool.run(args)
except Exception as e:
    result = f"Tool error: {str(e)}. Try a different approach."

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."