AIエージェント完全ガイド: LangChain・LangGraph・CrewAIで自律エージェント構築

ユーザー目標入力
      ↓
  [計画]
  目標をサブタスクに分解
      ↓
  [行動選択]
  次の行動を決定（どのツールを使うか？）
      ↓
  [ツール実行]
  選択したツールを実行
      ↓
  [結果観察]
  ツールの出力を確認
      ↓
  [目標達成？] → Yes → 最終回答を生成
       ↓ No
  行動選択に戻る

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%）上昇しています。

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"])

from openai import OpenAI
import json

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)

from langchain.tools import tool
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
import subprocess
import sqlite3

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

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)

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"])

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)

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"]])

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
import datetime

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)"""
    import math
    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("今話した会社の時価総額はどれくらいですか？"))  # メモリを使用

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
import operator

# 状態スキーマの定義
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)

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)

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from typing import TypedDict, List
import json

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"])

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)

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)

Crew
├── Agents - それぞれのロールとゴールを持つ
│   ├── Role: "シニアリサーチャー", "コンテンツライター"
│   ├── Goal: エージェントが達成しようとすること
│   ├── Backstory: 個性/専門知識
│   └── Tools: 利用可能なツール
└── Tasks
    ├── Description: 何をすべきか
    ├── Expected Output: 期待される結果
    └── Agent: 担当エージェント

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)

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)

# マネージャーが作業を委任する階層的構造
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
)

from langchain.memory import (
    ConversationBufferMemory,
    ConversationSummaryBufferMemory,
    ConversationEntityMemory,
)
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
import datetime

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

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"])

import docker
import tempfile
import os

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 = """
import pandas as pd
import json

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"])

import os
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("---")

import time
import datetime
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}秒")