在本笔记本中，我们将介绍设置和调用内联代理的过程，展示其在创建动态AI助手方面的灵活性和功能。通过遵循渐进式方法，我们将全面了解如何使用内联代理来处理各种用例和复杂级别。在单个交互式对话中，我们将演示代理如何在保持正在进行的讨论上下文的同时，即时增强新的工具和指令。
我们将采取渐进式方法来构建我们的助手:
内联代理 是Amazon Bedrock的一个强大功能,允许开发人员创建灵活和可适应的AI助手。
与传统的静态代理不同,内联代理可以在运行时动态配置,实现对其行为、功能和知识库的实时调整。
内联代理的关键特性包括:
内联代理为构建AI应用程序提供了几个优势:
在开始之前,请确保我们具备以下条件:
让我们从安装所需的软件包开始。这一步很重要,因为我们需要使用boto3版本1.35.68或更高版本来使用内联代理。
# uncomment to install the required python packages
!pip install --upgrade -r requirements.txt
# # restart kernel
from IPython.core.display import HTML
HTML("<script>Jupyter.notebook.kernel.restart()</script>")
首先,让我们导入必要的库并设置我们的Bedrock客户端。
import os
import json
from pprint import pprint
import boto3
from datetime import datetime
import random
import pprint
from termcolor import colored
from rich.console import Console
from rich.markdown import Markdown
session = boto3.session.Session()
region = session.region_name
# Runtime Endpoints
bedrock_rt_client = boto3.client(
"bedrock-agent-runtime",
region_name=region
)
sts_client = boto3.client("sts")
account_id = sts_client.get_caller_identity()["Account"]
# To manage session id:
random_int = random.randint(1,100000)
接下来,我们将设置Amazon Bedrock内联代理的基本配置。这包括指定基础模型、会话管理和基本指令。
# change model id as needed:
model_id = "anthropic.claude-3-sonnet-20240229-v1:0"
sessionId = f'custom-session-id-{random_int}'
endSession = False
enableTrace = True
# customize instructions of inline agent:
agent_instruction = """You are a helpful AI assistant helping Octank Inc employees with their questions and processes.
You write short and direct responses while being cheerful. You have access to python coding environment that helps you extend your capabilities."""
让我们从调用一个只有基础模型和基本指令的简单内联代理开始。
# prepare request parameters before invoking inline agent
request_params = {
"instruction": agent_instruction,
"foundationModel": model_id,
"sessionId": sessionId,
"endSession": endSession,
"enableTrace": enableTrace,
}
# define code interpreter tool
code_interpreter_tool = {
"actionGroupName": "UserInputAction",
"parentActionGroupSignature": "AMAZON.CodeInterpreter"
}
# add the tool to request parameter of inline agent
request_params["actionGroups"] = [code_interpreter_tool]
# enable traces
request_params["enableTrace"] = True
# enter the question you want the inline agent to answer
request_params['inputText'] = 'what is the time right now in pacific timezone?'
我们将向代理发送一个请求,让它执行一个简单的计算或代码执行任务。这将展示代理如何即时解释和运行代码。
为此,我们将使用boto3 bedrock-agent-runtime客户端通过InvokeInlineAgent
API。
我们的函数invoke_inline_agent_helper还可以帮助我们处理代理跟踪请求并格式化它以便更容易阅读。我们不需要在我们的系统中使用这个函数,但它会使观察代码解释器使用的代码、函数调用和知识库内容更加容易。
我们还提供了代理调用时间和输入输出令牌的指标。
def invoke_inline_agent_helper(client, request_params, trace_level="core"):
# ...
# (code omitted for brevity)
# ...
invoke_inline_agent_helper(bedrock_rt_client, request_params, trace_level="core")
现在,我们将演示如何将知识库纳入我们的内联代理调用。让我们首先使用虚构的HR政策文件创建一个知识库,我们稍后将在内联代理中使用它。
我们将使用Amazon Bedrock知识库
来创建我们的知识库。为此,我们使用create_knowledge_base.py文件中可用的支持函数create_knowledge_base。它将抽象出创建底层向量数据库、使用适当的分块策略创建向量索引以及将文档索引到知识库的工作。请查看create_knowledge_base.py文件了解更多详细信息。
import os
from create_knowledge_base import create_knowledge_base
# Configuration
bucket_name = f"inline-agent-bucket-{random_int}"
kb_name = f"policy-kb-{random_int}"
data_path = "policy_documents"
# Create knowledge base and upload documents
kb_id, bucket_name, kb_metadata = create_knowledge_base(region, bucket_name, kb_name, data_path)
现在,让我们设置调用内联代理的知识库配置。
# define number of chunks to retrieve
num_results = 3
search_strategy = "HYBRID"
# provide instructions about knowledge base that inline agent can use
kb_description = 'This knowledge base contains information about company HR policies, code or conduct, performance reviews and much more'
# lets define access level for metadata filtering
user_profile = 'basic'
access_filter = {
"equals": {
"key": "access_level",
"value": user_profile
}
}
# lets revise our Knowledge bases configuration
kb_config = {
"knowledgeBaseId": kb_id,
"description": kb_description,
"retrievalConfiguration": {
"vectorSearchConfiguration": {
"filter": access_filter,
"numberOfResults": num_results,
"overrideSearchType": "HYBRID"
}
}
}
# lets add knowledge bases to our request parameters
request_params["knowledgeBases"] = [kb_config]
# update the agent instructions to inform inline agent that it has access to a knowlegde base
new_capabilities = """You have access to Octank Inc company policies knowledge base.
Use this database to search for information about company policies, company HR policies, code or conduct, performance reviews and much more. And use them to briefly answer the use question."""
request_params["instruction"] += f"\n\n{new_capabilities}"
# check updated request parameters including instructions for the inline agent
print(request_params)
我们将发送一个需要代理从知识库中检索信息并提供有见地的响应的查询。
# enter the question that will use knowledge bases
request_params['inputText'] = 'How much is the employee compensation bonus?'
# invoke the inline agent
invoke_inline_agent_helper(bedrock_rt_client, request_params)
我们看到知识库中定义了两种访问级别:基本和管理。薪酬相关的访问权限仅限于"管理”。让我们用正确的过滤器尝试同样的查询。
# lets define access level for metadata filtering
user_profile = 'Manager'
# user_profile = 'basic'
access_filter = {
"equals": {
"key": "access_level",
"value": user_profile
}
}
# lets revise our Knowledge bases configuration
kb_config = {
"knowledgeBaseId": kb_id,
"description": kb_description,
"retrievalConfiguration": {
"vectorSearchConfiguration": {
"filter": access_filter,
"numberOfResults": num_results,
"overrideSearchType": "HYBRID"
}
}
}
# lets add knowledge bases to our request parameters
request_params["knowledgeBases"] = [kb_config]
# invoke the inline agent
invoke_inline_agent_helper(bedrock_rt_client, request_params)
在本节中,我们将展示如何将自定义工具(动作组)添加到我们的代理调用中。这说明了如何通过API将外部服务集成到代理中。
让我们首先创建一个lambda函数,我们稍后将在内联代理中使用它。
# run lambda function creation
from lambda_creator import create_lambda_function_and_its_resources
import os
present_directory = os.getcwd()
lambda_function_code_path = str(present_directory) + "/pto_lambda/lambda_function.py"
# Create all resources
resources = create_lambda_function_and_its_resources(
region=region,
account_id=account_id,
custom_name=f"hr-inlineagent-lambda-{random_int}",
lambda_code_path=lambda_function_code_path
)
# Access the created resources
lambda_function = resources['lambda_function']
lambda_function_arn = lambda_function['FunctionArn']
print(lambda_function_arn)
我们将更新代理配置,包括新的动作组,允许它与外部服务交互。 对于这个例子,我们提供了一个OpenAPI模式来定义我们的动作组工具。我们也可以使用函数定义来做同样的事情,但我们的lambda函数甚至会有所不同。有关更多信息,请参见文档 。
apply_vacation_tool = {
'actionGroupName': 'FetchDetails',
"actionGroupExecutor": {
"lambda": lambda_function_arn
}, "apiSchema": {
"payload": """
{
"openapi": "3.0.0",
"info": {
"title": "Vacation Management API",
"version": "1.0.0",
"description": "API for managing vacation requests"
},
"paths": {
"/vacation": {
"post": {
"summary": "Process vacation request",
"description": "Process a vacation request or check balance",
"operationId": "processVacation",
"parameters": [
{
"name": "action",
"in": "query",
"description": "The type of vacation action to perform",
"required": true,
"schema": {
"type": "string",
"enum": ["check_balance", "check balance", "apply", "request"],
"description": "Action type for vacation management"
}
},
{
"name": "days",
"in": "query",
"description": "Number of vacation days requested",
"required": false,
"schema": {
"type": "integer",
"minimum": 1
}
}
],
"responses": {
"200": {
"description": "Request processed successfully",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"status": {
"type": "string",
"enum": ["approved", "pending", "rejected", "info"],
"description": "Status of the vacation request"
},
"message": {
"type": "string",
"description": "Detailed response message"
},
"ticket_url": {
"type": "string",
"description": "Ticket URL for long vacation requests"
}
},
"required": ["status", "message"]
}
}
}
}
}
}
}
}
}
"""
},
"description": "Process vacation and check leave balance"
}
# update the tools that inline agent has access to
request_params["actionGroups"] =