AI Workflow（AIワークフロー）

AI Workflow（AIワークフロー）は、複数のAIモデル呼び出し、ツール使用、データ処理ステップを順序立てて組み合わせ、1本の自動化パイプラインにするシステムです。

LLMを一度呼び出すだけでも質問には答えられますが、現実のタスクでは多くの場合、ウェブ検索 → 情報抽出 → 分析 → レポート作成 → メール送信が必要です。AI Workflowはこれらのステップをつなぎ合わせ、AIが一文に答えるだけでなく、完全なタスクを自動で完了できるようにします。

直感的なアナロジー

組み立てラインを想像してください：

パターン	アナロジー	説明
単一LLM呼び出し	職人のようなもの	鉄を与えると、剣を返してくれる
AI Workflow	組み立てライン全体のようなもの	原材料が入ると、自動的に精錬 → 鍛造 → 焼入れ → 研磨 → 梱包を経て、完成品が出てくる

各ステップはAIモデル、コード関数、外部API、または人間による審査ノードにすることができます。

質問応答からタスク実行への進化

なぜAI Workflowが必要か

単一呼び出しの限界

1回のLLM呼び出しで完了できることは非常に限られています：

コンテキストウィンドウが有限：一度に本を一冊読み込めない
リアルタイム情報にアクセスできない：トレーニングデータに期限がある
操作を実行できない：実際にメールを送信したり、コードを書いて実行したりできない
自己検証ができない：生成エラーに気づいて修正できない
複雑なタスクはエラーが発生しやすい：一度に多くのことを行うと品質が低下する

AI Workflowが解決する5大核心問題

核心となる構成要素

完全なAI Workflowは以下の核心要素で構成されます。

各要素の詳細解説

LLM（大規模言語モデル）- Workflowの「頭脳」であり、推論、生成、意思決定を担当。よく使われるもの：GPT-4o、Claude 3.5、Gemini 1.5、ローカルLlama 3。

ツール（Tools）- AIが外部世界とやり取りするためのインターフェース。以下を含む：検索エンジン（Tavily、Serper、Bing）、コード実行環境（Python REPL、サンドボックス環境）、データベースクエリ（SQL、ベクトルDB）、外部API（天気、株価、メール、カレンダー）、ファイル操作（読み書き、PDF/Excel解析）。

記憶（Memory）短期記憶は現在のセッションの対話履歴（プロンプト内に保持）を保存し、長期記憶はベクトルデータベース + RAG によりセッションを跨ぐ永続保存を実現し、作業記憶はタスク実行中の中間状態を維持します。

状態（State）- タスクの各ステップ間で情報を伝達する媒体。リレーのバトンのようなもので、各ステップは前のステップの結果を読み取り、新しい結果を書き込むことができる。

ルーティング/条件（Router）- 前のステップの出力に基づいて次のステップの方向を動的に決定し、分岐・ループ・ジャンプなどの複雑なフロー制御を実現する。

人的介入（Human in the Loop）- 重要なポイントで一時停止し、人の確認を待つ。データ削除、メール送信、財務操作などの高リスク操作に適している。

6つの一般的なWorkflowパターン

以下は、AI Workflowで最もよく使われる6つの設計パターンです。単純なものから複雑なものまで、さまざまなシーンに適用できます。

パターン1：シーケンシャルチェーン（Sequential Chain）

最も基本的なパターンで、ステップA → B → Cが直線的に実行され、前のステップの出力が次のステップの入力になります。

観点	説明
適用シーン	ドキュメント処理パイプライン、コンテンツ生成、データ変換
利点	シンプルで予測可能
欠点	柔軟性に欠け、内容に応じて動的に調整できない

パターン2：条件付きルーティング（Conditional Routing）

特定のステップの出力内容に基づいて、異なる後続パスを動的に選択します。

観点	説明
適用シーン	スマートカスタマーサポート、多機能アシスタント、問題分類処理
利点	柔軟で、リソース利用率が高い
欠点	ルーティングロジックの慎重な設計が必要で、分類ミスが全体のフローに影響する

パターン3：並列実行（Parallel Execution）

複数のサブタスクを同時に実行し、最後に結果を集約します。

観点	説明
適用シーン	多次元分析、バッチ処理、独立したサブタスク
利点	速度が大幅に向上
欠点	並行制御と結果マージのロジックを処理する必要がある

パターン4：ReActループ（Reason + Act）

AIはまず推論して何をするかを決定し、次に行動してツールを呼び出し、結果に基づいて推論を続け、タスクが完了するまでループします。これはAI Agentの核心的なパターンです。

AI Agent、複雑なタスク実行、オープンな問題解決	利点
動的で柔軟、未知の状況に対応可能	欠点
ループ回数が制御できず、無限ループを防ぐために最大ステップ数を設定する必要がある	パターン5：Plan & Execute（計画後の実行）
まずLLMに完全な計画を立てさせ、その計画に従って段階的に実行します。ReActとの違いは「先に考えてから行動する」ことです。	パターン5：Plan & Execute（先に計画してから実行）

パターン5：Plan & Execute（計画後の実行）

競合を調査して

パターン6：マルチエージェント協調（Multi-Agent）

ユーザーのタスクを受け取り、専門Agentに割り当て、結果を集約します

複雑なソフトウェア開発、研究支援、企業自動化	利点
専門特化で品質が高く、拡張が容易	欠点
システムの複雑性が高く、Agent間の通信には慎重な設計が必要	主流フレームワークとツールの比較
以下は、現在最も主流なAI Workflowフレームワークの全体比較です。自分の状況に合わせて選択するのに役立ちます。	AI Workflow 主流フレームワーク全体比較

主流フレームワークとツールの比較

位置づけ

フレームワーク選定の決定木

あなたの状況に応じて、以下の決定木に従って適切なフレームワークを選んでください：

你的情况是什么？
│
├─── 没有编程基础，想用可视化工具搭建
│    ├─── 主要是 AI 应用（问答、生成）→ Dify（首选）
│    └─── 需要连接 Slack/邮件等 SaaS 系统 → n8n
│
├─── 有 Python 基础，代码优先
│    ├─── 做知识库 / RAG 系统 → LlamaIndex
│    ├─── 做多 Agent 协作，想快速上手 → CrewAI
│    ├─── 需要复杂有状态流程控制 → LangGraph
│    └─── 通用场景，想要最大生态 → LangChain
│
└─── 已有明确场景，生产级要求
     ├─── 高并发、精细控制 → LangGraph + LangSmith
     └─── 企业部署、私有化 → Dify 自托管

クイックスタート：Pythonコード例

以下の例では、最も単純なシーケンシャルチェーンから複雑なマルチエージェント連携まで、AI Workflowの実装方法を段階的に示します。

LangChainシーケンシャルチェーン

最も基本的なWorkflowパターンで、パイプ演算子 | を使って複数のLLM呼び出しステップを連結します。

依存パッケージをインストール：

pip install langchain langchain-openai

実例

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini", api_key="your-api-key")

# ─── 3つのステップを定義 ────────────────────────────────────────────
# ステップ1：記事を中国語に翻訳
translate_prompt = ChatPromptTemplate.from_template(
"下記の英文記事を中国語に翻訳し、元の意味を保持してください：\n\n{article}"
)

# ステップ2：要約を抽出
summarize_prompt = ChatPromptTemplate.from_template(
"次の記事を3つの要点にまとめてください。各要点は1文で：\n\n{translated}"
)

# ステップ3：タイトルを生成
title_prompt = ChatPromptTemplate.from_template(
"以下の要約に基づいて、魅力的な中国語タイトルを生成してください（15文字以内）：\n\n{summary}"
)

parser = StrOutputParser()

# ─── | 演算子でパイプラインに連結 ─────────────────────────────────
chain = (
{"translated": translate_prompt | llm | parser}
| {"summary": summarize_prompt | llm | parser,
"translated": lambda x: x["translated"]}
| title_prompt | llm | parser
)

# ─── 実行 ────────────────────────────────────────────────────
article = """
Artificial intelligence is transforming how we work and live.
From automating repetitive tasks to assisting in creative work,
AI tools are becoming indispensable in modern workflows...
"""

result = chain.invoke({"article": article})
print(result)

AI 正在重塑现代工作流：从自动化到创意辅助

ツール呼び出しエージェント（ReActモード）

ReAct は AI Agent の中心的なパターンであり、AI が思考と行動の間をループしながらタスク完了まで繰り返します。

実例

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain import hub
import requests, datetime

# ─── ツールを定義 ────────────────────────────────────────────────
@tool
def get_weather(city: str) -> str:
"""指定された都市の現在の天気情報を取得します"""
# 実際のプロジェクトでは実際の天気APIに置き換える
mock_data = {
"北京": "晴れ、気温22°C、微風",
"上海": "曇り、気温26°C、湿度75%",
"広州": "小雨、気温30°C、傘を持参することをおすすめ",
}
return mock_data.get(city, f"{city} の天気データはありません")

@tool
def search_web(query: str) -> str:
"""ウェブ上で情報を検索し、関連コンテンツの要約を返します"""
# 実際のプロジェクトでは Tavily / Serper API を接続する
return f"検索 '{query}' の結果：これはシミュレーションされた検索結果です..."

@tool
def calculate(expression: str) -> str:
"""数式を計算します。例：'2 + 3 * 4'"""
try:
result = eval(expression, {"__builtins__": {}}, {})
return str(result)
except Exception as e:
return f"計算エラー：{e}"

@tool
def get_date() -> str:
"""今日の日付を取得します"""
return datetime.date.today().strftime("%Y年%m月%d日")

# ─── Agent を作成 ───────────────────────────────────────────────
tools = [get_weather, search_web, calculate, get_date]
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# LangChain Hub の標準 ReAct プロンプトを使用
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True, # 各ステップの思考プロセスを出力してデバッグしやすくする
max_iterations=8, # 無限ループを防ぐ
handle_parsing_errors=True
)

# ─── 実行 ────────────────────────────────────────────────────
result = agent_executor.invoke({
"input": "今日は何日ですか？北京の天気はどうですか？もし外出して歩行5kmで約300カロリーを消費する場合、"
"同じ距離を走る場合の消費カロリーは歩行の約1.6倍です。走った場合の消費カロリーを計算してください。"
})

print(result["output"])
# Agent が自動判断：最初に get_date → get_weather("北京") → calculate("300*1.6") を呼び出す
# 最後にすべての情報を統合して完全な回答を返す

LangGraphステートフルWorkflow

LangGraph はグラフ方式で複雑なフローを定義し、各ノードは関数、エッジが遷移ロジックを定義します。

実例

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
import operator

# ─── 状態構造を定義（ノード間で受け渡すデータ）────────────────────────
class ResearchState(TypedDict):
topic: str # 研究テーマ
research_notes: str # 研究メモ
draft: str # 下書き
review_feedback: str # レビュー意見
final_report: str # 最終レポート
revision_count: Annotated[int, operator.add] # 修正回数（累積）

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

# ─── ノード関数を定義 ─────────────────────────────────────────────
def research_node(state: ResearchState) -> dict:
"""ノード1：調査フェーズ"""
response = llm.invoke(
f"次のテーマについて簡単に調査し、重要なポイントを5つ挙げてください：{state['topic']}"
)
return {"research_notes": response.content}

def write_node(state: ResearchState) -> dict:
"""ノード2：下書き作成"""
prompt = f"""
テーマ：{state['topic']}
調査メモ：{state['research_notes']}
{'前回のレビュー意見：' + state.get('review_feedback', '') if state.get('review_feedback') else ''}

上記の内容に基づいて、300字の分析レポートの下書きを作成してください。
"""
response = llm.invoke(prompt)
return {"draft": response.content, "revision_count": 1}

def review_node(state: ResearchState) -> dict:
"""ノード3：下書きレビュー"""
response = llm.invoke(
f"次のレポートをレビューし、品質が基準を満たしていれば 'APPROVED' と返答し、そうでなければ具体的な修正意見を出してください：\n\n{state['draft']}"
)
return {"review_feedback": response.content}

def finalize_node(state: ResearchState) -> dict:
"""ノード4：最終確定"""
return {"final_report": state["draft"]}

# ─── ルーティング関数：レビュー後の経路を決定 ─────────────────────────────
def should_revise(state: ResearchState) -> str:
if "APPROVED" in state["review_feedback"]:
return "finalize" # → 最終確定
elif state["revision_count"] >= 3:
return "finalize" # → 修正回数が3回を超えたら強制終了
else:
return "revise" # → 作成ノードに戻って修正

# ─── ワークフローグラフを構築 ─────────────────────────────────────────────
workflow = StateGraph(ResearchState)

# ノードを追加
workflow.add_node("research", research_node)
workflow.add_node("write", write_node)
workflow.add_node("review", review_node)
workflow.add_node("finalize", finalize_node)

# エントリポイントを設定
workflow.set_entry_point("research")

# エッジを追加（遷移ロジックを定義）
workflow.add_edge("research", "write") # 調査 → 作成
workflow.add_edge("write", "review") # 作成 → レビュー

# 条件付きエッジ：レビュー後に結果に応じて経路を選択
workflow.add_conditional_edges(
"review",
should_revise,
{
"revise": "write", # 修正が必要 → 作成に戻る
"finalize": "finalize" # 承認 → 最終確定
}
)

workflow.add_edge("finalize", END)

# コンパイルして実行
app = workflow.compile()

result = app.invoke({"topic": "生成AIがソフトウェア開発業界に与える影響", "revision_count": 0})
print(result["final_report"])

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

CrewAI は役割とタスクを定義することで、複数のAgentがチームのように協力できるようにします。

実例

from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

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

# ─── Agent（役割）を定義 ────────────────────────────────────────
researcher = Agent(
role="市場リサーチャー",
goal="対象テーマに関する包括的で正確な市場情報を収集・整理する",
backstory="あなたはベテランの市場アナリストであり、膨大な情報から重要な洞察を抽出するのが得意です。",
llm=llm,
verbose=True
)

analyst = Agent(
role="データアナリスト",
goal="リサーチャーが提供した情報に基づき、深い分析を行い、価値ある結論を導き出す。",
backstory="あなたはデータで語るのが得意で、情報の背後に隠れたトレンドやチャンスを見つけられます。",
llm=llm,
verbose=True
)

writer = Agent(
role="レポート作成エキスパート",
goal="分析結論を明確で専門的かつ説得力のあるレポートにまとめる。",
backstory="あなたは豊富なビジネスライティング経験を持ち、複雑な分析を分かりやすくできます。",
llm=llm,
verbose=True
)

# ─── Task（タスク）の定義 ────────────────────────────────────────
research_task = Task(
description="中国の新エネルギー自動車市場の現状を調査する。主要ブランド、市場シェア、成長トレンドを含む。",
expected_output="5つの重要データポイントを含む市場調査レポート。数字と具体的な事実を含む。",
agent=researcher
)

analysis_task = Task(
description="調査レポートに基づき、今後3年間の機会とリスクを分析し、投資格付けの提案を行う。",
expected_output="SWOT分析表 + 投資格付け（強く推奨/推奨/中立/慎重）+ 理由",
agent=analyst,
context=[research_task] # 調査タスクの出力に依存
)

writing_task = Task(
description="調査と分析を500字の専門的な投資ブリーフにまとめ、形式を明確にする。",
expected_output="エグゼクティブサマリー、市場現状、機会とリスク、投資提案の4つの部分を含むブリーフ。",
agent=writer,
context=[research_task, analysis_task]
)

# ─── チームを編成して実行 ───────────────────────────────────────────
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, writing_task],
process=Process.sequential, # 順次実行（hierarchical に変更も可）
verbose=True
)

result = crew.kickoff()
print(result)

Human-in-the-Loop（人間による審査ノード）

LangGraph は重要なステップで一時停止し、人間の確認を待ってから実行を継続できます。

実例

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict

class EmailState(TypedDict):
recipient: str
content: str
approved: bool

def draft_email(state: EmailState) -> dict:
"""メールの下書き"""
content = f"{state['recipient']} 様、\n\nこれは AI が作成したメール内容です...\n\n敬具"
return {"content": content}

def send_email(state: EmailState) -> dict:
"""メール送信（リスクの高い操作のため、人間のレビュー後に実行）"""
print(f"{state['recipient']} にメールを送信しました")
return {}

# ルーティング：人間のレビュー結果に基づいて送信するかどうかを決定
def check_approval(state: EmailState) -> str:
return "send" if state.get("approved") else END

workflow = StateGraph(EmailState)
workflow.add_node("draft", draft_email)
workflow.add_node("send", send_email)
workflow.set_entry_point("draft")

# draft 完了後に一時停止し、人間のレビューを待つ（interrupt_after）
workflow.add_conditional_edges("draft", check_approval, {"send": "send", END: END})
workflow.add_edge("send", END)

# MemorySaver を使用して中断と再開をサポート
memory = MemorySaver()
app = workflow.compile(
checkpointer=memory,
interrupt_after=["draft"] # draft ノードの後に一時停止
)

config = {"configurable": {"thread_id": "email-001"}}

# 初回実行：draft まで実行した後に一時停止
state = app.invoke({"recipient": "顧客A", "approved": False}, config)
print("下書きが生成されました。レビューを待っています：")
print(state["content"])

# 人間のレビュー後、状態を更新して実行を継続
user_input = input("\n送信を承認しますか？(y/n): ")
if user_input.lower() == "y":
app.update_state(config, {"approved": True})
app.invoke(None, config) # ブレークポイントから再開
print("メールを送信しました")
else:
print("送信をキャンセルしました")

Promptエンジニアリング：Workflowにおける重要テクニック

AI Workflow では、プロンプトの品質が各ノードの出力品質に直接影響します。

実例

# 曖昧な役割定義
# system_prompt = "あなたは AI アシスタントです。この文書を分析してください"

# 明確な役割定義（推奨）
system_prompt = """
あなたはプロの財務アナリストであり、財務報告書のリスクシグナルの特定に注力しています。
あなたのタスク：以下のレポートから、負債、キャッシュフロー、収益性に関するすべての重要データポイントを抽出してください。
出力形式：JSON。risk_level（high/medium/low）と key_findings（リスト）を含む。
注意：JSON のみを出力し、説明文を追加しないでください。
"""

実例

# 非構造化テキストを渡す
# state["previous_output"] = "分析結果：この会社は良さそうに見えるが、いくつかリスクがある..."

# 構造化データを渡す（推奨）
state["analysis_result"] = {
"risk_level": "medium",
"key_findings": ["負債比率45%、業界平均38%", "キャッシュフローはプラス、Q3 は前期比12%減少"],
"recommendation": "慎重に保有"
}

実例

# prompt で JSON 出力を強制する
prompt = """
以下のテキストの感情傾向を分析してください。
以下の JSON 形式に厳密に従って出力し、他の内容を追加しないでください：
{
"sentiment": "positive" | "negative" | "neutral",
"confidence": 0.0-1.0,
"reason": "簡単な説明"
}

テキスト：{text}
"""

エラーハンドリングとリトライ

実例

from tenacity import retry, stop_after_attempt, wait_exponential
import logging

@retry(
stop=stop_after_attempt(3), # 最大3回リトライ
wait=wait_exponential(multiplier=1, min=2, max=10) # 指数バックオフ待機
)
def call_llm_with_retry(prompt: str) -> str:
try:
response = llm.invoke(prompt)
return response.content
except Exception as e:
logging.error(f"LLM 呼び出しに失敗：{e}")
raise

def safe_parse_json(text: str) -> dict:
"""LLM が出力した JSON を安全に解析する"""
import json, re
# JSON 部分を抽出（LLM は余分な説明を追加することがある）
json_match = re.search(r'\{.*\}', text, re.DOTALL)
if json_match:
try:
return json.loads(json_match.group())
except json.JSONDecodeError:
pass
return {"error": "解析に失敗しました", "raw": text}

コスト管理

実例

# ─── タスクの重要度に応じてモデルを選択 ────────────────────────────────
def get_llm(task_type: str):
if task_type == "classification": # 分類タスクは軽量モデルで十分
return ChatOpenAI(model="gpt-4o-mini")
elif task_type == "generation": # 生成タスクは中規模モデルを使用
return ChatOpenAI(model="gpt-4o-mini")
elif task_type == "reasoning": # 複雑な推論は強力なモデルを使用
return ChatOpenAI(model="gpt-4o")

# ─── 重複リクエストのキャッシュ ──────────────────────────────────────────
from langchain.globals import set_llm_cache
from langchain_community.cache import InMemoryCache
set_llm_cache(InMemoryCache()) # 同じ入力はキャッシュを直接返し、重複課金しない

# ─── テキスト分割で長すぎるのを防止 ──────────────────────────────────────
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
chunk_size=2000, # 各ブロック2000文字
chunk_overlap=200 # 200文字のオーバーラップで文脈の連続性を確保
)
chunks = splitter.split_text(long_document)

代表的な適用シナリオ

AI Workflow はすでに多くの分野で広く実用化されており、以下は最も代表的な6つのシナリオです。

ベストプラクティスとよくある落とし穴

よくある落とし穴と解決策

以下は、初心者が AI Workflow を使用する際に最も遭遇しやすい問題です。

落とし穴	現象	解決策
幻覚のカスケード	前のステップで AI が誤った出力をし、それが事実として次のステップに渡され、エラーが増幅される	重要なノードに検証ステップを追加する；数値情報はツールで取得する
無限ループ	ReAct Agent がツールを繰り返し呼び出し、いつ停止すべきか分からない	max_iterations を設定する；LLM に明確な終了条件を与える
コンテキスト爆発	ステップが増えるにつれて、LLM に渡すテキストがどんどん長くなり、ウィンドウを超える	各ステップで必要なフィールドのみを渡す；要約で履歴情報を圧縮する
ツールの乱用	Agent明明直接回答できるのに、何度もツールを呼び出してしまう	ツールの説明を最適化する。LLMにツールが不要な場合を明確に伝える
JSONパース失敗	LLMの出力形式が不安定で、プログラムがクラッシュする	safe_parse_jsonを追加する。LangChain OutputParserを使う
費用超過	token使用量を見積もらず、月末の請求が予想を超える	まずgpt-4o-miniでテストする。LangSmithで使用量を監視する
並行処理の競合	複数のAgentが同時に同じ状態を書き込み、データが混乱する	LangGraphの組み込み状態管理を使う。共有可変状態を避ける

まとめと学習パス

核心知識ポイントの振り返り

推奨学習パス

第1段階：基礎を理解する（1週間）

LLM API呼び出しを理解し、OpenAI SDKでリクエストを送信できるようにする
本書のLangChainシーケンシャルチェーンの例を実行する
Prompt Engineeringの基本原則を理解する

第2段階：ツールとAgent（2週間）

@toolデコレータでツールを定義する方法を学ぶ
ReAct Agentの例を実行し、verboseログを観察してループロジックを理解する
Difyまたはn8nを使ってビジュアルWorkflowプロトタイプを構築する

第3段階：複雑なWorkflow（3週間）

LangGraph状態グラフを学び、条件分岐付きワークフローを実装する
CrewAIでデュアルAgent協調タスクを実装する
実際のツール（Tavily検索、Python REPL）を接続する
LangSmithでWorkflowを観測・デバッグする

第4段階：本番環境対応（継続的）

エラーリトライ、構造化出力パースを実装する
コスト監視と最適化（モデルの段階的利用）
本番環境にデプロイし、監視アラートを設定する

参考リソース

その他の拡張

エージェントツール

Python エージェントの実装