実に難解なタイトルになりましたね。
Strands Agentsには、マルチエージェントを実現する仕組みがあり、Graphマルチエージェントが構築できます。
LangGraphのGraph APIみたいなやつですね。
各ノードの処理が終わったら次のノードに遷移して処理が継続されます。
ドキュメントから引用すると、Agentをノードとして指定するように読み取れます。
...
# Create specialized agents
researcher = Agent(name="researcher", system_prompt="You are a research specialist...")
analyst = Agent(name="analyst", system_prompt="You are a data analysis specialist...")
fact_checker = Agent(name="fact_checker", system_prompt="You are a fact checking specialist...")
report_writer = Agent(name="report_writer", system_prompt="You are a report writing specialist...")
# Build the graph
builder = GraphBuilder()
# Add nodes
builder.add_node(researcher, "research")
builder.add_node(analyst, "analysis")
builder.add_node(fact_checker, "fact_check")
builder.add_node(report_writer, "report")
...
ただ、ノードの処理をエージェントじゃなくてもうちょっとカスタム処理を組み込みたいことってありますよね？はい、あります。
ということで、ノードの処理をAgentじゃない方法で実現する方法の紹介です。
MultiAgentBaseを継承したクラスを作成する
ほぼ種明かしなのですが、基底クラスMultiAgentBaseを継承したクラスを作成します。
MultiAgentBaseはこんな感じなので、継承したうえでinvoke_asyncと__call__を実装してやればOKです。
MultiAgentBase
class MultiAgentBase(ABC):
"""Base class for multi-agent helpers.
This class integrates with existing Strands Agent instances and provides
multi-agent orchestration capabilities.
"""
@abstractmethod
async def invoke_async(self, task: str | list[ContentBlock], **kwargs: Any) -> MultiAgentResult:
"""Invoke asynchronously."""
raise NotImplementedError("invoke_async not implemented")
@abstractmethod
def __call__(self, task: str | list[ContentBlock], **kwargs: Any) -> MultiAgentResult:
"""Invoke synchronously."""
raise NotImplementedError("__call__ not implemented")
試しに、Tavily検索だけをして生成AIを呼ばないSearchAgentクラスを作ってみました。
MultiAgentResultをreturnする必要があり、中身にNodeResultやらAgentResultやらが出てきて複雑怪奇ですが、AgentResultのmessageとして、検索結果をセットしてます。
class SearchAgent(MultiAgentBase):
def __init__(self, name: str):
super().__init__()
self.name = name
def __call__(self, task, **kwargs: Any) -> MultiAgentResult:
if isinstance(task, str):
query = task
elif isinstance(task, list):
query = task[-1]["text"]
tavily_serch = TavilyClient()
search_result = tavily_serch.search(query=query)
return MultiAgentResult(
status=Status.COMPLETED,
results={
self.name: NodeResult(
result=AgentResult(
stop_reason="end_turn",
message=Message(
role="assistant",
content=[{"text": json.dumps(search_result)}],
),
metrics=EventLoopMetrics(),
state=None,
)
)
},
)
async def invoke_async(self, task, **kwargs):
return self.__call__(task, **kwargs)
すると、あたかもAgentかのように呼び出せます。
agent = SearchAgent("search_agent")
result = agent("Amazon Bedrock")
print(result)
ということで、Graphのノードとして使ってみます。
report_writer = Agent(
name="report_writer",
system_prompt="あなたはレポートを作成するスペシャリストです。",
callback_handler=None,
)
builder = GraphBuilder()
builder.add_node(search_agent, "search_agent")
builder.add_node(report_writer, "report_writer")
builder.add_edge("search_agent", "report_writer")
graph = builder.build()
result = graph("Strands Agentsの特徴は？")
print(result.results["report_writer"].result.message["content"][-1]["text"])
# Strands Agentsの特徴
## 概要
Strands Agentsは、従来のAIエージェント開発フレームワークとは異なる革新的な設計思想を持つオープンソースのSDKです。
## 主な特徴
### 1. 独自の設計哲学
- **単なる「LLM呼び出しツール」にとどまらない設計思想**
- エージェントに「内面世界」を持たせるアプローチを採用
- 多くのフレームワークがプロンプト生成やシナリオ定義に留まる中、より高次元の機能を提供
### 2. 自律的な処理能力
- **開発者が複雑なワークフローを細かく定義する必要がない**
- モデル自身が以下の処理を自律的に実行：
- 計画立案
- 思考連鎖
- ツール呼び出し
- 自己反省
### 3. 簡潔で柔軟な設計
- **軽量かつ邪魔にならない設計**
- シンプルに動作し、完全にカスタマイズ可能
- 従来のフレームワークと比べ、簡易な記述でエージェントを定義できる
### 4. ツール使用前提の設計
- **生成AIがツールを使用することを前提とした設計**
- ツールや子エージェントによる機能拡張が容易
- より実用的なAIエージェントの構築が可能
## 他フレームワークとの違い
従来のフレームワークがプロンプト生成やシナリオ定義に焦点を当てる一方、Strands Agentsはエージェントの自律性と内面的な思考プロセスを重視し、より人間らしい判断と行動を可能にする革新的なアプローチを提供しています。
## 注意点
- 処理時間やトークン数の増加
- システム組み込み時の権限管理などの考慮が必要
あってるのかな。。？
実は、ノードとしてAgentしか設定できないんだと思って、機能リクエストを上げたところ、「もうできるぜ」と教えてくれました。
オープンソースって素晴らしい
import json
from typing import Any
from strands.agent import Agent, AgentResult
from strands.multiagent import GraphBuilder, MultiAgentBase, MultiAgentResult
from strands.multiagent.graph import NodeResult, Status
from strands.telemetry import EventLoopMetrics
from strands.types.content import Message
from tavily import TavilyClient
class SearchAgent(MultiAgentBase):
def __init__(self, name: str):
super().__init__()
self.name = name
def __call__(self, task, **kwargs: Any) -> MultiAgentResult:
if isinstance(task, str):
query = task
elif isinstance(task, list):
query = task[-1]["text"]
tavily_serch = TavilyClient()
search_result = tavily_serch.search(query=query)
return MultiAgentResult(
status=Status.COMPLETED,
results={
self.name: NodeResult(
result=AgentResult(
stop_reason="end_turn",
message=Message(
role="assistant",
content=[{"text": json.dumps(search_result)}],
),
metrics=EventLoopMetrics(),
state=None,
)
)
},
)
async def invoke_async(self, task, **kwargs):
return self.__call__(task, **kwargs)
search_agent = SearchAgent("search_agent")
result = search_agent("Amazon Bedrock")
# print(result)
report_writer = Agent(
name="report_writer",
system_prompt="あなたはレポートを作成するスペシャリストです。",
callback_handler=None,
)
builder = GraphBuilder()
builder.add_node(search_agent, "search_agent")
builder.add_node(report_writer, "report_writer")
builder.add_edge("search_agent", "report_writer")
graph = builder.build()
result = graph("Strands Agentsの特徴は？")
print(result.results["report_writer"].result.message["content"][-1]["text"])