GenAI > Dify >

使用亚马逊 Bedrock 进行检索增强生成 - 自动从 API 检索数据

请注意：本笔记本应该可以很好地与 SageMaker Studio 中的 Data Science 3.0 内核一起使用

在这个研讨会中，我们一直在使用语义相似性搜索进行非结构化文本检索。然而，亚马逊 Bedrock 可以利用的另一种重要类型的检索是从 API 进行结构化数据检索。结构化数据检索对于用最新信息增强 LLM 应用程序非常有用，因为这些信息可以以可重复的方式检索，但输出总是在变化。我们可能会问 LLM 的一个例子是"我在亚马逊上订购的袜子什么时候会到达?"。在本笔记本中，我们将展示如何将 LLM 与后端 API 服务集成，以便通过 RAG 回答用户的问题。

具体来说，我们将构建一个能够根据自然语言告诉我们天气的工具。这是一个相当简单的例子，但它很好地展示了 LLM 如何使用多个 API 工具来检索动态数据来增强提示。以下是我们今天将构建的体系结构的可视化。

api

让我们开始吧!

安装依赖项

!pip install xmltodict --quiet

设置 `boto3` 连接

import boto3
import os
from IPython.display import Markdown, display, Pretty

region = os.environ.get("AWS_REGION")
boto3_bedrock = boto3.client(
    service_name='bedrock-runtime',
    region_name=region,
)

定义 API 工具

我们需要做的第一件事是为我们的 LLM 定义可访问的工具。在这种情况下，我们将定义本地 Python 函数，但重要的是要知道这些可以是任何类型的应用程序服务。这些工具在 AWS 上的示例包括…

AWS Lambda 函数
Amazon RDS 数据库连接
Amazon DynamnoDB 表

更一般的示例包括…

REST API
数据仓库、数据湖和数据库
计算引擎

在这种情况下，我们在下面定义了两个工具，它们可以访问外部 API

根据自然语言输入检索某个地方的纬度和经度
根据输入的纬度和经度检索天气

import requests

def get_weather(latitude: str, longitude: str):
    url = f"https://api.open-meteo.com/v1/forecast?latitude={latitude}&longitude={longitude}&current_weather=true"
    response = requests.get(url)
    return response.json()

def get_lat_long(place: str):
    url = "https://nominatim.openstreetmap.org/search"
    params = {'q': place, 'format': 'json', 'limit': 1}
    headers = {'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/39.0.2171.95 Safari/537.36'}
    response = requests.get(url, params=params,headers=headers).json()
    if response:
        lat = response[0]["lat"]
        lon = response[0]["lon"]
        return {"latitude": lat, "longitude": lon}
    else:
        return None

def call_function(tool_name, parameters):
    func = globals()[tool_name]
    output = func(**parameters)
    return output

我们还定义了一个名为 call_function 的函数，用于抽象工具名称。我们可以看到下面确定拉斯维加斯天气的示例。

place = 'Las Vegas'
lat_long_response = call_function('get_lat_long', {'place' : place})
weather_response = call_function('get_weather', lat_long_response)
print(f'Weather in {place} is...')
weather_response

正如我们所料，我们必须向我们的 LLM 描述我们的工具，以便它知道如何使用它们。下面的字符串以 XML 友好的格式描述了纬度/经度和天气的 Python 函数，我们之前在研讨会中已经看到过。

get_weather_description = """\
<tool_description>
<tool_name>get_weather</tool_name>
<parameters>
<name>latitude</name>
<name>longitude</name>
</parameters>
</tool_description>
"""

get_lat_long_description = """
<tool_description>
<tool_name>get_lat_long</tool_name>
<parameters>
<name>place</name>  
</parameters>
</tool_description>"""

list_of_tools_specs = [get_weather_description, get_lat_long_description]
tools_string = ''.join(list_of_tools_specs)
print(tools_string)

定义提示以使用工具编排我们的 LLM

现在工具已经以编程方式和字符串的形式定义好了，我们可以开始编排将回答用户问题的流程了。实现这一目标的第一步是创建一个定义 Claude 操作规则的提示。在下面的提示中，我们明确指示 Claude 如何使用工具来回答这些问题。

from langchain import PromptTemplate

TOOL_TEMPLATE = """\
Your job is to formulate a solution to a given <user-request> based on the instructions and tools below.

Use these Instructions: 
1. In this environment you have access to a set of tools and functions you can use to answer the question.
2. You can call the functions by using the <function_calls> format below.
3. Only invoke one function at a time and wait for the results before invoking another function.
4. The Results of the function will be in xml tag <function_results>. Never make these up. The values will be provided for you.
5. Only use the information in the <function_results> to answer the question.
6. Once you truly know the answer to the question, place the answer in <answer></answer> tags. Make sure to answer in a full sentence which is friendly.
7. Never ask any questions

<function_calls>
<invoke>
<tool_name>$TOOL_NAME</tool_name>
<parameters>
<$PARAMETER_NAME>$PARAMETER_VALUE</$PARAMETER_NAME>
...
</parameters>
</invoke>
</function_calls>

Here are the tools available:
<tools>
{tools_string}
</tools>

<user-request>
{user_input}
</user-request>

H: What is the first step in order to solve this problem?

A:
"""
TOOL_PROMPT = PromptTemplate.from_template(TOOL_TEMPLATE)

执行 RAG 工作流

有了我们的提示和结构化工具，我们现在可以编写一个编排函数，它将逐步完成回答用户问题的逻辑任务。在下面的单元格中，我们使用 invoke_model 函数生成 Claude 的响应，并使用 single_retriever_step 函数来迭代调用工具。一般流程如下…

用户输入应用程序的输入
将用户输入与原始提示合并，并发送给 LLM 以确定下一步
如果 LLM 知道答案，它将回答，我们就完成了。如果不是，请转到步骤 4。
LLM 将确定要使用哪个工具来回答问题。
我们将按 LLM 的指示使用该工具并检索结果。
我们将结果作为更多上下文提供给原始提示。
我们询问 LLM 下一步或它是否知道答案。
返回步骤 3。

如果这有点令人困惑，不要担心，我们很快就会通过一个示例来演练这个流程!

import xmltodict
import json

def invoke_model(prompt):
    client = boto3.client(service_name='bedrock-runtime', region_name=os.environ.get("AWS_REGION"),)
    body = json.dumps({"prompt": prompt, "max_tokens_to_sample": 500, "temperature": 0,})
    modelId = "anthropic.claude-instant-v1"
    response = client.invoke_model(
        body=body, modelId=modelId, accept="application/json", contentType="application/json"
    )
    return json.loads(response.get("body").read()).get("completion")

def single_retriever_step(prompt, output):

    # first check if the model has answered the question
    done = False
    if '<answer>' in output:
        answer = output.split('<answer>')[1]
        answer = answer.split('</answer>')[0]
        done = True
        return done, answer
    
    # if the model has not answered the question, go execute a function
    else:

        # parse the output for any 
        function_xml = output.split('<function_calls>')[1]
        function_xml = function_xml.split('</function_calls>')[0]
        function_dict = xmltodict.parse(function_xml)
        func_name = function_dict['invoke']['tool_name']
        parameters = function_dict['invoke']['parameters']

        # call the function which was parsed
        func_response = call_function(func_name, parameters)

        # create the next human input
        func_response_str = '\n\nHuman: Here is the result from your function call\n\n'
        func_response_str = func_response_str + f'<function_results>\n{func_response}\n</function_results>'
        func_response_str = func_response_str + '\n\nIf you know the answer, say it. If not, what is the next step?\n\nAssistant:'

        # augment the prompt
        prompt = prompt + output + func_response_str
    return done, prompt

让我们从第一个示例 What is the weather in Las Vegas? 开始。下面的代码询问 LLM 第一步是什么，我们会注意到 LLM 能够确定它首先需要使用 get_lat_long 工具。

user_input = 'What is the weather in Las Vegas?'
next_step = TOOL_PROMPT.format(tools_string=tools_string, user_input=user_input)

output = invoke_model(next_step).strip()
done, next_step = single_retriever_step(next_step, output)
if not done:
    display(Pretty(f'{output}'))
else:
    display(Pretty('Final answer from LLM:\n'+f'{next_step}'))

很好，Claude 已经确定我们应该首先调用纬度和经度工具。接下来的步骤就像第一步一样进行编排。这一次，Claude 使用第一个请求中的纬度/经度来询问该特定位置的天气。

output = invoke_model(next_step).strip()
done, next_step = single_retriever_step(next_step, output)
if not done:
    display(Pretty(f'{output}'))
else:
    display(Pretty('Final answer from LLM:\n'+f'{next_step}'))

最后，LLM 能够根据上面的输入函数回答这个问题。太棒了!

output = invoke_model(next_step).strip()
done, next_step = single_retriever_step(next_step, output)
if not done:
    display(Pretty(f'{output}'))
else:
    display(Pretty('Final answer from LLM:\n'+f'{next_step}'))

让我们再试一个例子，看看如何在这个例子中使用不同的地方(新加坡)。请注意，我们将 for 循环设置为 5 次迭代，尽管模型只使用了其中的 3 次。这种迭代上限在代理工作流中很常见，应该根据我们的用例进行调整。

user_input = 'What is the weather in Singapore?'
next_step = TOOL_PROMPT.format(tools_string=tools_string, user_input=user_input)

for i in range(5):
    output = invoke_model(next_step).strip()
    done, next_step = single_retriever_step(next_step, output)
    if not done:
        display(Pretty(f'{output}'))
    else:
        display(Pretty('Final answer from LLM:\n'+f'{next_step}'))
        break

下一步

现在我们已经使用了几种不同的检索系统，让我们继续下一个笔记本，我们可以在那里应用我们到目前为止学到的技能!