AI导哈

AI代理构建工具

Strands Agent是一个开源SDK，用于构建和执行AI代理，支持计划、思维链、工具调用和反思等功能，适用于多种LLM模型和工具的连接与执行。

  • by kyopark2014
  • Python
  • 本地部署
  • 8 星标
  • 更新于
在 GitHub 查看 kyopark2014/dwfb-project

使用说明

代理AI-Strands

其中 Strands代理说明如何使用实现Agentic AI。Strands Agent是为构建和运行AI agent而设计的开源SDK。可以轻松利用agent功能，如计划（planning）、链接（chaining thoughts）、工具调用和Reflection。这将LLM模型与tool相关联，并利用模型的推理能力规划和运行工具。目前支持Amazon Bedrock、Anthropic和Meta的模型，有Accenture、Anthropic和Meta等公司参与。

这里使用的architecture如下所示。为了验证和实现Agent的基本行为，它以docker的形式安装在EC2中，使用ALB和Cloudront，用户可以使用streamlit测试其行为。Agent生成的图片或文档可以使用S3共享，您可以使用嵌入在EC2中的MCP server/client使用互联网搜索（Tavily）、RAG（knowledge base）AWS tools（use-aws）和AWS Document。

Strands agent包括： 代理循环因为有，所以选择并运行适当的工具，并通过reasoning重复执行所需的操作。

可以并行处理工具，如下所示：

agent = Agent(
    max_parallel_tools=4  
)

Strands Agent的使用方法

操作体系结构

flowchart TB
  subgraph UI["Streamlit (app.py)"]
    M[Agent]
    SKUI[Skill / Strands Tool / MCP 선택]
  end

  subgraph LLM["Amazon Bedrock"]
    BR[Bedrock Runtime]
  end

  subgraph Skills["Agent Skills (skill.py)"]
    SRC["skills/*/SKILL.md"]
    BSP[build_skill_prompt]
    GSI[get_skill_instructions]
  end

  subgraph StrandsStack["Strands Agents SDK (strands_agent.py)"]
    RSA[run_strands_agent]
    A[Agent]
    SA[stream_async]
    BM[BedrockModel]
    BT["Built-in: execute_code, bash, upload_file_to_s3"]
    ST["strands_tools: current_time, file_read, file_write"]
    MCP[MCPClient / MCPClientManager]
  end

  subgraph MCPServers["MCP Servers (mcp_config.py)"]
    T[tavily]
    R[retrieve]
    AWS[aws documentation]
    WF[web_fetch / korea_weather / trade_info]
  end

  subgraph Storage["Artifacts / S3"]
    ART[artifacts/]
    S3[(S3)]
  end

  M --> RSA
  SKUI -->|skill_list| BSP

  RSA --> A
  A --> SA
  A --> BM
  BM --> BR
  A --> BT
  A --> ST
  A --> MCP
  A --> GSI
  BSP -->|system_prompt| A
  GSI --> SRC
  MCP --> MCPServers
  BT --> ART
  BT --> S3
模式 模块 说明
日常对话 chat.general_conversation 对话历史记录+Bedrock Runtime invoke_model_with_response_stream 流媒体
抹布 chat.run_rag_with_knowledge_base 基岩知识库검색(retrieve）后通过Bedrock Runtime生成答案
代理 strands_agent.run_strands_agent Strands SDK+Strands_tools+MCP+技能
图像分析 chat.summarize_image ChatBedrock多模式（图像+文本）分析

在Streamlit中运行agent

app.py用户可以选择“RAG”、“Agent”，如下所示。“Agent”使用Strands agent作为MCP，必要时可以使用tool利用RAG等。为Streamlit的UI输入和结果response 会话状态管理为。

if prompt := st.chat_input("메시지를 입력하세요."):
    with st.chat_message("user"):  
        st.markdown(prompt)
    st.session_state.messages.append({"role": "user", "content": prompt})

    with st.status("thinking...", expanded=True, state="running") as status:
        notification_queue = NotificationQueue(container=status)
        skill_list = selected_skills if selected_skills else []

        response, image_urls = asyncio.run(strands_agent.run_strands_agent(
            query=prompt, 
            strands_tools=selected_strands_tools, 
            mcp_servers=selected_mcp_servers, 
            skill_list=skill_list,
            notification_queue=notification_queue))

Agent的运行

使用system prompt、model和tool信息创建agent，如下所示。

def create_agent(system_prompt, tools):
    if system_prompt==None:
        system_prompt = (
            "당신의 이름은 서연이고, 질문에 대해 친절하게 답변하는 사려깊은 인공지능 도우미입니다."
            "상황에 맞는 구체적인 세부 정보를 충분히 제공합니다." 
            "모르는 질문을 받으면 솔직히 모른다고 말합니다."
        )
    model = get_model()    
    agent = Agent(
        model=model,
        system_prompt=system_prompt,
        tools=tools,
        conversation_manager=conversation_manager
    )
    return agent

chat.py运行Agent并通过stream接收结果并显示，如下所示。此时，在event中只提取“data”，然后保存为full_response，然后显示为markdown，如下所示。

async def run_strands_agent(query, strands_tools, mcp_servers, containers):
    await strands_agent.initiate_agent(
        system_prompt=None, 
        strands_tools=strands_tools, 
        mcp_servers=mcp_servers
    )

    final_result = current = ""
    with strands_agent.mcp_manager.get_active_clients(mcp_servers) as _:
        agent_stream = strands_agent.agent.stream_async(query)
        
        async for event in agent_stream:
            text = ""            
            if "data" in event:
                text = event["data"]
                logger.info(f"[data] {text}")
                current += text

            elif "result" in event:
                final = event["result"]                
                message = final.message
                if message:
                    content = message.get("content", [])
                    result = content[0].get("text", "")
                    final_result = result
    return final_result

利用对话履历

如果您想使用对话内容继续对话，可以使用SlidingWindowConversationManager获取和利用以前的对话（如下图所示）的window_size。详细代码包括： chat.py请参阅。

from strands.agent.conversation_manager import SlidingWindowConversationManager

conversation_manager = SlidingWindowConversationManager(
    window_size=10,  
)

agent = Agent(
    model=model,
    system_prompt=system,
    tools=[    
        calculator, 
        current_time,
        use_aws    
    ],
    conversation_manager=conversation_manager
)

利用MCP

将stdio_mcp_client指定为MCPClient后，使用list_tools_sync提取tool信息，如下所示。MCP tool可与strands tool一起使用，如下所示。

from strands.tools.mcp import MCPClient
from strands_tools import calculator, current_time, use_aws

stdio_mcp_client = MCPClient(lambda: stdio_client(
    StdioServerParameters(command="uvx", args=["awslabs.aws-documentation-mcp-server@latest"])
))

with stdio_mcp_client as client:
    aws_documentation_tools = client.list_tools_sync()
    logger.info(f"aws_documentation_tools: {aws_documentation_tools}")

    tools=[    
        calculator, 
        current_time,
        use_aws
    ]

    tools.extend(aws_documentation_tools)

    agent = Agent(
        model=model,
        system_prompt=system,
        tools=tools,
        conversation_manager=conversation_manager
    )

此外，用于wikipedia搜索的MCP服务器示例如下：详细代码包括： mcpserver_wikipedia.py请参阅。

from mcp.server.fastmcp import FastMCP
import wikipedia
import logging
import sys

logging.basicConfig(
    level=logging.INFO,  # Default to INFO level
    format='%(filename)s:%(lineno)d | %(message)s',
    handlers=[
        logging.StreamHandler(sys.stderr)
    ]
)
logger = logging.getLogger("rag")

mcp = FastMCP(
    "Wikipedia",
    dependencies=["wikipedia"],
)

@mcp.tool()
def search(query: str):
    logger.info(f"Searching Wikipedia for: {query}")
    
    return wikipedia.search(query)

@mcp.tool()
def summary(query: str):
    return wikipedia.summary(query)

@mcp.tool()
def page(query: str):
    return wikipedia.page(query)

@mcp.tool()
def random():
    return wikipedia.random()

@mcp.tool()
def set_lang(lang: str):
    wikipedia.set_lang(lang)
    return f"Language set to {lang}"

if __name__ == "__main__":
    mcp.run()

动态绑定MCP Server

定义MCPClientManager以动态管理MCP服务器。add_client通过MCP服务器的name、command、args和env定义MCP Client。

class MCPClientManager:
    def __init__(self):
        self.clients: Dict[str, MCPClient] = {}
        
    def add_client(self, name: str, command: str, args: List[str], env: dict[str, str] = {}) -> None:
        """Add a new MCP client"""
        self.clients[name] = MCPClient(lambda: stdio_client(
            StdioServerParameters(
                command=command, args=args, env=env
            )
        ))
    
    def remove_client(self, name: str) -> None:
        """Remove an MCP client"""
        if name in self.clients:
            del self.clients[name]
    
    @contextmanager
    def get_active_clients(self, active_clients: List[str]):
        """Manage active clients context"""
        active_contexts = []
        for client_name in active_clients:
            if client_name in self.clients:
                active_contexts.append(self.clients[client_name])

        if active_contexts:
            with contextlib.ExitStack() as stack:
                for client in active_contexts:
                    stack.enter_context(client)
                yield
        else:
            yield

# Initialize MCP client manager
mcp_manager = MCPClientManager()

使用Streamlit实现的 app.py在中选择tool可获得mcp_tools。在创建agent时，从active client获取tool list并将其用作tools，如下所示。

tools = []
for mcp_tool in mcp_servers:
    with mcp_manager.get_active_clients([mcp_tool]) as _:
        client = mcp_manager.get_client(mcp_tool)
        mcp_servers_list = client.list_tools_sync()
        tools.extend(mcp_servers_list)

agent = create_agent(system_prompt, tools)
tool_list = get_tool_list(tools)

生成的agent使用mcp_manager运行，如下所示。

with mcp_manager.get_active_clients(mcp_tools) as _:
    agent_stream = agent.stream_async(question)
    
    tool_name = ""
    async for event in agent_stream:
        if "message" in event:
            message = event["message"]
            for content in message["content"]:                
                if "text" in content:
                    final_response = content["text"]

调整输出语句以符合Streamlit

按如下方式运行Agent以获取agent_stream。

with mcp_manager.get_active_clients(mcp_servers) as _:
    agent_stream = agent.stream_async(question)

为了获得用户体验，必须以stream的形式获得输出。它从agent_stream中取出event，然后从“data”中提取，并将stream结果收集到current_response中，如下所示：

async for event in agent_stream:
    if "data" in event:
        text_data = event["data"]
        current_response += text_data

        containers["notification"][index].markdown(current_response)

Strands agent通过multi step reasoning多次得出结果。要获得最终结果，请从message的content中提取text，并只提取最后一个。另外，由于每个tool的reference都不同，因此从tool content的text中提取reference，如下所示。

if "message" in event:
    message = event["message"]
    for msg_content in message["content"]:                
        result = msg_content["text"]
        current_response = ""

        tool_content = msg_content["toolResult"]["content"]
        for content in tool_content:
            content, urls, refs = get_tool_info(tool_name, content["text"])
            if refs:
                for r in refs:
                    references.append(r)

名为generate_image_with_colors的工具的最终图像路径将从event_loop_metrics中提取并用作image_urls，如下所示。

if "event_loop_metrics" in event and \
    hasattr(event["event_loop_metrics"], "tool_metrics") and \
    "generate_image_with_colors" in event["event_loop_metrics"].tool_metrics:
    tool_info = event["event_loop_metrics"].tool_metrics["generate_image_with_colors"].tool
    if "input" in tool_info and "filename" in tool_info["input"]:
        fname = tool_info["input"]["filename"]
        if fname:
            url = f"{path}/{s3_image_prefix}/{parse.quote(fname)}.png"
            if url not in image_urls:
                image_urls.append(url)

分发

分发

连接AWS console上的CloudShell。安装python、pip、git、boto3，如下所示。

sudo yum install python3 python3-pip git docker -y
pip install boto3

导入git source，如下所示。

git clone https://github.com/kyopark2014/dwfb-project

使用installer.py开始安装，如下所示。

cd dwfb-project && python3 installer.py

API实现所需的credential由secret管理。因此，安装时需要输入必要的credential，请按照以下方式提前准备credential：

  • 一般互联网搜索： 塔维利搜索连接并注册后发放API密钥。这个以tvly开头。

安装完成后，请通过以下连接到Cloudront验证操作：

如果不再需要基础架构，请使用uninstaller.py将其删除。

python uninstaller.py

更新已部署的应用程序

连接到AWS console的EC2 启动实例选择名为“app-for-dwfb”的instance id，如下所示。

选择“connect”，然后选择Session Manager进行连接。

更新后，请在浏览器中再次检查，如下所示：

cd ~/dwfb-project/ && sudo ./update.sh

检查运行日志

EC2控制台在中选择名为“app-for-dwfb”的instance id，然后使用EC2的Session Manager进行连接。

首先检查当前docker container ID，如下所示。

sudo docker ps

使用container ID检查日志，如下所示。

sudo docker logs [container ID]

实际运行结果如下：

在Local中运行

为了充分利用AWS环境， 安装AWS CLI必须。如果是在EC2上部署，则不需要单独安装。在Local中安装时，请参阅以下命令。

curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip" 
unzip awscliv2.zip
sudo ./aws/install

使用AWS CLI注册AWS credential，如下所示。

aws configure

安装过程中出现的各种问题 基洛气候使用快速修改。可以安装如下，但在Windows上 Kiro安装从下载并安装。运行时在单元格中键入“kiro-cli”。

curl -fsSL https://cli.kiro.dev/install | bash

使用venv配置环境可以方便地管理软件包。设置环境：

python -m venv .venv
source .venv/bin/activate

转到之后下载的github文件夹，然后安装以下所需的其他软件包：

pip install -r requirements.txt

然后使用以下命令运行streamlit：

streamlit run application/app.py

运行结果

参考

Strands Python示例

Strands代理SDK

Strands代理商样品

内置工具示例

介绍Strands Agents，一个开源的AI代理SDK

use_aws.py

查看Strands Agents和开源AI代理SDK

基于亚马逊基岩的药物发现剂

Strands特工-Swarm

Strands Agent流媒体演示

使用生成型AI自动执行AWS安全检查：从Q CLI到Strands Agents

利用AI Agent解决EKS应用程序和基础架构问题

与Strands Agents和AgentCore一起实施生物和制药研究助理

Strands代理商&亚马逊基岩代理商核心워크샵

使用Agentic AI实现评论管理自动化

Strands Agent Workshop（中文）

代理人工智能研讨会：人工智能基金经理

Agentic AI基金经理

Workshop-Strands SDK和AgentCore的代理AI

相关 MCP