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

class WebSearchInput(BaseModel):
    query: str = Field(description="検索するクエリ")
    max_results: int = Field(default=5, description="返す結果の数")

class TavilySearchTool(BaseTool):
    name: str = "web_search"
    description: str = "最新情報をWebで検索します。リアルタイム情報が必要なときに使用してください。"
    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"]
        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="Markdown形式の2000字技術ブログポスト",
    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="""あなたはチームを調整するオーケストレーターです。
    タスクを分析し、適切な専門エージェントに委任します。
    すべての結果を統合して最終回答を生成します。""",
    llm_config={"config_list": config_list}
)

coder = autogen.AssistantAgent(
    name="Coder",
    system_message="あなたはPythonコードを書いて実行する専門家です。",
    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="データ可視化コードを書いてください")

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", "JPY"],
                    "description": "表示通貨"
                }
            },
            "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": "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

@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()

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()

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 = client.beta.threads.create()
client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="アップロードされた論文からAgentic AIの主要な限界を分析してください"
)

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)