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

Thought: 현재 상황을 분석하고 다음 행동 결정
Action: tool_name(arguments) 형태로 도구 호출
Observation: 도구 실행 결과 수신
... 반복 ...
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()]

# ReAct 프롬프트 템플릿 로드
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": "2026년 현재 가장 인기 있는 AI 에이전트 프레임워크 3가지를 검색하고 비교표를 만들어줘"
})

cot_prompt = """
문제를 단계별로 풀어보세요:

문제: {problem}

풀이 과정:
1. 먼저 주어진 정보를 정리합니다.
2. 필요한 계산/추론을 수행합니다.
3. 중간 결과를 검증합니다.
4. 최종 답을 도출합니다.
"""

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,           # 각 레벨에서 생성할 가지 수
    c=4,           # 평가 깊이
    verbose=True
)

from mem0 import Memory

# mem0 초기화 (벡터 DB로 Qdrant 사용)
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"

# 메모리 저장
memory.add(
    messages=[
        {"role": "user", "content": "나는 Python을 주로 쓰고 FastAPI를 좋아해"},
        {"role": "assistant", "content": "알겠습니다! Python/FastAPI 선호를 기억할게요."}
    ],
    user_id=user_id
)

# 메모리 검색 및 활용
relevant_memories = memory.search(
    query="사용자가 선호하는 언어는?",
    user_id=user_id
)

context = "\n".join([m["memory"] for m in relevant_memories])
print(f"관련 메모리: {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):
        """대화/작업 에피소드를 메모리에 저장"""
        metadata = metadata or {}
        metadata["timestamp"] = datetime.now().isoformat()
        self.store.add_texts([content], metadatas=[metadata])

    def recall(self, query: str, k: int = 3):
        """관련 에피소드 검색"""
        docs = self.store.similarity_search(query, k=k)
        return [doc.page_content for doc in docs]

# 사용 예시
memory = EpisodicMemory()
memory.store_episode(
    "사용자가 FastAPI 프로젝트 구조에 대해 질문했고, 성공적으로 답변했다",
    {"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="검색할 쿼리")
    max_results: int = Field(default=5, description="반환할 결과 수")

class TavilySearchTool(BaseTool):
    name: str = "web_search"
    description: str = "최신 정보를 웹에서 검색합니다. 실시간 정보가 필요할 때 사용하세요."
    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"요약: {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 서버 구현 예시 (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="특정 도시의 현재 날씨를 조회합니다",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "도시명"},
                    "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"]
        # 실제 날씨 API 호출
        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. 상태 정의
class AgentState(TypedDict):
    messages: Annotated[Sequence, operator.add]
    tool_calls: list
    iteration_count: int

# 2. LLM 및 도구 설정
llm = ChatAnthropic(model="claude-opus-4-5")
tools = [WebSearchTool(), PythonREPLTool()]
llm_with_tools = llm.bind_tools(tools)

# 3. 노드 정의
def call_model(state: AgentState) -> AgentState:
    """LLM 호출 노드"""
    response = llm_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "iteration_count": state["iteration_count"] + 1
    }

def call_tools(state: AgentState) -> AgentState:
    """도구 실행 노드"""
    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. 라우팅 함수 (조건부 엣지)
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:  # 무한 루프 방지
            return "tools"
    return "end"

# 5. 그래프 구성
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")  # 도구 실행 후 다시 에이전트로

# 6. 메모리 체크포인트 추가
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# 실행 (thread_id로 대화 세션 관리)
config = {"configurable": {"thread_id": "session_001"}}
result = app.invoke(
    {"messages": [HumanMessage(content="2026년 AI 에이전트 트렌드를 검색하고 요약해줘")], "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 리서처",
    goal="최신 AI 에이전트 기술 트렌드를 심층 조사한다",
    backstory="""당신은 AI 분야 전문 리서처입니다.
    최신 논문, 블로그, GitHub를 분석하여 핵심 인사이트를 추출합니다.""",
    tools=[search_tool],
    llm="claude-opus-4-5",
    verbose=True
)

writer = Agent(
    role="기술 작가",
    goal="리서치 결과를 읽기 쉬운 기술 보고서로 작성한다",
    backstory="""당신은 복잡한 AI 개념을 명확하게 설명하는 전문 작가입니다.""",
    tools=[file_writer],
    llm="claude-opus-4-5",
    verbose=True
)

# 태스크 정의
research_task = Task(
    description="2026년 LLM 에이전트 트렌드 Top 5를 조사하세요. 각 트렌드마다 구체적인 사례와 영향을 포함하세요.",
    expected_output="5개 트렌드의 상세 분석 (각 500자 이상)",
    agent=researcher
)

writing_task = Task(
    description="리서치 결과를 바탕으로 기술 블로그 포스트를 작성하세요.",
    expected_output="마크다운 형식의 2000자 기술 블로그 포스트",
    agent=writer,
    output_file="ai_trends_2026.md"
)

# Crew 실행 (순차 프로세스)
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="""당신은 팀을 조율하는 오케스트레이터입니다.
    작업을 분석하고 적절한 전문가 에이전트에게 위임합니다.
    모든 결과를 통합하여 최종 답변을 생성합니다.""",
    llm_config={"config_list": config_list}
)

# 코드 실행 에이전트
coder = autogen.AssistantAgent(
    name="Coder",
    system_message="당신은 Python 코드를 작성하고 실행하는 전문가입니다.",
    llm_config={"config_list": config_list, "functions": [...]}
)

# 사용자 프록시 (코드 실행 담당)
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="데이터 시각화 코드를 작성해줘")

import anthropic
import json

client = anthropic.Anthropic()

# 도구 정의
tools = [
    {
        "name": "get_stock_price",
        "description": "특정 종목의 현재 주가와 변동률을 조회합니다",
        "input_schema": {
            "type": "object",
            "properties": {
                "symbol": {
                    "type": "string",
                    "description": "주식 종목 코드 (예: AAPL, MSFT)"
                },
                "currency": {
                    "type": "string",
                    "enum": ["USD", "KRW"],
                    "description": "표시 통화"
                }
            },
            "required": ["symbol"]
        }
    }
]

def process_tool_call(tool_name: str, tool_input: dict) -> str:
    """도구 실행 로직"""
    if tool_name == "get_stock_price":
        # 실제 API 호출 (예시)
        return json.dumps({
            "symbol": tool_input["symbol"],
            "price": 185.92,
            "change_percent": "+2.3%",
            "currency": tool_input.get("currency", "USD")
        })

# 에이전트 루프
messages = [{"role": "user", "content": "Apple 주가를 알려줘"}]

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_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, Optional

@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:
    """경로 기반 에이전트 평가"""
    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:
    """불필요한 중복 도구 호출 수"""
    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:
    """화면 캡처 후 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()

# Computer-use 에이전트
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": "브라우저를 열고 GitHub 최신 트렌딩 저장소를 확인해줘"},
            {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": take_screenshot()}}
        ]
    }]
)

from openai import OpenAI
import time

client = OpenAI()

# Assistants 생성 (도구 + 지식베이스 포함)
assistant = client.beta.assistants.create(
    name="AI 기술 분석가",
    instructions="당신은 AI/ML 기술 전문가입니다. 최신 논문과 기술 문서를 분석하여 인사이트를 제공합니다.",
    model="gpt-4o",
    tools=[
        {"type": "file_search"},   # 파일 기반 RAG
        {"type": "code_interpreter"}  # 코드 실행
    ]
)

# 파일 업로드 및 벡터 저장소 생성
vector_store = client.beta.vector_stores.create(name="AI 논문 저장소")
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 생성 및 메시지 추가
thread = client.beta.threads.create()
client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="업로드된 논문에서 Agentic AI의 주요 한계점을 분석해줘"
)

# Run 실행 및 결과 대기
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)