目次

1. [AIエージェントとは？](#1-aiエージェントとは)

2. [ReAct - 推論と行動](#2-react)

3. [ツール使用](#3-ツール使用)

4. [LangChain完全ガイド](#4-langchain)

5. [LangGraph - ステートマシンエージェント](#5-langgraph)

6. [LlamaIndexエージェント](#6-llamaindex)

7. [CrewAI - マルチエージェント協調](#7-crewai)

8. [エージェントメモリ](#8-エージェントメモリ)

9. [コード実行エージェント](#9-コード実行エージェント)

10. [エージェント評価とモニタリング](#10-エージェント評価とモニタリング)

1. AIエージェントとは？

1.1 エージェントの定義

AIエージェントとは、**環境を認識し、行動を選択・実行して目標を達成する自律システム**のことです。

単純なチャットボットとの比較：

| 特性 | チャットボット | AIエージェント |

| -------- | ---------------- | -------------------------------- |

| 行動能力 | テキスト生成のみ | ツール使用、コード実行、検索など |

| 計画 | なし | マルチステップ計画 |

| メモリ | 会話内のみ | 長期記憶が可能 |

| 自律性 | 低 | 高 |

| 実行 | 単一応答 | 目標達成まで反復 |

1.2 エージェントの4つのコアコンポーネント

**1. LLM（脳）**

すべての推論と判断を担当。「次に何をすべきか？」「この結果は目標を満たしているか？」といった質問に答えます。

**2. ツール使用（手）**

外部世界とのインターフェース。Web検索、計算機、コード実行、データベースクエリ、APIコールなどが含まれます。

**3. メモリ（記憶）**

短期記憶（会話履歴）、長期記憶（ベクターDB）、エピソード記憶（過去の経験）。

**4. プランニング（戦略）**

複雑な目標をサブタスクに分解し、実行順序を決定します。

1.3 エージェントの実行ループ

ユーザー目標入力

↓

[計画]

目標をサブタスクに分解

↓

[行動選択]

次の行動を決定（どのツールを使うか？）

↓

[ツール実行]

選択したツールを実行

↓

[結果観察]

ツールの出力を確認

↓

[目標達成？] → Yes → 最終回答を生成

↓ No

行動選択に戻る

1.4 AIエージェントの応用例

- **リサーチエージェント**: Web検索 → 情報収集 → 要約レポート

- **コードエージェント**: 要件分析 → コード作成 → テスト実行

- **データ分析エージェント**: データ読み込み → 分析 → 可視化

- **カスタマーサービスエージェント**: クエリ特定 → システム照会 → 応答

- **DevOpsエージェント**: 監視 → 問題検出 → 自動修復

2. ReAct

ReAct（Reasoning + Acting）は2022年に発表されたエージェントフレームワークです。コアは「考えて行動する」ループです。

2.1 ReActフレームワーク

従来のChain-of-Thought（CoT）は考えるだけです。エージェントには考える + 行動する + 観察するが必要です。

Thought: この質問に答えるために最新の株価データが必要

Action: Search["サムスン電子 株価 2026"]

Observation: サムスン電子現在の株価: 78,000ウォン、前日比+2.3%

Thought: 価格データを取得した。次に変動額を計算すべきだ

Action: Calculator[78000 * 0.023]

Observation: 1794

Thought: 株価は昨日より1,794ウォン上昇した。これで回答できる

Final Answer: サムスン電子は現在78,000ウォンで、前日より1,794ウォン（+2.3%）上昇しています。

2.2 ReActエージェントの実装

from langchain import hub

from langchain.agents import create_react_agent, AgentExecutor

from langchain_openai import ChatOpenAI

from langchain.tools import Tool

from langchain_community.tools import DuckDuckGoSearchRun

from langchain_core.prompts import PromptTemplate

REACT_TEMPLATE = """You are a helpful AI assistant.

You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer

Thought: you should always think about what to do

Action: the action to take, should be one of [{tool_names}]

Action Input: the input to the action

Observation: the result of the action

... (this Thought/Action/Observation can repeat N times)

Thought: I now know the final answer

Final Answer: the final answer to the original input question

Begin!

Question: {input}

Thought:{agent_scratchpad}"""

react_prompt = PromptTemplate.from_template(REACT_TEMPLATE)

llm = ChatOpenAI(model="gpt-4o", temperature=0)

search = DuckDuckGoSearchRun()

tools = [

Tool(

name="Search",

func=search.run,

description="Useful for searching for current events and up-to-date data"

),

Tool(

name="Calculator",

func=lambda x: eval(x),

description="Useful for math calculations. Input is a Python expression"

)

]

agent = create_react_agent(llm, tools, react_prompt)

agent_executor = AgentExecutor(

agent=agent,

tools=tools,

verbose=True,

max_iterations=10,

handle_parsing_errors=True

)

result = agent_executor.invoke({

"input": "2026年3月のビットコインの現在価格は？1年前と比較してどうですか？"

})

print(result["output"])

2.3 ReActの限界

- **ハルシネーション**: 存在しないツールアクションを生成する可能性がある

- **無限ループ**: 終了条件なしに繰り返す可能性がある

- **長いコンテキスト**: 蓄積されたThought/Action/Observationのステップがコンテキストウィンドウを超える可能性がある

3. ツール使用

ツールはエージェントが外部世界とインターフェースする方法です。

3.1 OpenAI Function Calling

OpenAIのFunction Callingにより、LLMが構造化された方法で関数を呼び出せます。

from openai import OpenAI

client = OpenAI()

functions = [

{

"name": "get_weather",

"description": "特定の都市の現在の天気を取得",

"parameters": {

"type": "object",

"properties": {

"city": {

"type": "string",

"description": "天気を取得する都市名"

},

"unit": {

"type": "string",

"enum": ["celsius", "fahrenheit"],

"description": "温度の単位"

}

},

"required": ["city"]

}

},

{

"name": "search_database",

"description": "内部データベースから情報を検索",

"parameters": {

"type": "object",

"properties": {

"query": {

"type": "string",

"description": "検索クエリ"

},

"table": {

"type": "string",

"enum": ["users", "products", "orders"],

"description": "検索するテーブル"

}

},

"required": ["query"]

}

}

]

def get_weather(city: str, unit: str = "celsius") -> dict:

return {

"city": city,

"temperature": 15,

"unit": unit,

"condition": "sunny",

"humidity": 60

}

def search_database(query: str, table: str = "products") -> list:

return [{"id": 1, "name": "Sample Product", "price": 100}]

available_tools = {

"get_weather": get_weather,

"search_database": search_database

}

def run_agent_with_tools(user_message: str) -> str:

messages = [{"role": "user", "content": user_message}]

while True:

response = client.chat.completions.create(

model="gpt-4o",

messages=messages,

tools=[{"type": "function", "function": f} for f in functions],

tool_choice="auto"

)

message = response.choices[0].message

if not message.tool_calls:

return message.content

messages.append(message)

for tool_call in message.tool_calls:

func_name = tool_call.function.name

func_args = json.loads(tool_call.function.arguments)

print(f" ツールコール: {func_name}({func_args})")

if func_name in available_tools:

result = available_tools[func_name](**func_args)

else:

result = {"error": f"Unknown function: {func_name}"}

messages.append({

"role": "tool",

"tool_call_id": tool_call.id,

"content": json.dumps(result)

})

answer = run_agent_with_tools("ロンドンの天気は？傘が必要ですか？")

print(answer)

3.2 多様なツールの例

from langchain.tools import tool

from langchain_community.tools import WikipediaQueryRun

from langchain_community.utilities import WikipediaAPIWrapper

@tool

def calculator(expression: str) -> str:

"""数学的計算を実行。Python式を入力。"""

try:

result = eval(expression)

return str(result)

except Exception as e:

return f"計算エラー: {e}"

@tool

def run_python_code(code: str) -> str:

"""Pythonコードを実行して結果を返す。"""

try:

local_vars = {}

exec(code, {"__builtins__": {}}, local_vars)

output = local_vars.get('result', '結果変数が見つかりません')

return str(output)

except Exception as e:

return f"コード実行エラー: {e}"

@tool

def query_database(sql: str) -> str:

"""SQLiteデータベースに対してSQLクエリを実行。"""

try:

conn = sqlite3.connect("agent_db.sqlite")

cursor = conn.cursor()

cursor.execute(sql)

rows = cursor.fetchall()

conn.close()

return str(rows)

except Exception as e:

return f"DBエラー: {e}"

@tool

def send_email(to: str, subject: str, body: str) -> str:

"""メールを送信する。"""

print(f"送信先: {to}")

print(f"件名: {subject}")

print(f"本文: {body[:100]}...")

return f"{to}へのメール送信完了。"

wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())

@tool

def search_wikipedia(query: str) -> str:

"""Wikipediaで情報を検索する。"""

return wikipedia.run(query)

4. LangChain

LangChainはLLMアプリケーション構築のための標準フレームワークです。

4.1 LangChainのコアコンポーネント

LangChainアーキテクチャ

├── Models (LLM, Chat Models, Embeddings)

├── Prompts (PromptTemplate, ChatPromptTemplate)

├── Chains (LLMChain, SequentialChain, LCEL)

├── Memory (Buffer, Summary, VectorStore)

├── Agents (ReAct, OpenAI Functions)

├── Tools (Built-in + Custom)

└── Retrievers (VectorStore, MultiQuery)

4.2 LCEL（LangChain Expression Language）

モダンなLangChainはLCELパイプラインを使用します。

from langchain_openai import ChatOpenAI

from langchain_core.prompts import ChatPromptTemplate

from langchain_core.output_parsers import StrOutputParser

from langchain_core.runnables import RunnableParallel

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

基本チェーン

prompt = ChatPromptTemplate.from_messages([

("system", "あなたは専門的なアナリストです。"),

("human", "{question}")

])

chain = prompt | llm | StrOutputParser()

result = chain.invoke({"question": "AIエージェントの利点は何ですか？"})

print(result)

構造化出力

from pydantic import BaseModel, Field

class AnalysisResult(BaseModel):

summary: str = Field(description="要約")

key_points: list[str] = Field(description="重要ポイントのリスト")

recommendation: str = Field(description="推奨事項")

structured_chain = prompt | llm.with_structured_output(AnalysisResult)

result = structured_chain.invoke({"question": "LangChainとLlamaIndexを比較してください"})

print(result.summary)

print(result.key_points)

並列チェーン

parallel_chain = RunnableParallel({

"pros": ChatPromptTemplate.from_template("{topic}の利点は何ですか？") | llm | StrOutputParser(),

"cons": ChatPromptTemplate.from_template("{topic}のデメリットは何ですか？") | llm | StrOutputParser(),

})

result = parallel_chain.invoke({"topic": "AIエージェント"})

print("利点:", result["pros"])

print("デメリット:", result["cons"])

4.3 メモリ管理

from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory

from langchain.chains import ConversationChain

from langchain_openai import ChatOpenAI

バッファメモリ（全メッセージを保持）

buffer_memory = ConversationBufferMemory(

memory_key="history",

return_messages=True

)

サマリーメモリ（古い会話を要約）

summary_memory = ConversationSummaryMemory(

llm=ChatOpenAI(model="gpt-4o-mini"),

memory_key="history",

return_messages=True

)

llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

conversation = ConversationChain(

llm=llm,

memory=buffer_memory,

verbose=True

)

r1 = conversation.predict(input="私の名前はアリスです")

r2 = conversation.predict(input="私の名前は何ですか？") # 記憶している！

print(r1, r2)

ベクターストアベースの長期記憶

from langchain.memory import VectorStoreRetrieverMemory

from langchain_openai import OpenAIEmbeddings

from langchain_community.vectorstores import FAISS

embeddings = OpenAIEmbeddings()

vectorstore = FAISS.from_texts(["dummy"], embeddings)

retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

vector_memory = VectorStoreRetrieverMemory(retriever=retriever)

vector_memory.save_context(

{"input": "好きな食べ物は寿司です"},

{"output": "分かりました！"}

)

relevant = vector_memory.load_memory_variables({"prompt": "食べ物を推薦してください"})

print(relevant)

4.4 RAG（Retrieval Augmented Generation）

from langchain_community.document_loaders import WebBaseLoader

from langchain_text_splitters import RecursiveCharacterTextSplitter

from langchain_openai import OpenAIEmbeddings, ChatOpenAI

from langchain_community.vectorstores import Chroma

from langchain.chains import RetrievalQA

from langchain_core.prompts import ChatPromptTemplate

loader = WebBaseLoader("https://example.com/document")

documents = loader.load()

splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)

chunks = splitter.split_documents(documents)

embeddings = OpenAIEmbeddings()

vectorstore = Chroma.from_documents(chunks, embeddings)

llm = ChatOpenAI(model="gpt-4o", temperature=0)

rag_prompt = ChatPromptTemplate.from_template("""

以下のコンテキストのみに基づいて質問に回答してください。

コンテキストに答えがない場合は、分からないと伝えてください。

コンテキスト:

{context}

質問: {question}

回答:""")

qa_chain = RetrievalQA.from_chain_type(

llm=llm,

retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),

chain_type_kwargs={"prompt": rag_prompt},

return_source_documents=True

)

result = qa_chain.invoke({"query": "主要なポイントは何ですか？"})

print("回答:", result["result"])

print("ソース:", [doc.metadata for doc in result["source_documents"]])

4.5 完全なLangChainエージェント

from langchain_openai import ChatOpenAI

from langchain.agents import create_openai_tools_agent, AgentExecutor

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

from langchain.memory import ConversationBufferWindowMemory

from langchain.tools import tool

from langchain_community.tools import DuckDuckGoSearchRun

search = DuckDuckGoSearchRun()

@tool

def get_current_datetime() -> str:

"""現在の日時を返す。"""

return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool

def calculate(expression: str) -> str:

"""数学的計算を実行。例: 2+2, 10*5, sqrt(16)"""

safe_dict = {k: getattr(math, k) for k in dir(math) if not k.startswith('_')}

safe_dict['abs'] = abs

try:

return str(eval(expression, {"__builtins__": {}}, safe_dict))

except Exception as e:

return f"計算エラー: {e}"

@tool

def web_search(query: str) -> str:

"""最新情報をWebで検索する。"""

return search.run(query)

tools = [get_current_datetime, calculate, web_search]

prompt = ChatPromptTemplate.from_messages([

("system", """あなたは役立つAIアシスタントです。

ユーザーの質問に正確かつ丁寧に回答してください。

情報収集に必要な場合はツールを使用してください。"""),

MessagesPlaceholder(variable_name="chat_history"),

("human", "{input}"),

MessagesPlaceholder(variable_name="agent_scratchpad"),

])

llm = ChatOpenAI(model="gpt-4o", temperature=0)

memory = ConversationBufferWindowMemory(

memory_key="chat_history",

return_messages=True,

k=10

)

agent = create_openai_tools_agent(llm, tools, prompt)

agent_executor = AgentExecutor(

agent=agent,

tools=tools,

memory=memory,

verbose=True,

max_iterations=5,

handle_parsing_errors=True

)

def chat(message: str) -> str:

result = agent_executor.invoke({"input": message})

return result["output"]

print(chat("今日の日付は？"))

print(chat("AIエージェントの最近のトレンドを教えてください"))

print(chat("今話した会社の時価総額はどれくらいですか？")) # メモリを使用

5. LangGraph

LangGraphはエージェントを**ステートマシン**として実装します。複雑なループ、分岐、条件付き実行が可能になります。

5.1 なぜLangGraphか？

標準LangChainエージェントの限界：

- 線形実行のみ（ループが難しい）

- 不便な状態管理

- 複雑な分岐

- Human-in-the-loopが困難

LangGraphのソリューション：

- グラフベースの実行フロー

- 明示的な状態管理

- 条件付きエッジによる分岐

- 割り込みポイント

5.2 LangGraphの基礎

from langgraph.graph import StateGraph, END

from langgraph.prebuilt import ToolNode

from langchain_openai import ChatOpenAI

from langchain_core.messages import HumanMessage

from typing import TypedDict, Annotated, Sequence

状態スキーマの定義

class AgentState(TypedDict):

messages: Annotated[Sequence, operator.add]

next: str

llm = ChatOpenAI(model="gpt-4o", temperature=0)

tools = [web_search, calculate, get_current_datetime]

llm_with_tools = llm.bind_tools(tools)

def agent_node(state: AgentState) -> AgentState:

"""LLMが次のアクションを決定"""

messages = state["messages"]

response = llm_with_tools.invoke(messages)

return {"messages": [response]}

def should_continue(state: AgentState) -> str:

"""続行するか終了するかを決定（条件付きエッジ）"""

messages = state["messages"]

last_message = messages[-1]

if hasattr(last_message, 'tool_calls') and last_message.tool_calls:

return "tools"

return END

tool_node = ToolNode(tools)

workflow = StateGraph(AgentState)

workflow.add_node("agent", agent_node)

workflow.add_node("tools", tool_node)

workflow.set_entry_point("agent")

workflow.add_conditional_edges(

"agent",

should_continue,

{

"tools": "tools",

END: END

}

)

workflow.add_edge("tools", "agent")

app = workflow.compile()

result = app.invoke({

"messages": [HumanMessage(content="ロンドンの天気は？")]

})

print(result["messages"][-1].content)

5.3 Human-in-the-Loop

from langgraph.graph import StateGraph, END

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()

class ApprovalState(TypedDict):

messages: Annotated[Sequence, operator.add]

pending_action: str

approved: bool

def agent_node(state: ApprovalState) -> ApprovalState:

messages = state["messages"]

response = llm_with_tools.invoke(messages)

if hasattr(response, 'tool_calls') and response.tool_calls:

tool_name = response.tool_calls[0]['name']

if tool_name in ["send_email", "delete_file", "make_payment"]:

return {

"messages": [response],

"pending_action": tool_name,

"approved": False

}

return {"messages": [response]}

def human_approval_node(state: ApprovalState) -> ApprovalState:

"""人間の承認ノード（割り込み）"""

print(f"\n承認が必要なアクション: {state['pending_action']}")

print("'approve'で続行、'reject'でキャンセル")

return state

def check_approval(state: ApprovalState) -> str:

if state.get("approved"):

return "execute"

elif state.get("pending_action") and not state.get("approved"):

return "human_approval"

return END

workflow = StateGraph(ApprovalState)

workflow.add_node("agent", agent_node)

workflow.add_node("human_approval", human_approval_node)

workflow.add_node("tools", ToolNode(tools))

workflow.set_entry_point("agent")

workflow.add_conditional_edges("agent", check_approval, {

"human_approval": "human_approval",

"execute": "tools",

END: END

})

workflow.add_edge("tools", "agent")

app = workflow.compile(

checkpointer=memory,

interrupt_before=["human_approval"]

)

thread_id = "session_001"

config = {"configurable": {"thread_id": thread_id}}

result = app.invoke(

{"messages": [HumanMessage(content="チームに会議招待を送ってください")]},

config=config

)

人間の承認後、再開

app.update_state(config, {"approved": True})

final_result = app.invoke(None, config=config)

5.4 LangGraphを使用したリサーチエージェント

from langgraph.graph import StateGraph, END

from langchain_openai import ChatOpenAI

from langchain_core.messages import HumanMessage

from typing import TypedDict, List

class ResearchState(TypedDict):

topic: str

search_queries: List[str]

search_results: List[str]

draft: str

final_report: str

iteration: int

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def plan_queries(state: ResearchState) -> ResearchState:

topic = state["topic"]

response = llm.invoke([

HumanMessage(content=f"""トピック: {topic}

このトピックを徹底的にリサーチするための5つの検索クエリを生成してください。

JSONで返してください: {{"queries": ["query1", "query2", ...]}}""")

])

queries = json.loads(response.content)["queries"]

return {"search_queries": queries}

def execute_searches(state: ResearchState) -> ResearchState:

queries = state["search_queries"]

results = []

for query in queries:

result = search.run(query)

results.append(f"[{query}]\n{result}")

return {"search_results": results}

def write_draft(state: ResearchState) -> ResearchState:

topic = state["topic"]

results = "\n\n".join(state["search_results"])

response = llm.invoke([

HumanMessage(content=f"""トピック: {topic}

収集した情報:

{results}

上記に基づいて詳細なリサーチレポートの草稿を作成してください。""")

])

return {"draft": response.content, "iteration": state.get("iteration", 0) + 1}

def review_and_improve(state: ResearchState) -> ResearchState:

draft = state["draft"]

response = llm.invoke([

HumanMessage(content=f"""以下のリサーチレポートの草稿をレビューして改善してください:

{draft}

改善点:

1. 正確性の確認

2. 論理的な流れの改善

3. 重要情報の追加

4. 結論の強化

最終レポートを作成してください。""")

])

return {"final_report": response.content}

def should_improve(state: ResearchState) -> str:

if state.get("iteration", 0) < 2:

return "improve"

return "finalize"

research_graph = StateGraph(ResearchState)

research_graph.add_node("plan_queries", plan_queries)

research_graph.add_node("execute_searches", execute_searches)

research_graph.add_node("write_draft", write_draft)

research_graph.add_node("review_and_improve", review_and_improve)

research_graph.set_entry_point("plan_queries")

research_graph.add_edge("plan_queries", "execute_searches")

research_graph.add_edge("execute_searches", "write_draft")

research_graph.add_conditional_edges(

"write_draft",

should_improve,

{

"improve": "execute_searches",

"finalize": "review_and_improve"

}

)

research_graph.add_edge("review_and_improve", END)

research_app = research_graph.compile()

result = research_app.invoke({

"topic": "2026年のAIエージェント技術トレンド",

"search_queries": [],

"search_results": [],

"draft": "",

"final_report": "",

"iteration": 0

})

print(result["final_report"])

6. LlamaIndex

LlamaIndexはデータ中心のAIエージェントフレームワークです。

6.1 LlamaIndexエージェント

from llama_index.core.agent import ReActAgent

from llama_index.core.tools import FunctionTool, QueryEngineTool

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

from llama_index.llms.openai import OpenAI

from llama_index.core.settings import Settings

Settings.llm = OpenAI(model="gpt-4o", temperature=0)

def multiply(a: float, b: float) -> float:

"""2つの数を掛け算する。"""

return a * b

def add(a: float, b: float) -> float:

"""2つの数を足し算する。"""

return a + b

multiply_tool = FunctionTool.from_defaults(fn=multiply)

add_tool = FunctionTool.from_defaults(fn=add)

documents = SimpleDirectoryReader("./data").load_data()

index = VectorStoreIndex.from_documents(documents)

query_engine = index.as_query_engine(similarity_top_k=3)

query_tool = QueryEngineTool.from_defaults(

query_engine=query_engine,

name="knowledge_base",

description="社内文書から情報を検索する。"

)

agent = ReActAgent.from_tools(

[multiply_tool, add_tool, query_tool],

llm=Settings.llm,

verbose=True,

max_iterations=10

)

response = agent.chat(

"社内文書でAIポリシーを見つけて、500ドルと1000ドルの罰金額を合計してください"

)

print(response)

6.2 マルチドキュメントRAGエージェント

from llama_index.core.agent import ReActAgent

from llama_index.core.tools import QueryEngineTool

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

from llama_index.core.node_parser import SentenceSplitter

docs_finance = SimpleDirectoryReader("./finance_docs").load_data()

docs_hr = SimpleDirectoryReader("./hr_docs").load_data()

docs_technical = SimpleDirectoryReader("./technical_docs").load_data()

splitter = SentenceSplitter(chunk_size=512)

finance_index = VectorStoreIndex.from_documents(docs_finance, transformations=[splitter])

hr_index = VectorStoreIndex.from_documents(docs_hr, transformations=[splitter])

tech_index = VectorStoreIndex.from_documents(docs_technical, transformations=[splitter])

tools = [

QueryEngineTool.from_defaults(

query_engine=finance_index.as_query_engine(),

name="finance_qa",

description="財務、会計、予算に関する質問に答える"

),

QueryEngineTool.from_defaults(

query_engine=hr_index.as_query_engine(),

name="hr_qa",

description="HR、採用、福利厚生に関する質問に答える"

),

QueryEngineTool.from_defaults(

query_engine=tech_index.as_query_engine(),

name="tech_qa",

description="技術仕様と開発ガイドに関する質問に答える"

),

]

agent = ReActAgent.from_tools(tools, verbose=True)

response = agent.chat("2026年のIT予算と新しい採用計画について教えてください")

print(response)

7. CrewAI

CrewAIはロールベースのマルチエージェント協調フレームワークです。

7.1 CrewAIのコアコンセプト

Crew

├── Agents - それぞれのロールとゴールを持つ

│ ├── Role: "シニアリサーチャー", "コンテンツライター"

│ ├── Goal: エージェントが達成しようとすること

│ ├── Backstory: 個性/専門知識

│ └── Tools: 利用可能なツール

└── Tasks

├── Description: 何をすべきか

├── Expected Output: 期待される結果

└── Agent: 担当エージェント

7.2 リサーチチームエージェント

from crewai import Agent, Task, Crew, Process

from crewai_tools import SerperDevTool, WebsiteSearchTool

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

search_tool = SerperDevTool()

web_tool = WebsiteSearchTool()

researcher = Agent(

role="シニアリサーチャー",

goal="与えられたトピックに関する包括的で正確な情報を収集する",

backstory="""あなたは10年の経験を持つプロフェッショナルなリサーチャーです。

複雑なトピックを体系的に調査し、信頼性の高い最新ソースから

重要な洞察を抽出することが得意です。""",

tools=[search_tool, web_tool],

llm=llm,

verbose=True

)

analyst = Agent(

role="データアナリスト",

goal="収集した情報を分析してパターンとトレンドを特定する",

backstory="""あなたはデータ分析の専門家です。生データから意味のあるパターンを発見し、

統計的手法とビジネス知識を組み合わせて実用的な洞察を導き出します。""",

tools=[search_tool],

llm=llm,

verbose=True

)

writer = Agent(

role="コンテンツライター",

goal="分析結果から明確で説得力のあるレポートを作成する",

backstory="""あなたは技術的な内容を一般の読者に説明できるプロフェッショナルなライターです。

複雑な分析結果をストーリーテリングで伝えることが好きです。""",

llm=llm,

verbose=True

)

research_task = Task(

description="""トピック「AI エージェント市場分析 2026」をリサーチ:

1. 最新のトレンドと動向

2. 主要プレイヤーとそのアプローチ

3. 潜在的な機会とリスク

4. 関連する統計とデータ

少なくとも5つの信頼できるソースを引用してください。""",

expected_output="リサーチサマリー（最低500字）",

agent=researcher

)

analysis_task = Task(

description="""リサーチャーが収集した情報を分析:

1. 3つの重要なトレンドを特定

2. SWOT分析を実施

3. 短期予測（6〜12ヶ月）

4. 主要なリスク要因

客観的でデータ駆動の分析を提供してください。""",

expected_output="分析レポート（最低400字）",

agent=analyst,

context=[research_task]

)

report_task = Task(

description="""リサーチと分析を統合してプロフェッショナルなレポートを作成:

レポート構造:

1. エグゼクティブサマリー

2. 現状分析

3. 重要な洞察

4. 推奨事項

5. 結論

プロフェッショナルかつ説得力を持って書いてください。""",

expected_output="完成したレポート（最低800字）",

agent=writer,

context=[research_task, analysis_task]

)

research_crew = Crew(

agents=[researcher, analyst, writer],

tasks=[research_task, analysis_task, report_task],

process=Process.sequential,

verbose=True

)

result = research_crew.kickoff(inputs={"topic": "AIエージェント市場分析2026"})

print(result)

7.3 ソフトウェア開発エージェントチーム

from crewai import Agent, Task, Crew, Process

from crewai_tools import CodeInterpreterTool

code_interpreter = CodeInterpreterTool()

product_manager = Agent(

role="プロダクトマネージャー",

goal="要件を明確に定義して開発計画を作成する",

backstory="技術要件をビジネス目標に結びつける10年の経験を持つPM。",

llm=llm,

verbose=True

)

senior_developer = Agent(

role="シニアデベロッパー",

goal="高品質でスケーラブルなコードを書く",

backstory="Python、FastAPI、Reactに精通したフルスタックデベロッパー。",

tools=[code_interpreter],

llm=llm,

verbose=True

)

qa_engineer = Agent(

role="QAエンジニア",

goal="コードを徹底的にテストして品質を確保する",

backstory="バグを見つけることが好きなソフトウェアテストの専門家。",

tools=[code_interpreter],

llm=llm,

verbose=True

)

requirements_task = Task(

description="""ユーザー認証APIの技術要件を定義:

1. ユーザーストーリー

2. 機能要件リスト

3. 非機能要件（パフォーマンス、セキュリティ）

4. APIデザイン（エンドポイントリスト）""",

expected_output="要件ドキュメント",

agent=product_manager

)

development_task = Task(

description="""要件ドキュメントに基づいてPython FastAPIコードを作成:

1. 完全なAPI実装

2. データモデル（Pydantic）

3. エラーハンドリング

4. コードコメント""",

expected_output="完全なPythonコード",

agent=senior_developer,

context=[requirements_task]

)

testing_task = Task(

description="""作成されたコードのレビューとテスト:

1. コードレビュー（バグ、セキュリティ脆弱性）

2. ユニットテストの作成

3. エッジケーステスト

4. 改善提案""",

expected_output="テストレポートと改善されたコード",

agent=qa_engineer,

context=[development_task]

)

dev_crew = Crew(

agents=[product_manager, senior_developer, qa_engineer],

tasks=[requirements_task, development_task, testing_task],

process=Process.sequential,

verbose=True

)

result = dev_crew.kickoff(

inputs={"feature_request": "ユーザー認証API（JWT ベース）"}

)

print(result)

7.4 階層的CrewAI

マネージャーが作業を委任する階層的構造

manager = Agent(

role="プロジェクトマネージャー",

goal="チームをコーディネートして最高の結果を出す",

backstory="各チームメンバーの強みを最大化する経験豊富なPM。",

llm=llm,

verbose=True,

allow_delegation=True # 他のエージェントに委任可能

)

hierarchical_crew = Crew(

agents=[manager, researcher, analyst, writer],

tasks=[report_task], # 最終タスクのみ定義（残りは自動分配）

process=Process.hierarchical,

manager_agent=manager,

verbose=True

)

8. エージェントメモリ

8.1 メモリアーキテクチャ

from langchain.memory import (

ConversationBufferMemory,

ConversationSummaryBufferMemory,

ConversationEntityMemory,

)

from langchain_community.vectorstores import FAISS

from langchain_openai import OpenAIEmbeddings, ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

embeddings = OpenAIEmbeddings()

1. 短期記憶 - 直近のNメッセージ

from langchain.memory import ConversationBufferWindowMemory

short_term = ConversationBufferWindowMemory(k=5, return_messages=True)

2. サマリーメモリ - 古い会話を要約

summary_memory = ConversationSummaryBufferMemory(

llm=llm,

max_token_limit=1000,

return_messages=True

)

3. エンティティメモリ - 重要な事実を抽出

entity_memory = ConversationEntityMemory(llm=llm, return_messages=True)

4. 長期記憶 - ベクターDB

class LongTermMemory:

def __init__(self):

self.vectorstore = FAISS.from_texts(["init"], embeddings)

self.retriever = self.vectorstore.as_retriever(search_kwargs={"k": 5})

def save(self, text: str, metadata: dict = None):

self.vectorstore.add_texts([text], metadatas=[metadata or {}])

def recall(self, query: str) -> list:

docs = self.retriever.get_relevant_documents(query)

return [doc.page_content for doc in docs]

5. エピソード記憶 - 過去のエージェント経験

class EpisodicMemory:

def __init__(self):

self.episodes = []

def save_episode(self, task: str, actions: list, result: str, success: bool):

episode = {

"task": task,

"actions": actions,

"result": result,

"success": success,

"timestamp": datetime.datetime.now().isoformat()

}

self.episodes.append(episode)

def find_similar_episodes(self, current_task: str) -> list:

return [e for e in self.episodes if e["success"]]

統合メモリシステム

class AgentMemorySystem:

def __init__(self):

self.short_term = ConversationBufferWindowMemory(k=10)

self.long_term = LongTermMemory()

self.episodic = EpisodicMemory()

self.entities = {}

def save_message(self, role: str, content: str):

self.short_term.save_context(

{"input": content if role == "human" else ""},

{"output": content if role == "ai" else ""}

)

def get_relevant_context(self, query: str) -> str:

recent = self.short_term.load_memory_variables({})

long_term = self.long_term.recall(query)

past_episodes = self.episodic.find_similar_episodes(query)

context = f"""最近の会話: {recent.get('history', '')}

関連する記憶: {'; '.join(long_term[:3])}

類似した過去の経験: {past_episodes[:2] if past_episodes else 'なし'}"""

return context

memory_system = AgentMemorySystem()

9. コード実行エージェント

9.1 Python REPLエージェント

from langchain_experimental.tools import PythonREPLTool

from langchain.agents import create_openai_tools_agent, AgentExecutor

from langchain_openai import ChatOpenAI

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

python_repl = PythonREPLTool()

llm = ChatOpenAI(model="gpt-4o", temperature=0)

data_analysis_prompt = ChatPromptTemplate.from_messages([

("system", """あなたはプロフェッショナルなデータアナリストです。

Python、pandas、matplotlib、seabornに精通しています。

データ分析のリクエストを受けたら、コードを書いて実行し、結果を導き出してください。

常にコードと一緒に分析結果を説明してください。"""),

MessagesPlaceholder(variable_name="chat_history"),

("human", "{input}"),

MessagesPlaceholder(variable_name="agent_scratchpad"),

])

agent = create_openai_tools_agent(llm, [python_repl], data_analysis_prompt)

data_agent = AgentExecutor(agent=agent, tools=[python_repl], verbose=True)

result = data_agent.invoke({

"input": """以下のデータを分析してください:

sales = [100, 150, 120, 200, 180, 250, 220, 300, 280, 350, 320, 400]

months = ['1月', '2月', '3月', '4月', '5月', '6月',

'7月', '8月', '9月', '10月', '11月', '12月']

月次売上トレンドを分析して成長率を計算してください。""",

"chat_history": []

})

print(result["output"])

9.2 Dockerサンドボックスコード実行

class DockerCodeExecutor:

"""Dockerコンテナ内でコードを安全に実行する"""

def __init__(self, image="python:3.11-slim", timeout=30):

self.client = docker.from_env()

self.image = image

self.timeout = timeout

def execute(self, code: str, packages: list = None) -> dict:

"""

DockerコンテナでPythonコードを実行

Returns: {success: bool, output: str, error: str}

"""

with tempfile.TemporaryDirectory() as tmpdir:

code_file = os.path.join(tmpdir, "script.py")

with open(code_file, "w") as f:

f.write(code)

install_cmd = ""

if packages:

pkgs = " ".join(packages)

install_cmd = f"pip install {pkgs} -q && "

try:

container = self.client.containers.run(

self.image,

command=f'sh -c "{install_cmd}python /code/script.py"',

volumes={tmpdir: {"bind": "/code", "mode": "ro"}},

remove=True,

network_mode="none", # ネットワークアクセスをブロック

mem_limit="256m", # メモリ制限

cpu_period=100000,

cpu_quota=50000, # CPU使用率50%制限

timeout=self.timeout,

stdout=True,

stderr=True

)

return {

"success": True,

"output": container.decode("utf-8"),

"error": ""

}

except docker.errors.ContainerError as e:

return {

"success": False,

"output": "",

"error": e.stderr.decode("utf-8") if e.stderr else str(e)

}

except Exception as e:

return {"success": False, "output": "", "error": str(e)}

executor = DockerCodeExecutor()

code = """

data = {'name': ['Alice', 'Bob', 'Charlie'], 'score': [85, 92, 78]}

df = pd.DataFrame(data)

result = df.describe().to_dict()

print(json.dumps(result, indent=2))

"""

result = executor.execute(code, packages=["pandas"])

print(result["output"])

10. エージェント評価とモニタリング

10.1 LangSmithトレーシング

from langsmith import Client

LangSmithの設定

os.environ["LANGCHAIN_TRACING_V2"] = "true"

os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"

os.environ["LANGCHAIN_PROJECT"] = "ai-agent-evaluation"

from langsmith.run_helpers import traceable

@traceable(run_type="chain")

def run_agent_with_tracking(user_input: str):

"""LangSmithでトラッキングされるエージェント実行"""

result = agent_executor.invoke({"input": user_input})

return result

LangSmithクライアントで実行データを照会

client = Client()

runs = client.list_runs(

project_name="ai-agent-evaluation",

run_type="chain"

)

for run in list(runs)[:5]:

print(f"Run ID: {run.id}")

print(f"ステータス: {run.status}")

print(f"実行時間: {run.end_time - run.start_time if run.end_time else 'N/A'}")

print(f"トークン使用量: {run.total_tokens}")

print("---")

10.2 エージェントパフォーマンスメトリクス

from dataclasses import dataclass, field

from typing import List, Optional

@dataclass

class AgentMetrics:

task: str

success: bool

total_time: float

num_iterations: int

tools_used: List[str]

tokens_used: int

error_message: Optional[str] = None

final_answer: Optional[str] = None

class AgentEvaluator:

"""エージェントパフォーマンス評価システム"""

def __init__(self, agent_executor):

self.agent = agent_executor

self.metrics_history: List[AgentMetrics] = []

def evaluate(self, task: str, expected_keywords: list = None) -> AgentMetrics:

start_time = time.time()

try:

result = self.agent.invoke({"input": task})

total_time = time.time() - start_time

answer = result.get("output", "")

success = True

if expected_keywords:

success = any(kw.lower() in answer.lower() for kw in expected_keywords)

metrics = AgentMetrics(

task=task,

success=success,

total_time=total_time,

num_iterations=0,

tools_used=[],

tokens_used=0,

final_answer=answer

)

except Exception as e:

metrics = AgentMetrics(

task=task,

success=False,

total_time=time.time() - start_time,

num_iterations=0,

tools_used=[],

tokens_used=0,

error_message=str(e)

)

self.metrics_history.append(metrics)

return metrics

def batch_evaluate(self, test_cases: list) -> dict:

results = []

for case in test_cases:

task = case["task"]

keywords = case.get("expected_keywords", [])

metrics = self.evaluate(task, keywords)

results.append(metrics)

successes = [r for r in results if r.success]

success_rate = len(successes) / len(results) if results else 0

avg_time = sum(r.total_time for r in results) / len(results)

return {

"total_tasks": len(results),

"success_rate": success_rate,

"avg_response_time": avg_time,

"failed_tasks": [r.task for r in results if not r.success],

"detailed_results": results

}

def generate_report(self) -> str:

if not self.metrics_history:

return "評価データなし"

total = len(self.metrics_history)

successes = sum(1 for m in self.metrics_history if m.success)

avg_time = sum(m.total_time for m in self.metrics_history) / total

report = f"""

=== エージェントパフォーマンスレポート ===

総タスク数: {total}

成功率: {successes/total*100:.1f}%

平均応答時間: {avg_time:.2f}秒

失敗したタスク:

"""

for m in self.metrics_history:

if not m.success:

report += f" - {m.task}: {m.error_message or '品質チェック失敗'}\n"

return report

test_cases = [

{

"task": "今何時ですか？",

"expected_keywords": ["2026", ":", "時", "分"]

},

{

"task": "100の平方根は？",

"expected_keywords": ["10"]

},

{

"task": "AIエージェントの主要コンポーネントを説明してください",

"expected_keywords": ["LLM", "ツール", "メモリ"]

}

]

evaluator = AgentEvaluator(agent_executor)

report = evaluator.batch_evaluate(test_cases)

print(f"成功率: {report['success_rate']*100:.1f}%")

print(f"平均応答時間: {report['avg_response_time']:.2f}秒")

まとめ

AIエージェントは単純なチャットボットを超えて**真に自律的なAIシステム**へと進化しています。

フレームワーク選択ガイド：

| ユースケース | 推奨フレームワーク |

| --------------------------- | ----------------------- |

| 迅速なプロトタイピング | LangChain |

| 複雑なワークフロー | LangGraph |

| ドキュメントQ&Aエージェント | LlamaIndex |

| マルチエージェント協調 | CrewAI |

| カスタムフレームワーク | OpenAI Function Calling |

**エージェント開発のベストプラクティス**：

1. **小さく始める** - シンプルなReActエージェントから始める

2. **明確なツール説明** - 各ツールの説明を明示的にする

3. **メモリを事前設計** - どの情報を記憶する必要があるかを計画する

4. **エラーハンドリング** - ツールの失敗とループ防止は必須

5. **すべてを監視する** - LangSmithですべての実行をトレース

6. **コストを管理** - トークン使用量を密に監視する

参考文献

- [ReActペーパー](https://arxiv.org/abs/2210.03629) - Yao et al., 2022

- [LangChainドキュメント](https://python.langchain.com/docs/)

- [LangGraphドキュメント](https://langchain-ai.github.io/langgraph/)

- [CrewAIドキュメント](https://docs.crewai.com/)

- [LlamaIndexドキュメント](https://docs.llamaindex.ai/)