AI Agent Engineerキャリアガイド：2025年最もホットなAI職種の全て

┌─────────────────────────────────────────┐
│              User Request               │
│   「先四半期の売上レポートを分析して       │
│    重要なインサイトをSlackで共有して」     │
└────────────────┬────────────────────────┘
                 │
                 v
┌─────────────────────────────────────────┐
│         1. Reasoning（推論）              │
│   「売上レポートを取得する必要がある。     │
│    DBからデータをクエリしよう。」          │
└────────────────┬────────────────────────┘
                 │
                 v
┌─────────────────────────────────────────┐
│         2. Action（行動）                │
│   Tool Call: query_database(             │
│     "SELECT ... FROM sales              │
│      WHERE quarter = 'Q4_2024'")        │
└────────────────┬────────────────────────┘
                 │
                 v
┌─────────────────────────────────────────┐
│         3. Observation（観察）            │
│   結果：総売上 $12.5M、前四半期比 +15%    │
└────────────────┬────────────────────────┘
                 │
                 v
┌─────────────────────────────────────────┐
│         4. Reasoning（再推論）            │
│   「データを分析してインサイトを抽出。     │
│    その後Slackにメッセージを送ろう。」     │
└────────────────┬────────────────────────┘
                 │
                 v
┌─────────────────────────────────────────┐
│         5. Action（行動）                │
│   Tool Call: send_slack_message(         │
│     channel="#sales",                    │
│     message="Q4売上主要インサイト...")     │
└────────────────┬────────────────────────┘
                 │
                 v
┌─────────────────────────────────────────┐
│         6. Final Response               │
│   「分析完了！Slack #salesチャンネルに     │
│    インサイトを共有しました。」           │
└─────────────────────────────────────────┘

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from typing import TypedDict, Annotated
import operator

# 状態定義
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_action: str

# LLM設定
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

# ノード定義
def agent_node(state: AgentState) -> AgentState:
    """エージェント推論ノード"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState) -> str:
    """ツール呼び出しが必要か判断"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# グラフ構成
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(tools))

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {
    "tools": "tools",
    END: END,
})
graph.add_edge("tools", "agent")

# コンパイルと実行
app = graph.compile()
result = app.invoke({
    "messages": [HumanMessage(content="東京の今日の天気を教えて")]
})

from crewai import Agent, Task, Crew, Process

# エージェント定義
researcher = Agent(
    role="Senior Research Analyst",
    goal="Discover cutting-edge AI trends",
    backstory="You are an expert AI researcher with 10 years of experience.",
    tools=[search_tool, web_scraper],
    verbose=True,
)

writer = Agent(
    role="Technical Content Writer",
    goal="Write engaging technical blog posts",
    backstory="You are a skilled technical writer who can explain complex topics simply.",
    verbose=True,
)

# タスク定義
research_task = Task(
    description="Research the latest AI agent frameworks released in 2025.",
    expected_output="A comprehensive list of frameworks with pros and cons.",
    agent=researcher,
)

writing_task = Task(
    description="Write a blog post based on the research findings.",
    expected_output="A 1000-word blog post with code examples.",
    agent=writer,
    context=[research_task],
)

# クルー実行
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()

import anthropic

client = anthropic.Anthropic()

# ツール定義
tools = [
    {
        "name": "get_weather",
        "description": "Get current weather for a city",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "City name"
                }
            },
            "required": ["city"]
        }
    }
]

# エージェンティックループ
messages = [{"role": "user", "content": "東京とソウルの天気を比較して"}]

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

    # ツール呼び出しがなければ終了
    if response.stop_reason == "end_turn":
        break

    # ツール呼び出し処理
    for block in response.content:
        if block.type == "tool_use":
            tool_result = execute_tool(block.name, block.input)

            messages.append({"role": "assistant", "content": response.content})
            messages.append({
                "role": "user",
                "content": [{
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": tool_result,
                }]
            })

print(response.content[0].text)

from swarm import Swarm, Agent

client = Swarm()

# 専門エージェント定義
def transfer_to_sales():
    """営業チームに転送"""
    return sales_agent

def transfer_to_support():
    """サポートチームに転送"""
    return support_agent

triage_agent = Agent(
    name="Triage Agent",
    instructions="顧客の意図を把握し適切なチームに転送します。",
    functions=[transfer_to_sales, transfer_to_support],
)

sales_agent = Agent(
    name="Sales Agent",
    instructions="製品情報を案内し購入をサポートします。",
    functions=[get_product_info, create_quote],
)

support_agent = Agent(
    name="Support Agent",
    instructions="技術的な問題を解決しチケットを作成します。",
    functions=[search_kb, create_ticket],
)

# 実行
response = client.run(
    agent=triage_agent,
    messages=[{"role": "user", "content": "製品の価格が知りたいです"}],
)

                    ┌──────────────┐
                    │   Router     │
                    │   Agent      │
                    └──────┬───────┘
                           │
              ┌────────────┼────────────┐
              v            v            v
        ┌──────────┐ ┌──────────┐ ┌──────────┐
        │ Code     │ │ Data     │ │ Writing  │
        │ Agent    │ │ Agent    │ │ Agent    │
        └──────────┘ └──────────┘ └──────────┘

        ┌────────────────────────────────┐
        │       Orchestrator Agent        │
        │   (タスク分解、結果統合)          │
        └───┬──────────┬──────────┬──────┘
            │          │          │
            v          v          v
     ┌──────────┐ ┌──────────┐ ┌──────────┐
     │ Research │ │ Analysis │ │ Report   │
     │ Worker   │ │ Worker   │ │ Worker   │
     └──────────┘ └──────────┘ └──────────┘

     ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
     │ Extract  │───>│ Transform│───>│ Analyze  │───>│ Report   │
     │ Agent    │    │ Agent    │    │ Agent    │    │ Agent    │
     └──────────┘    └──────────┘    └──────────┘    └──────────┘

     ┌──────────┐    ┌──────────┐
     │ Generator│───>│ Evaluator│
     │ Agent    │<───│ Agent    │
     └──────────┘    └──────────┘
         │  ^             │
         │  └─────────────┘
         │   (フィードバックループ)
         v
     ┌──────────┐
     │  Output  │
     └──────────┘

from langgraph.checkpoint.postgres import PostgresSaver

# PostgreSQLチェックポインタ
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/agents"
)

# チェックポインティング有効なエージェント
app = graph.compile(checkpointer=checkpointer)

# スレッドIDで会話を維持
config = {"configurable": {"thread_id": "user-123-session-456"}}
result = app.invoke(
    {"messages": [HumanMessage(content="最初の質問")]},
    config=config,
)

# 同じスレッドで会話を続ける
result2 = app.invoke(
    {"messages": [HumanMessage(content="前の回答に対するフォローアップ")]},
    config=config,
)

from tenacity import retry, stop_after_attempt, wait_exponential

def create_robust_agent():
    """エラー処理を強化したエージェント"""

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=30),
    )
    def call_tool_with_retry(tool_name: str, tool_input: dict):
        """ツール呼び出しリトライロジック"""
        try:
            result = execute_tool(tool_name, tool_input)
            return result
        except RateLimitError:
            raise  # リトライ
        except InvalidInputError as e:
            return f"入力エラー: {str(e)}"  # リトライしない
        except Exception as e:
            return f"予期しないエラー: {str(e)}"

    return call_tool_with_retry

class CostController:
    """エージェントコストコントローラー"""

    def __init__(self, max_budget: float = 1.0, max_steps: int = 20):
        self.max_budget = max_budget
        self.max_steps = max_steps
        self.current_cost = 0.0
        self.current_steps = 0

    def track_usage(self, model: str, input_tokens: int, output_tokens: int):
        """トークン使用量追跡"""
        token_prices = {
            "gpt-4o": {"input": 2.50 / 1_000_000, "output": 10.00 / 1_000_000},
            "claude-sonnet": {"input": 3.00 / 1_000_000, "output": 15.00 / 1_000_000},
        }
        prices = token_prices.get(model, {"input": 0.01, "output": 0.03})
        cost = (input_tokens * prices["input"]) + (output_tokens * prices["output"])
        self.current_cost += cost
        self.current_steps += 1

    def should_continue(self) -> bool:
        """実行を続行するか判断"""
        if self.current_cost >= self.max_budget:
            return False
        if self.current_steps >= self.max_steps:
            return False
        return True

from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver

# 承認が必要なツールを表示
def send_email(to: str, subject: str, body: str) -> str:
    """メールを送信します。[REQUIRES_APPROVAL]"""
    return f"Email sent to {to}"

def search_web(query: str) -> str:
    """ウェブ検索を実行します。"""
    return f"Search results for: {query}"

# interrupt_beforeで特定ノードの前に中断
agent = create_react_agent(
    llm,
    tools=[send_email, search_web],
    checkpointer=MemorySaver(),
)

config = {"configurable": {"thread_id": "approval-demo"}}
result = agent.invoke(
    {"messages": [HumanMessage(content="john@example.comにレポートを送って")]},
    config=config,
)

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│   Agent      │     │   MCP        │     │  External    │
│   (LLM)      │────>│   Client     │────>│  MCP Servers │
│              │<────│              │<────│              │
└──────────────┘     └──────────────┘     └──────┬───────┘
                                                  │
                          ┌───────────────────────┼──────────────┐
                          │                       │              │
                    ┌─────┴─────┐  ┌─────────┐  ┌┴────────┐
                    │ GitHub    │  │ Slack   │  │ DB      │
                    │ Server    │  │ Server  │  │ Server  │
                    └───────────┘  └─────────┘  └─────────┘

from langsmith import Client
from langsmith.evaluation import evaluate

client = Client()

# 評価データセット作成
dataset = client.create_dataset("agent-eval-v1")
client.create_examples(
    dataset_id=dataset.id,
    inputs=[
        {"query": "東京の現在の天気は？"},
        {"query": "AAPL株価を調べてチャートを作って"},
        {"query": "このコードのバグを見つけて修正して: ..."},
    ],
    outputs=[
        {"expected_tools": ["get_weather"], "expected_result_contains": "東京"},
        {"expected_tools": ["get_stock_price", "create_chart"]},
        {"expected_tools": ["analyze_code", "fix_code"]},
    ],
)

# 評価関数
def task_completion_evaluator(run, example):
    """タスク完了を評価"""
    expected_tools = set(example.outputs["expected_tools"])
    actual_tools = set()

    for step in run.child_runs or []:
        if step.run_type == "tool":
            actual_tools.add(step.name)

    tool_coverage = len(expected_tools & actual_tools) / len(expected_tools)
    return {"score": tool_coverage, "key": "tool_coverage"}

# 評価実行
results = evaluate(
    agent_function,
    data="agent-eval-v1",
    evaluators=[task_completion_evaluator],
)

def sanitize_tool_output(output: str) -> str:
    """ツール出力から悪意ある指示をフィルタリング"""
    dangerous_patterns = [
        "ignore previous instructions",
        "you are now",
        "system prompt",
        "disregard",
        "override",
    ]
    lower_output = output.lower()
    for pattern in dangerous_patterns:
        if pattern in lower_output:
            return "[FILTERED: Potential injection detected]"
    return output

class ActionSandbox:
    """エージェントのアクションを安全に制限"""

    def __init__(self):
        self.allowed_actions = {
            "read": True,
            "write": False,     # 承認が必要
            "delete": False,    # 承認が必要
            "execute": False,   # 承認が必要
            "network": True,
        }
        self.blocked_domains = ["*.internal.corp", "admin.*"]
        self.max_file_size = 10 * 1024 * 1024  # 10MB

    def check_action(self, action_type: str, details: dict) -> bool:
        """アクションの許可を確認"""
        if not self.allowed_actions.get(action_type, False):
            return False
        if "url" in details:
            for pattern in self.blocked_domains:
                if self._match_pattern(details["url"], pattern):
                    return False
        return True

class PermissionModel:
    """階層的権限モデル"""

    LEVELS = {
        "read_only": 1,
        "standard": 2,
        "elevated": 3,
        "admin": 4,
    }

    def __init__(self, level: str = "standard"):
        self.level = self.LEVELS[level]
        self.permissions = self._get_permissions()

    def _get_permissions(self) -> dict:
        perms = {"search": True, "read_file": True}
        if self.level >= 2:
            perms.update({"write_file": True, "api_call": True})
        if self.level >= 3:
            perms.update({"send_email": True, "create_issue": True})
        if self.level >= 4:
            perms.update({"delete": True, "admin_actions": True})
        return perms

    def can_perform(self, action: str) -> bool:
        return self.permissions.get(action, False)