Split View: LLM 에이전트 & Agentic AI 완전 정복: ReAct, 멀티에이전트, MCP까지

Thought:     Analyze the current situation and decide on the next action
Action:      Call a tool in the form tool_name(arguments)
Observation: Receive the tool execution result
...repeat...
Final Answer: Derive the final answer

from langchain.agents import create_react_agent
from langchain_anthropic import ChatAnthropic
from langchain.tools import DuckDuckGoSearchRun, PythonREPLTool
from langchain import hub

llm = ChatAnthropic(model="claude-opus-4-5", temperature=0)
tools = [DuckDuckGoSearchRun(), PythonREPLTool()]

# Load the ReAct prompt template
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)

from langchain.agents import AgentExecutor
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = agent_executor.invoke({
    "input": "Search for the top 3 AI agent frameworks in 2026 and create a comparison table"
})

cot_prompt = """
Solve the problem step by step:

Problem: {problem}

Solution process:
1. First, organize the given information.
2. Perform the necessary calculations or reasoning.
3. Verify intermediate results.
4. Derive the final answer.
"""

from langchain_experimental.tot.base import ToTChain
from langchain_experimental.tot.thought_generation import ProposePromptStrategy

tot_chain = ToTChain.from_llm(
    llm=llm,
    checker=checker,
    k=3,    # Number of branches to generate at each level
    c=4,    # Evaluation depth
    verbose=True
)

from mem0 import Memory

# Initialize mem0 (using Qdrant as vector DB)
config = {
    "vector_store": {
        "provider": "qdrant",
        "config": {
            "collection_name": "agent_memory",
            "host": "localhost",
            "port": 6333,
        }
    },
    "llm": {
        "provider": "anthropic",
        "config": {
            "model": "claude-opus-4-5",
            "temperature": 0,
        }
    }
}

memory = Memory.from_config(config)
user_id = "user_123"

# Store memory
memory.add(
    messages=[
        {"role": "user", "content": "I mainly use Python and prefer FastAPI"},
        {"role": "assistant", "content": "Got it! I'll remember your Python/FastAPI preference."}
    ],
    user_id=user_id
)

# Search and use memory
relevant_memories = memory.search(
    query="What programming language does the user prefer?",
    user_id=user_id
)

context = "\n".join([m["memory"] for m in relevant_memories])
print(f"Relevant memories: {context}")

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from datetime import datetime

class EpisodicMemory:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings()
        self.store = Chroma(
            collection_name="episodes",
            embedding_function=self.embeddings,
            persist_directory="./episodic_memory"
        )

    def store_episode(self, content: str, metadata: dict = None):
        """Store a conversation or task episode in memory"""
        metadata = metadata or {}
        metadata["timestamp"] = datetime.now().isoformat()
        self.store.add_texts([content], metadatas=[metadata])

    def recall(self, query: str, k: int = 3):
        """Retrieve relevant episodes"""
        docs = self.store.similarity_search(query, k=k)
        return [doc.page_content for doc in docs]

# Usage example
memory = EpisodicMemory()
memory.store_episode(
    "User asked about FastAPI project structure. Successfully answered.",
    {"task_type": "coding", "success": True}
)

from langchain.tools import BaseTool
from pydantic import BaseModel, Field
import httpx
from typing import Optional

class WebSearchInput(BaseModel):
    query: str = Field(description="The query to search for")
    max_results: int = Field(default=5, description="Number of results to return")

class TavilySearchTool(BaseTool):
    name: str = "web_search"
    description: str = "Searches the web for up-to-date information. Use when real-time information is needed."
    args_schema: type[BaseModel] = WebSearchInput
    api_key: str = ""

    def _run(self, query: str, max_results: int = 5) -> str:
        url = "https://api.tavily.com/search"
        payload = {
            "api_key": self.api_key,
            "query": query,
            "max_results": max_results,
            "include_answer": True,
        }
        response = httpx.post(url, json=payload)
        data = response.json()

        results = []
        if data.get("answer"):
            results.append(f"Summary: {data['answer']}\n")
        for r in data.get("results", []):
            results.append(f"- {r['title']}: {r['content'][:200]}...")
        return "\n".join(results)

    async def _arun(self, query: str, max_results: int = 5) -> str:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                "https://api.tavily.com/search",
                json={"api_key": self.api_key, "query": query, "max_results": max_results}
            )
        return self._process_response(response.json())

# MCP server implementation (Python SDK)
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

app = Server("my-tool-server")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="get_weather",
            description="Retrieves the current weather for a specific city",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name"},
                    "units": {"type": "string", "enum": ["celsius", "fahrenheit"]}
                },
                "required": ["city"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "get_weather":
        city = arguments["city"]
        weather_data = await fetch_weather(city)
        return [types.TextContent(type="text", text=str(weather_data))]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from typing import TypedDict, Annotated, Sequence
import operator

# 1. Define state
class AgentState(TypedDict):
    messages: Annotated[Sequence, operator.add]
    tool_calls: list
    iteration_count: int

# 2. Set up LLM and tools
llm = ChatAnthropic(model="claude-opus-4-5")
tools = [WebSearchTool(), PythonREPLTool()]
llm_with_tools = llm.bind_tools(tools)

# 3. Define nodes
def call_model(state: AgentState) -> AgentState:
    """LLM call node"""
    response = llm_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "iteration_count": state["iteration_count"] + 1
    }

def call_tools(state: AgentState) -> AgentState:
    """Tool execution node"""
    last_message = state["messages"][-1]
    tool_results = []

    for tool_call in last_message.tool_calls:
        tool = next(t for t in tools if t.name == tool_call["name"])
        result = tool.invoke(tool_call["args"])
        tool_results.append(
            ToolMessage(content=str(result), tool_call_id=tool_call["id"])
        )
    return {"messages": tool_results}

# 4. Routing function (conditional edges)
def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        if state["iteration_count"] < 10:  # Prevent infinite loops
            return "tools"
    return "end"

# 5. Build the graph
graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", call_tools)

graph.set_entry_point("agent")
graph.add_conditional_edges(
    "agent",
    should_continue,
    {"tools": "tools", "end": END}
)
graph.add_edge("tools", "agent")  # Return to agent after tool execution

# 6. Add memory checkpointing
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# Run (manage conversation sessions with thread_id)
config = {"configurable": {"thread_id": "session_001"}}
result = app.invoke(
    {"messages": [HumanMessage(content="Search for AI agent trends in 2026 and summarize them")], "iteration_count": 0},
    config=config
)

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, FileWriterTool

search_tool = SerperDevTool()
file_writer = FileWriterTool()

researcher = Agent(
    role="AI Researcher",
    goal="Conduct in-depth research on the latest AI agent technology trends",
    backstory="""You are an expert researcher in the AI field.
    You analyze the latest papers, blogs, and GitHub repositories to extract key insights.""",
    tools=[search_tool],
    llm="claude-opus-4-5",
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Write readable technical reports from research results",
    backstory="""You are a professional writer who explains complex AI concepts clearly.""",
    tools=[file_writer],
    llm="claude-opus-4-5",
    verbose=True
)

research_task = Task(
    description="Research the Top 5 LLM agent trends of 2026. Include specific examples and impacts for each trend.",
    expected_output="Detailed analysis of 5 trends (500+ words each)",
    agent=researcher
)

writing_task = Task(
    description="Write a technical blog post based on the research results.",
    expected_output="A 2000-word technical blog post in Markdown format",
    agent=writer,
    output_file="ai_trends_2026.md"
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff()

import autogen

config_list = [{"model": "claude-opus-4-5", "api_key": "YOUR_KEY"}]

orchestrator = autogen.AssistantAgent(
    name="Orchestrator",
    system_message="""You are an orchestrator who coordinates the team.
    You analyze tasks and delegate them to the appropriate specialist agents.
    You integrate all results to generate the final answer.""",
    llm_config={"config_list": config_list}
)

coder = autogen.AssistantAgent(
    name="Coder",
    system_message="You are an expert at writing and executing Python code.",
    llm_config={"config_list": config_list}
)

user_proxy = autogen.UserProxyAgent(
    name="UserProxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    code_execution_config={"work_dir": "coding", "use_docker": False}
)

groupchat = autogen.GroupChat(
    agents=[orchestrator, coder, user_proxy],
    messages=[],
    max_round=12
)
manager = autogen.GroupChatManager(groupchat=groupchat)
user_proxy.initiate_chat(manager, message="Write a data visualization script")

import anthropic
import json

client = anthropic.Anthropic()

tools = [
    {
        "name": "get_stock_price",
        "description": "Retrieves the current stock price and change percentage for a specific ticker",
        "input_schema": {
            "type": "object",
            "properties": {
                "symbol": {
                    "type": "string",
                    "description": "Stock ticker symbol (e.g., AAPL, MSFT)"
                },
                "currency": {
                    "type": "string",
                    "enum": ["USD", "KRW"],
                    "description": "Display currency"
                }
            },
            "required": ["symbol"]
        }
    }
]

def process_tool_call(tool_name: str, tool_input: dict) -> str:
    if tool_name == "get_stock_price":
        return json.dumps({
            "symbol": tool_input["symbol"],
            "price": 185.92,
            "change_percent": "+2.3%",
            "currency": tool_input.get("currency", "USD")
        })

messages = [{"role": "user", "content": "What's the current Apple stock price?"}]

while True:
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=4096,
        tools=tools,
        messages=messages
    )

    messages.append({"role": "assistant", "content": response.content})

    if response.stop_reason == "end_turn":
        final_text = next(
            block.text for block in response.content
            if hasattr(block, "text")
        )
        print(f"Final answer: {final_text}")
        break

    if response.stop_reason == "tool_use":
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                result = process_tool_call(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result
                })
        messages.append({"role": "user", "content": tool_results})

from dataclasses import dataclass
from typing import List

@dataclass
class AgentTrajectory:
    task: str
    steps: List[dict]  # {"thought": ..., "action": ..., "observation": ...}
    final_answer: str
    success: bool
    total_tokens: int

def evaluate_trajectory(trajectory: AgentTrajectory) -> dict:
    """Trajectory-based agent evaluation"""
    metrics = {
        "task_success": trajectory.success,
        "efficiency": calculate_efficiency(trajectory.steps),
        "redundant_steps": count_redundant_steps(trajectory.steps),
        "error_recovery": check_error_recovery(trajectory.steps),
        "tool_usage_appropriateness": evaluate_tool_usage(trajectory.steps),
        "cost_efficiency": 1000 / trajectory.total_tokens
    }
    return metrics

def count_redundant_steps(steps: List[dict]) -> int:
    """Count unnecessary duplicate tool calls"""
    seen_actions = set()
    redundant = 0
    for step in steps:
        action_key = f"{step.get('action_type')}:{step.get('action_input')}"
        if action_key in seen_actions:
            redundant += 1
        seen_actions.add(action_key)
    return redundant

import anthropic
import base64
from PIL import ImageGrab

def take_screenshot() -> str:
    """Capture screen and encode as base64"""
    screenshot = ImageGrab.grab()
    screenshot.save("/tmp/screenshot.png")
    with open("/tmp/screenshot.png", "rb") as f:
        return base64.b64encode(f.read()).decode()

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=4096,
    tools=[
        {"type": "computer_20241022", "name": "computer", "display_width_px": 1920, "display_height_px": 1080},
        {"type": "text_editor_20241022", "name": "str_replace_editor"},
        {"type": "bash_20241022", "name": "bash"}
    ],
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "Open a browser and check the latest trending GitHub repositories"},
            {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": take_screenshot()}}
        ]
    }]
)

from openai import OpenAI
import time

client = OpenAI()

assistant = client.beta.assistants.create(
    name="AI Technology Analyst",
    instructions="You are an AI/ML technology expert. Analyze the latest papers and technical documents to provide insights.",
    model="gpt-4o",
    tools=[
        {"type": "file_search"},     # File-based RAG
        {"type": "code_interpreter"}  # Code execution
    ]
)

# Upload file and create vector store
vector_store = client.beta.vector_stores.create(name="AI Papers Repository")
with open("ai_papers_2026.pdf", "rb") as f:
    client.beta.vector_stores.file_batches.upload_and_poll(
        vector_store_id=vector_store.id,
        files=[("ai_papers_2026.pdf", f)]
    )

thread = client.beta.threads.create()
client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="Analyze the main limitations of Agentic AI from the uploaded papers"
)

run = client.beta.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id
)

while run.status in ["queued", "in_progress"]:
    run = client.beta.threads.runs.retrieve(thread_id=thread.id, run_id=run.id)
    time.sleep(1)

messages = client.beta.threads.messages.list(thread_id=thread.id)
print(messages.data[0].content[0].text.value)