Strands Agents 1.0 소개: 프로덕션에 간편하게 바로 적용 가능한 다중 에이전트 오케스트레이션

from strands import Agent, tool 
from strands_tools import calculator, file_write, python_repl, journal 

@tool 

def web_search(query: str) -> str: 
    return "Dummy web search results here!" 

# 특화된 에이전트 생성 

research_analyst_agent = Agent( 
    system_prompt="You are a research specialist who gathers and analyzes information about local startup markets", 
    tools=[web_search, calculator, file_write, python_repl] 
) 

travel_advisor_agent = Agent( 
    system_prompt="You are a travel expert who helps with trip planning and destination advice", 
    tools=[web_search, journal] 
) 

# 에이전트를 도구로 변환 

@tool 

def research_analyst(query: str) -> str: 
    response = research_analyst_agent(query) 
    return str(response) 

@tool 

def travel_advisor(query: str) -> str: 
    response = travel_advisor_agent(query) 
    return str(response) 

# 오케스트레이터는 자연스럽게 전문가에게 위임

executive_assistant = Agent(  
    tools=[research_analyst, travel_advisor]  
) 

result = executive_assistant("I have a business meeting in Portland next week. Suggest a nice place to stay near the local startup scene, and suggest a few startups to visit") 

from strands import Agent
from strands_tools import handoff_to_user

SYSTEM_PROMPT="""
Answer the user's support query. Ask them questions with the handoff_to_user tool when you need more information
"""

# 에이전트의 도구 목록에 handoff_to_user 도구 포함
agent = Agent(
    system_prompt=SYSTEM_PROMPT,
    tools=[handoff_to_user]
)

# 에이전트는 고객을 위한 질문이 포함된 handoff_to_user 도구를 호출
agent("I have a question about my order.")

from strands import Agent

SYSTEM_PROMPT="""
Answer the user's support query. Ask them questions when you need more information
"""

agent = Agent(
    system_prompt=SYSTEM_PROMPT,
)

# 에이전트는 텍스트로 스트리밍하여 질문
agent("I have a question about my order.")

import logging  
from strands import Agent 
from strands.multiagent import Swarm 
from strands_tools import memory, calculator, file_write 

# Strands 디버그 로그 수준을 활성화하고 stderr에 출력 

logging.getLogger("strands.multiagent").setLevel(logging.DEBUG) 
logging.basicConfig( 
    format="%(levelname)s | %(name)s | %(message)s", 
    handlers=[logging.StreamHandler()] 
) 

researcher = Agent( 
    name="researcher", 
    system_prompt="You research topics thoroughly using your memory and built-in knowledge", 
    tools=[memory] 
) 

analyst = Agent( 
    name="analyst", 
    system_prompt="You analyze data and create insights", 
    tools=[calculator, memory] 
) 

writer = Agent( 
    name="writer", 
    system_prompt="You write comprehensive reports based on research and analysis", 
    tools=[file_write, memory] 
) 

# Swarm은 자동으로 에이전트를 조율 

market_research_team = Swarm([researcher, analyst, writer]) 
result = market_research_team( 
    "What is the history of AI since 1950? Create a comprehensive report" 
) 

from strands import Agent  
from strands.multiagent import GraphBuilder  

analyzer_agent = Agent(  
    name="analyzer",  
    system_prompt="Analyze customer requests and categorize them",  
    tools=[text_classifier, sentiment_analyzer]  
) 

normal_processor = Agent(  
    name="normal_processor",  
    system_prompt="Handle routine requests automatically",  
    tools=[knowledge_base, auto_responder]  
)  

 critical_processor = Agent(  
    name="critical_processor",  
    system_prompt="Handle critical requests quickly",  
    tools=[knowledge_base, escalate_to_support_agent]  
)  

# 결정론적 워크플로우 구축 

builder = GraphBuilder()  
builder.add_node(analyzer_agent, "analyze")  
builder.add_node(normal_processor, "normal_processor")  
builder.add_node(critical_processor, "critical_processor") 

# 조건부 라우팅 정의

def is_approved(state): 
    return True 
def is_critical(state): 
    return False 

builder.add_edge("analyze", "normal_processor", condition=is_approved)  
builder.add_edge("analyze", "critical_processor", condition=is_critical)  
builder.set_entry_point("analyze")  
customer_support_graph = builder.build() 

# 사용자 입력으로 그래프 실행 

results = customer_support_graph("I need help with my order!") 

from strands import Agent, tool 
from strands.multiagent import GraphBuilder, Swarm 
from strands_tools import memory, calculator, python_repl, file_write  

# 단일 에이전트로 시작

agent = Agent(tools=[memory]) 

# 리드 오케스트레이터 에이전트가 상담할 수 있는 전문 에이전트 생성
data_analyst = Agent(name="analyst", tools=[calculator, python_repl]) 

@tool 

def data_analyst_tool(query: str) -> str: 
    return str(data_analyst(query)) 

analyst_orchestrator = Agent(tools=[memory, data_analyst_tool]) # 도구로서의 에이전트(Agents-as-tools) 

# 패턴을 함께 구성 - Swarm을 사용하는 그래프

researcher = Agent(name="researcher", tools=[memory]) 
writer = Agent(name="writer", tools=[file_write]) 

research_swarm = Swarm([researcher, analyst_orchestrator, writer]) 

review_agent = Agent(system_prompt="Review the research quality and suggest improvements") 

builder = GraphBuilder() 

builder.add_node(research_swarm, "research") # 그래프 노드로서의 Swarm(Swarm as graph node) 

builder.add_node(review_agent, "review") 

builder.add_edge("research", "review") 

graph = builder.build() 

# 패턴은 자연스럽게 중첩됨 - 그래프 내 스웜, 도구로서의 에이전트 등  

result = graph("How has green energy evolved over the last few years?") 

from strands import Agent 
from strands.multiagent.a2a import A2AServer 
from strands_tools.a2a_client import A2AClientToolProvider 

# A2A 프로토콜을 통해 에이전트 제공 

local_agent = Agent(name="analyzer", tools=[web_search, data_analysis]) 

a2a_agent = A2AServer(agent=local_agent, port=9000) 

a2a_agent.serve() # AgentCard는 http://localhost:9000/.well-known/agent.json에서 사용 가능

# 원격 A2A 에이전트 사용

partner_agent_url = "https://partner.com" 

cloud_agent_url = "https://cloud.ai" 

# 원격 A2A 활성화 에이전트에 연결

a2a_tool_provider = A2AClientToolProvider(known_agent_urls=[partner_agent_url, cloud_agent_url]) 

# 원격 에이전트 오케스트레이션

orchestrator = Agent(tools=[a2a_tool_provider.tools]) 

from strands import Agent 
from strands.session.file_session_manager import FileSessionManager 

# 파일 기반 스토리지로 세션 관리자 생성 

Session_manager = FileSessionManager(session_id=”customer_support”, base_dir="./agent_sessions") 

# 에이전트는 모든 대화를 자동으로 유지 

agent = Agent( 
    id="support_bot_1", 
    session_manager=session_manager, 
    tools=[knowledge_base, ticket_system] 
) 

# 대화가 진행됨에 따라 메시지가 자동으로 저장됨

agent("Help me reset my password") 

agent("I can't access my email") 

# 나중에, 재시작 후에도 전체 대화 복원

restored_session_manager = FileSessionManager(session_id=”customer_support”, base_dir="./agent_sessions") 

restored_agent = Agent( 
    id="support_bot_1",  
    session_manager=restored_session_manager, 
    tools=[knowledge_base, ticket_system] 
) 

# 에이전트는 전체 대화 기록을 기억하고 원활하게 계속할 수 있음

import asyncio 
from fastapi import FastAPI 
from fastapi.responses import StreamingResponse 
from strands import Agent 
from strands_tools import calculator 

app = FastAPI() 

@app.post("/chat") 

async def chat_endpoint(message: str): 
    async def stream_response(): 
        agent = Agent(tools=[web_search, calculator]) 
        # 실시간으로 에이전트 응답 스트리밍 
        async for event in agent.stream_async(message): 
            if "data" in event: 
                yield f"data: {event['data']}\n\n" 
            elif "current_tool_use" in event: 
                yield f"event: tool\ndata: Using {event['current_tool_use']['name']}\n\n" 
    return StreamingResponse(stream_response(), media_type="text/event-stream") 

# 동시 실행 에이전트 평가 

async def evaluate_models_concurrently(prompt: str): 
    async def stream(agent: Agent): 
        print(f"STARTING: {agent.name}") 
        async for event in agent.stream_async(prompt): 
            # 이벤트 처리 
        print(f"ENDING: {agent.name}") 
        return event[“result”]  # 마지막 이벤트는 에이전트 결과 
    agents = [ 
        Agent(name="claude", model="us.anthropic.claude-3-7-sonnet-20250219-v1:0”), 
        Agent(name="deepseek”, model="us.deepseek.r1-v1:0”), 
        Agent(name="nova", model="us.amazon.nova-pro-v1:0") 
    ] 

    # 모든 에이전트 동시에 실행
    responses = await asyncio.gather(*[stream(agent) for agent in agents]) 

    return responses 

from strands import Agent
from strands.models import BedrockModel
from strands.models.openai import OpenAIModel
from strands.models.anthropic import AnthropicModel

# 다양한 모델 제공자 구성
bedrock_model = BedrockModel(
    model_id="us.amazon.nova-pro-v1:0",
    temperature=0.3,
    top_p=0.8,
    region_name="us-west-2"
)

openai_model = OpenAIModel(
    client_args={
        "api_key": "your-api-key",
    },
    model_id="gpt-4o",
    params={
        "max_tokens": 1000,
        "temperature": 0.7,
    }
)

anthropic_model = AnthropicModel(
    client_args={
        "api_key": "your-api-key",
    },
    max_tokens=1028,
    model_id="claude-3-7-sonnet-20250219",
    params={
        "temperature": 0.5,
    }
)

# 모델 교체 또는 동일한 시스템에서 다른 에이전트에 대해 다른 모델 사용
researcher = Agent(
    name="researcher",
    model=anthropic_model,
    tools=[web_search]
)

writer = Agent(
    name="writer", 
    model=openai_model,
    tools=[document_formatter]
)

analyzer = Agent(
    name="analyzer",
    model=bedrock_model,
    tools=[data_processor]
)