探索語意核心 BedrockAgent

BedrockAgent API 文件即將推出。

BedrockAgent API 文件即將上線。

Java 中目前無法使用的功能。

將 Semantic Kernel Bedrock Agents 套件新增至 .NET 專案：

dotnet add package Microsoft.SemanticKernel.Agents.Bedrock --prerelease

這將引進 Bedrock 的語意核心 SDK 支援，包括 AWS SDK for Bedrock 的相依性。 您可能也需要設定 AWS 認證（例如透過環境變數或預設 AWS 設定）。 AWS SDK 將會使用您設定的認證;請確定您已在環境或 AWS 設定檔中設定、 AWS_ACCESS_KEY_IDAWS_SECRET_ACCESS_KEY和預設區域。 （如需詳細資訊，請參閱 AWS 關於認證設定的檔。

使用 AWS 額外項目安裝 Semantic Kernel 套件：

pip install semantic-kernel[aws]

這可確保必要的 AWS 函式庫（例如 boto3）會隨 Semantic Kernel 一起安裝。 在 Python 中使用 Bedrock 代理程式之前，請確定您的 AWS 認證和區域已正確設定（例如，藉由設定環境變數或使用 AWS CLI）。 您應該設定 AWS_ACCESS_KEY_ID、AWS_SECRET_ACCESS_KEY 和 AWS_DEFAULT_REGION（或 AWS 設定檔），以便 boto3 能夠進行驗證。

Java 中目前無法使用的功能。

using Amazon.Bedrock;
using Amazon.Bedrock.Model;
using Amazon.BedrockRuntime;
using Microsoft.SemanticKernel.Agents.Bedrock;

// 1. Define a new agent on the Amazon Bedrock service
IAmazonBedrock bedrockClient = new AmazonBedrockClient();  // uses default AWS credentials & region
var createRequest = new CreateAgentRequest 
{
    AgentName = "<foundation model ID>",          // e.g., "anthropic.claude-v2" or other model
    FoundationModel = "<foundation model ID>",    // the same model, or leave null if AgentName is the model
    AgentResourceArn = "<agent role ARN>",        // IAM role ARN with Bedrock permissions
    Instruction = "<agent instructions>"
};
CreateAgentResponse createResponse = await bedrockClient.CreateAgentAsync(createRequest);

// (Optional) Provide a description as needed:
// createRequest.Description = "<agent description>";

// After creation, the agent is in a "NOT_PREPARED" state.
// Prepare the agent to load tools and finalize setup:
await bedrockClient.PrepareAgentAsync(new PrepareAgentRequest 
{
    AgentId = createResponse.Agent.AgentId
});

// 2. Create a Semantic Kernel agent instance from the Bedrock agent definition
IAmazonBedrockRuntime runtimeClient = new AmazonBedrockRuntimeClient();
BedrockAgent agent = new BedrockAgent(createResponse.Agent, bedrockClient, runtimeClient);

在上述程式代碼中，我們會先使用 AWS SDK （AmazonBedrockClient） 在 Bedrock 上建立代理程式，並指定代理程式應承擔的基礎模型、名稱、指示，以及 IAM 角色的 ARN。 Bedrock 服務會回應代理人定義（包括唯一的 AgentId）。 接著，我們會呼叫 PrepareAgentAsync 將代理程序轉換為就緒狀態（代理程式會從 CREATING 狀態移至NOT_PREPARED，然後在就緒後移至 PREPARED）。 最後，我們會使用傳回的定義和 AWS 用戶端來建構 BedrockAgent 物件。 此 BedrockAgent 實例是我們將用來傳送訊息和接收回應的內容。

import boto3
from semantic_kernel.agents import BedrockAgent

# 1. Define and prepare a new agent on Amazon Bedrock
agent = await BedrockAgent.create_and_prepare_agent(
    name="<agent name>", 
    instructions="<agent instructions>",
    foundation_model="<foundation model ID>",
    agent_resource_role_arn="<agent role ARN>"
)

在上述範例中， BedrockAgent.create_and_prepare_agent 處理完整的建立流程：它會使用 AWS 設定（透過 boto3）在 Bedrock 上建立具有指定名稱、基礎模型和指示的代理程式，然後自動等候代理程式達到就緒狀態（在內部執行準備步驟）。 結果是可供 BedrockAgent 使用的實例。 在幕後，此方法會使用您的預設認證來建立 AWS 用戶端（適用於 Bedrock 和 Bedrock Runtime），因此請確定您的 AWS 環境已設定。 如果您需要自訂組態，您也可以手動建構 AWS 用戶端，並將其當做參數（例如 client= boto3.client("bedrock") 及 runtime_client= boto3.client("bedrock-runtime")）傳遞至 create_and_prepare_agent 呼叫。

Java 中目前無法使用的功能。

針對 .NET，Bedrock 代理程式的標識碼是可透過 agent.Id存取的字串。 若要依標識符擷取現有的代理，請使用 AWS Bedrock 用戶端，然後建構新的 BedrockAgent：

string existingAgentId = "<your agent ID>";
var getResponse = await bedrockClient.GetAgentAsync(new GetAgentRequest { AgentId = existingAgentId });
BedrockAgent agent = new BedrockAgent(getResponse.Agent, bedrockClient, runtimeClient);

在此，我們會在IAmazonBedrock用戶端上呼叫GetAgentAsync（具有已知標識符），這會傳回代理程式的定義（名稱、模型、指示等等）。 然後，我們會使用該定義和相同的用戶端，初始化新的 BedrockAgent 。 此代理程序實例將會連結到現有的 Bedrock 代理程式。

在 Python 中，您可以使用 AWS Bedrock boto3 用戶端，依據 ID 來擷取代理程式，然後將它包裝在 BedrockAgent 中。

import asyncio, boto3
from semantic_kernel.agents import BedrockAgent

agent_id = "<your agent ID>"
bedrock_client = boto3.client("bedrock")  # Bedrock service client
# Fetch the agent's definition from AWS
agent_info = await asyncio.to_thread(bedrock_client.get_agent, AgentId=agent_id)
# Create the BedrockAgent instance from the retrieved definition
agent = BedrockAgent(agent_model=agent_info["agent"])

在此代碼段中，我們使用 boto3 呼叫 Bedrock 服務上的 get_agent，由於 boto3 會阻塞，因此透過 asyncio.to_thread 以執行緒的方式運行它。 傳回的 agent_info["agent"] 包含代理程式的詳細數據（標識符、名稱、狀態等），我們傳遞至 BedrockAgent 建構函式。 因為我們未明確提供 AWS 用戶端給 BedrockAgent，因此系統會在內部建立具有預設設定的新用戶端。 （或者，如果您有客戶端，您可以提供 client= 和 runtime_client= 以重複使用客戶端。）

Java 中目前無法使用的功能。

Bedrock 代理人線程的詳細數據由 BedrockAgentThread 類別抽象化（該類別實作了通用的 AgentThread 介面）。 BedrockAgent目前只支援 類型BedrockAgentThread為的線程。

BedrockAgent agent = /* (your BedrockAgent instance, as created above) */;

// Start a new conversation thread for the agent
AgentThread agentThread = new BedrockAgentThread(runtimeClient);
try
{
    // Send a user message and iterate over the response(s)
    var userMessage = new ChatMessageContent(AuthorRole.User, "<your user input>");
    await foreach (ChatMessageContent response in agent.InvokeAsync(userMessage, agentThread))
    {
        Console.WriteLine(response.Content);
    }
}
finally
{
    // Clean up the thread and (optionally) the agent when done
    await agentThread.DeleteAsync();
    await agent.Client.DeleteAgentAsync(new DeleteAgentRequest { AgentId = agent.Id });
}

在此範例中，我們會明確建立 BedrockAgentThread，並傳入 runtimeClient，以便與 Bedrock 執行階段服務通訊。 接著，我們呼叫 agent.InvokeAsync(...)，並使用 ChatMessageContent 來代表使用者訊息。 InvokeAsync 會傳回異步的回應串流 – 實際上，Bedrock 代理程式通常會針對每次調用傳回一個最終回應（因為中繼工具動作會個別處理），因此您通常會從迴圈中取得單 ChatMessageContent 一回應。 我們列印出助理的回復（response.Content）。 在最後一個區塊中，我們會刪除線程，以結束 AWS 上的 Bedrock 會話。 在此案例中，我們也會刪除代理程式本身（因為我們只針對此範例建立代理程式）–此步驟是選擇性的，只有在您不打算再次重複使用代理程式時才需要 （請參閱下面的刪除 BedrockAgent）。

您可以針對後續呼叫重複使用相同的 agentThread 來繼續現有的交談。 例如，您可以循環讀取使用者輸入，並在每次使用相同的線程呼叫 InvokeAsync ，以執行多回合對話。 您也可以建立具有已知會話標識符的 BedrockAgentThread，以繼續先前儲存的交談：

string sessionId = "<existing Bedrock session ID>";
AgentThread thread = new BedrockAgentThread(runtimeClient, sessionId);
// Now `InvokeAsync` using this thread will continue the conversation from that session

在 Python 中使用 Bedrock Agent 與 BedrockAgentThread 管理會話的方式類似。 您可以啟動新的線程，或傳遞現有的線程以繼續交談：

from semantic_kernel.agents import BedrockAgentThread

# Assume `agent` is your BedrockAgent instance
USER_INPUTS = ["Hello", "What's your name?"]

thread = BedrockAgentThread()  # start a new conversation thread (session)
try:
    for user_input in USER_INPUTS:
        response = await agent.get_response(messages=user_input, thread=thread)
        print(response)  # print the assistant's reply
        thread = response.thread  # update thread (BedrockAgentThread) for next turn
finally:
    await thread.delete() if thread else None

在此程式代碼中，我們會迴圈執行數個用戶輸入。 在每個反覆專案上，我們會使用使用者訊息和目前的線程來呼叫 agent.get_response(...) 。 第一次呼叫會開始貝德洛克會話，並返回一個包含助理答案的 AgentResponseItem（或 ChatMessageContent）。 我們會列印回應，然後取用 response.thread——這是已用新訊息內容更新的相同 BedrockAgentThread——以用於下一輪。 交談之後（在此範例中為兩回合），我們會刪除線程以結束 AWS 上的會話。

如果您在呼叫中省略 thread 參數， agent.get_response 或 agent.invoke 會自動為該調用建立新的線程，並將它包含在回應中。

或者，您也可以將訊息清單傳遞至 get_response 或使用異步串流調用，一次傳送一批訊息。 例如，若要針對單一提示串流助理的回應（令牌依據令牌）：

# Streaming a single response from the Bedrock agent
async for partial in agent.invoke_stream(messages="Tell me a joke.", thread=thread):
    print(partial.content, end="")

方法 invoke_stream(...) 會在產生回應時產生 ChatMessageContent 物件。 藉由逐一遍歷，您可以逐步輸出助理的答案（這裡我們列印字元時不加換行符，以形成完整回應）。

Java 中目前無法使用的功能。

使用 Bedrock 客戶端依代理 ID 刪除。 例如：

await bedrockAgent.Client.DeleteAgentAsync(new() { AgentId = bedrockAgent.Id });

在此呼叫之後，代理程式的狀態將會變更，且將無法再使用。 （嘗試叫用已刪除的代理程式會導致錯誤。

呼叫代理程式的刪除方法。 例如：

await agent.delete_agent()

這會呼叫 Bedrock 服務來刪除代理程式（並在內部將對象標示 BedrockAgent 為已刪除）。 您可以透過檢查 agent.id 或（如果提供）旗標來驗證（例如 _is_deleted）。

Java 中目前無法使用的功能。

from semantic_kernel.contents import ChatMessageContent, FunctionCallContent, FunctionResultContent
from semantic_kernel.functions import kernel_function

# Define a sample plugin with two functions
class MenuPlugin:
    @kernel_function(description="Provides a list of specials from the menu.")
    def get_specials(self) -> str:
        return "Soup: Clam Chowder; Salad: Cobb Salad; Drink: Chai Tea"

    @kernel_function(description="Provides the price of a menu item.")
    def get_item_price(self, menu_item: str) -> str:
        return "$9.99"

# Callback to handle intermediate messages
async def handle_intermediate_steps(message: ChatMessageContent) -> None:
    for item in (message.items or []):
        if isinstance(item, FunctionCallContent):
            print(f"Function Call:> {item.name} with arguments: {item.arguments}")
        elif isinstance(item, FunctionResultContent):
            print(f"Function Result:> {item.result} for function: {item.name}")
        else:
            print(f"[Intermediate] {item}")

# Create the BedrockAgent with the plugin (assuming agent is not yet created above)
agent = await BedrockAgent.create_and_prepare_agent(
    name="MenuAgent",
    instructions="You are a restaurant assistant.",
    foundation_model="<model ID>",
    agent_resource_role_arn="<role ARN>",
    plugins=[MenuPlugin()]  # include our custom plugin
)

# Start a conversation with intermediate callback
thread = BedrockAgentThread()
user_queries = [
    "Hello!",
    "What are the specials today?",
    "What is the special drink?",
    "How much is that?"
]
try:
    for query in user_queries:
        print(f"# User: {query}")
        async for response in agent.invoke(messages=query, thread=thread, on_intermediate_message=handle_intermediate_steps):
            print(f"# Assistant: {response}")
            thread = response.thread
finally:
    await thread.delete() if thread else None
    await agent.delete_agent()

在此程式代碼中，每當代理程式需要從 MenuPlugin （例如 get_specials 或 get_item_price） 呼叫函式時，回 handle_intermediate_steps 呼就會列印出函式呼叫的行，另一行用於函式結果。 然後，每個使用者查詢的最終回應會如常顯示。 藉由觀察中繼內容，您可以追蹤代理人如何到達其答案（使用哪個工具、工具產生的結果等等）。

例如，輸出可能如下所示：

# User: Hello!
# Assistant: Hello! How can I assist you today?
# User: What are the specials today?
Function Call:> MenuPlugin-get_specials with arguments: {}
Function Result:> Soup: Clam Chowder; Salad: Cobb Salad; Drink: Chai Tea for function: MenuPlugin-get_specials
# Assistant: The specials today include Clam Chowder for the soup, Cobb Salad, and Chai Tea as a special drink.
# User: What is the special drink?
# Assistant: The special drink is Chai Tea.
# User: How much is that?
Function Call:> MenuPlugin-get_item_price with arguments: {"menu_item": "Chai Tea"}
Function Result:> $9.99 for function: MenuPlugin-get_item_price
# Assistant: The special drink (Chai Tea) costs $9.99.

在上述互動中，中繼列印會顯示代理程式在適當時間成功呼叫 MenuPlugin.get_specials 和 MenuPlugin.get_item_price ，並使用其結果來回應使用者。 您可以視需要在應用程式邏輯中記錄或使用這些中繼詳細數據（例如，顯示代理程序採取的步驟）。

BedrockAgent （C#） 中中繼訊息的回呼支援遵循類似的模式，但確切的 API 正在開發中。 （未來的版本將允許註冊委派來處理 FunctionCallContent 和 FunctionResultContent，並在 InvokeAsync 期間進行處理。）

Java 中目前無法使用的功能。

使用宣告式規格可以簡化組態，特別是如果您想要輕鬆地切換代理程式設定或使用組態檔方法。 對於 Bedrock 代理程式，YAML 定義看起來可能如下：

type: bedrock_agent
name: MenuAgent
description: Agent that answers questions about a restaurant menu
instructions: You are a restaurant assistant that provides daily specials and prices.
model:
  id: anthropic.claude-v2
agent_resource_role_arn: arn:aws:iam::123456789012:role/BedrockAgentRole
tools:
  - type: code_interpreter
  - type: user_input
  - name: MenuPlugin
    type: kernel_function

在這個（假設的）YAML 中，我們定義類型為bedrock_agent的代理，為其命名並提供指示，根據標識 ID 指定基礎模型，並提供它應該使用的角色 ARN。 我們也宣告了幾個工具：一個啟用內建程式代碼解釋器、另一個啟用內建的使用者輸入工具，以及自定義 MenuPlugin（這會在程式碼中個別定義，並註冊為核心函式）。 這類檔案會以人類可讀取的形式封裝代理程序的設定。

若要從 YAML 具現化代理程式，請使用靜態載入器搭配適當的處理站。 例如：

string yamlText = File.ReadAllText("bedrock-agent.yaml");
var factory = new BedrockAgentFactory();  // or an AggregatorAgentFactory if multiple types are used
Agent myAgent = await KernelAgentYaml.FromAgentYamlAsync(kernel, yamlText, factory);

這將使用提供的核心和工廠剖析 YAML，並根據 type 字段產生 BedrockAgent 實例（或其他類型）。

貝洛基特宣告式規格處理即將推出。

Java 中目前無法使用的功能。