利用 Bedrock 的代理

这个笔记本应该可以很好地与 SageMaker Studio 中的 Data Science 3.0 内核一起使用。我们也可以在本地设置上运行，只要我们有正确的 IAM 凭证来通过 Bedrock 调用 Claude 模型

在这个演示笔记本中，我们演示了通过 Bedrock 使用 Anthropic 的 Claude 模型进行函数调用的实现。这个笔记本受到 Anthropic 团队原始工作 的启发，并将其修改以与 Amazon Bedrock 一起使用。

这个笔记本需要访问 Bedrock 中的 anthropic.claude-3-sonnet-20240229-v1:0 模型

概述

对话界面，如聊天机器人和虚拟助手，可用于增强我们客户的用户体验。它们使用自然语言处理(NLP)和机器学习算法来理解和响应用户查询，可用于各种应用程序，如客户服务、销售和电子商务，以快速高效地回答用户。通常，它们通过从各种渠道(如网站、社交媒体平台和消息应用程序)获取信息而得到增强，这涉及一个复杂的工作流程，如下所示。

使用 Amazon Bedrock 的聊天机器人

Amazon Bedrock - Agents Interface

构建 - 关键元素

构建上下文感知聊天机器人的第一个过程是识别可由 LLM 调用的工具。

第二个过程是用户请求编排、交互、调用和返回结果

架构[天气查找]

我们搜索并查找纬度和经度，然后调用天气应用程序获取预报

Amazon Bedrock - Agents Interface

如果我们没有这些库，请取消注释并安装下面的库

!pip install langchain==0.1.17
!pip install langchain-anthropic
!pip install boto3==1.34.95
!pip install faiss-cpu==1.8.0

要安装 langchain-aws

我们可以运行 pip install langchain-aws

要获取最新版本，请使用以下命令 #- 从我们机器上的终端运行 cd ~ mkdir temp_t cd temp_t git clone https://github.com/langchain-ai/langchain-aws/ pip install ./langchain-aws/libs/aws/

设置

⚠️ ⚠️ ⚠️ 在运行这个笔记本之前，确保我们有所需的库并可以访问本笔记本中的天气 api。⚠️ ⚠️ ⚠️

import os
from typing import Optional

# 外部依赖项:
import boto3
from botocore.config import Config


def get_bedrock_client(assumed_role: Optional[str] = None, region: Optional[str] = 'us-east-1',runtime: Optional[bool] = True,external_id=None, ep_url=None):
    """创建一个 boto3 客户端用于 Amazon Bedrock,可选择配置覆盖 
    """
    target_region = region

    print(f"创建新客户端\n  使用区域: {target_region}:external_id={external_id}: ")
    session_kwargs = {"region_name": target_region}
    client_kwargs = {**session_kwargs}

    profile_name = os.environ.get("AWS_PROFILE")
    if profile_name:
        print(f"  使用配置文件: {profile_name}")
        session_kwargs["profile_name"] = profile_name

    retry_config = Config(
        region_name=target_region,
        retries={
            "max_attempts": 10,
            "mode": "standard",
        },
    )
    session = boto3.Session(**session_kwargs)

    if assumed_role:
        print(f"  使用角色: {assumed_role}", end='')
        sts = session.client("sts")
        if external_id:
            response = sts.assume_role(
                RoleArn=str(assumed_role),
                RoleSessionName="langchain-llm-1",
                ExternalId=external_id
            )
        else:
            response = sts.assume_role(
                RoleArn=str(assumed_role),
                RoleSessionName="langchain-llm-1",
            )
        print(f"使用角色: {assumed_role} ... sts::成功!")
        client_kwargs["aws_access_key_id"] = response["Credentials"]["AccessKeyId"]
        client_kwargs["aws_secret_access_key"] = response["Credentials"]["SecretAccessKey"]
        client_kwargs["aws_session_token"] = response["Credentials"]["SessionToken"]

    if runtime:
        service_name='bedrock-runtime'
    else:
        service_name='bedrock'

    if ep_url:
        bedrock_client = session.client(service_name=service_name,config=retry_config,endpoint_url = ep_url, **client_kwargs )
    else:
        bedrock_client = session.client(service_name=service_name,config=retry_config, **client_kwargs )

    print("boto3 Bedrock 客户端成功创建!")
    print(bedrock_client._endpoint)
    return bedrock_client

import json
import os
import sys

import boto3
import botocore



# ---- ⚠️ 取消注释并根据我们的 AWS 设置编辑以下行 ⚠️ ----

# os.environ["AWS_DEFAULT_REGION"] = "<REGION_NAME>"  # 例如 "us-east-1"
# os.environ["AWS_PROFILE"] = "<YOUR_PROFILE>"
# os.environ["BEDROCK_ASSUME_ROLE"] = "<YOUR_ROLE_ARN>"  # 例如 "arn:aws:..."


bedrock_runtime = get_bedrock_client() #
#     assumed_role=os.environ.get("BEDROCK_ASSUME_ROLE", None),
#     region=os.environ.get("AWS_DEFAULT_REGION", None)
# )

Anthropic Claude

输入


"messages": [
    {"role": "user", "content": "Hello, Claude"},
    {"role": "assistant", "content": "Hello!"},
    {"role": "user", "content": "Can you describe LLMs to me?"}
        
]
{
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": 100,
    "messages": messages,
    "temperature": 0.5,
    "top_p": 0.9
}

输出

{
    'id': 'msg_01T',
    'type': 'message',
    'role': 'assistant',
    'content': [
        {
            'type': 'text',
            'text': '当然,这个概念...'
        }
    ],
    'model': 'model_id',
    'stop_reason': 'max_tokens',
    'stop_sequence': None,
    'usage': {'input_tokens':xy, 'output_tokens': yz}}

Bedrock 模型

Anthropic Claude

关键是让 LLM(即 Claude 模型)知道它可以调用的一组"工具”，即它可以在一组标签之间调用的函数。这是可能的，因为 Anthropic 的 Claude 模型已经在其训练语料库中广泛训练了这些标签。

然后提供一种逐步调用工具的方法，直到获得正确的答案。我们创建一组可调用的函数，下面我们展示了一个示例函数，可以根据需要进行修改。

漂亮打印的帮助函数

from io import StringIO
import sys
import textwrap
from langchain.llms.bedrock import Bedrock
from typing import Optional, List, Any
from langchain.callbacks.manager import CallbackManagerForLLMRun

def print_ww(*args, width: int = 100, **kwargs):
    """像 print() 一样,但将输出换行到 `width` 个字符(默认 100)"""
    buffer = StringIO()
    try:
        _stdout = sys.stdout
        sys.stdout = buffer
        print(*args, **kwargs)
        output = buffer.getvalue()
    finally:
        sys.stdout = _stdout
    for line in output.splitlines():
        print("\n".join(textwrap.wrap(line, width=width)))

第 1 节. 连接性和调用

调用模型以确保连接

import json 
modelId = "anthropic.claude-3-sonnet-20240229-v1:0" #"anthropic.claude-v2"

messages=[
    { 
        "role":'user', 
        "content":[{
            'type':'text',
            'text': "什么是量子力学? "
        }]
    },
    { 
        "role":'assistant', 
        "content":[{
            'type':'text',
            'text': "它是物理学的一个分支,描述了物质和能量如何以离散的能量值进行交互"
        }]
    },
    { 
        "role":'user', 
        "content":[{
            'type':'text',
            'text': "你能更详细地解释一下离散能量吗?"
        }]
    }
]
body=json.dumps(
        {
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 100,
            "messages": messages,
            "temperature": 0.5,
            "top_p": 0.9
        }  
    )  
    
response = bedrock_runtime.invoke_model(body=body, modelId=modelId)
response_body = json.loads(response.get('body').read())
print_ww(response_body)


def test_sample_claude_invoke(prompt_str,boto3_bedrock ):
    modelId = "anthropic.claude-3-sonnet-20240229-v1:0" #"anthropic.claude-v2"
    messages=[{ 
        "role":'user', 
        "content":[{
            'type':'text',
            'text': prompt_str
        }]
    }]
    body=json.dumps(
        {
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 100,
            "messages": messages,
            "temperature": 0.5,
            "top_p": 0.9
        }  
    )  
    response = boto3_bedrock.invoke_model(body=body, modelId=modelId)
    response_body = json.loads(response.get('body').read())
    return response_body


test_sample_claude_invoke("什么是量子力学", bedrock_runtime)

使用 Chain 中的 BedrockChat 类调用 Langchain

from langchain_community.chat_models import BedrockChat
from langchain_core.messages import HumanMessage
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory

llm_chat = BedrockChat(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0", 
    model_kwargs={"temperature": 0.1},
    #region_name="us-west-2",
    client=bedrock_runtime,
)

memory = ConversationBufferMemory()

conversation = ConversationChain(
    llm=llm_chat, verbose=True, memory=memory
)

resp = conversation.predict(input="Hi there!")
print_ww(resp)

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
#from langchain_openai import ChatOpenAI

modelId = "anthropic.claude-3-sonnet-20240229-v1:0" #"anthropic.claude-v2"
cl_llm = BedrockChat(
    model_id=modelId,
    client=bedrock_runtime,
    #model_kwargs={"max_tokens_to_sample": 100},
    model_kwargs={"temperature": 0.1},
)
memory = ConversationBufferMemory()
conversation = ConversationChain( llm=cl_llm, verbose=True, memory=memory)


prompt_t = ChatPromptTemplate.from_messages([("human", "解释一下 {question}.")])
chain_t = prompt_t | cl_llm | StrOutputParser()

resp = chain_t.invoke({'question':"什么是量子物理"}) # has to match the variables
print_ww(resp)

第 2 节 – 工具

首先我们展示工具如何工作以及如何本地完成

创建一组帮助函数

我们将创建一组可以在我们的应用程序中重复使用的函数

  1. 我们需要创建一个提示模板。这个模板有助于 Bedrock 模型理解工具以及如何调用它们。
  2. 创建一种方法来读取可用的工具并将其添加到用于调用 Claude 的提示中
  3. 调用函数，该函数将负责使用正确的参数实际调用该函数
  4. 格式化结果以帮助模型利用结果进行总结
  5. 添加到提示。返回的结果需要添加到提示中，并再次调用模型以获得正确的结果

查看此笔记本了解更多详细信息 

from langchain_community.chat_models import BedrockChat
from langchain_core.messages import HumanMessage
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory

添加工具

递归添加可用的工具

使用 LangChain

现在使用 langchain 和注释来创建工具并调用函数。我们将调用天气 api。第一个工具将用于查找纬度，第二个工具将用于获取天气

import requests

from langchain.tools import tool
from langchain.tools import StructuredTool
from langchain.agents import load_tools
from langchain.agents import initialize_agent， Tool
from langchain.agents import AgentType
from langchain import LLMMathChain

@tool ("get_lat_long")
def get_lat_long(place: