介绍

假设你想买东西。你访问电子商务网站并使用搜索选项找到你想要的东西。也许你要买多件商品，所以这个过程不是很高效。现在考虑这种情况：打开一个应用程序，用简单的英语描述你想要的东西，然后按回车键。你·不必担心搜索和价格比较，因为应用程序会自动为你处理。

我们先来看一些例子。

我将使用 Groq Cloud，特别是本文中使用的模型。此应用程序的初始工作流程应包括一个嵌入模型、一个检索器和两个用于处理用户购买兴趣和成本相关问题的主要工具。总之，我们需要类似于下图中描述的内容。

现在，我们必须使用 LLM 协调框架。为此，我选择了我最喜欢的 Haystack。

加载数据并编制索引

既然我们有了 RAG 管道，第一步就应该建立一个文档索引服务。在本演示中，我将使用 Haystack 提供的内存矢量数据库。请注意，我们的矢量数据库中的每个文档都包含：

• 内容--我们用来执行相似性搜索的内容
• Id - 唯一标识符
• Price - 产品价格
• URL - 产品 URL

当我们的 RAG 管道被调用时，Content 字段将用于矢量搜索。所有其他字段都作为元数据包含在内。保留这些元数据至关重要，因为它们对于前端向用户展示至关重要。

让我们看看如何实现这一点。

from haystack import Pipeline, Document
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.writers import DocumentWriter
from haystack.components.embedders import SentenceTransformersDocumentEmbedder
from haystack.components.generators import OpenAIGenerator
from haystack.utils import Secret
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.builders import PromptBuilder
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.dataclasses import ChatMessage
import pandas as pd
# Load product data from CSV
df = pd.read_csv("product_sample.csv")
# Initialize an in-memory document store
document_store = InMemoryDocumentStore()
# Convert the product data into Haystack Document objects
documents = [
    Document(
        content=item.product_name, 
        meta={
            "id": item.uniq_id, 
            "price": item.selling_price, 
            "url": item.product_url
        }
    ) for item in df.itertuples()
]
# Create a pipeline for indexing the documents
indexing_pipeline = Pipeline()
# Add a document embedder to the pipeline using Sentence Transformers model
indexing_pipeline.add_component(
    instance=SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"), name="doc_embedder"
)
# Add a document writer to the pipeline to store documents in the document store
indexing_pipeline.add_component(instance=DocumentWriter(document_store=document_store), name="doc_writer")
# Connect the embedder's output to the writer's input
indexing_pipeline.connect("doc_embedder.documents", "doc_writer.documents")
# Run the indexing pipeline to process and store the documents
indexing_pipeline.run({"doc_embedder": {"documents": documents}})

很好，我们已经完成了人工智能代理应用程序的第一步。现在是时候构建产品标识符工具了。为了更好地理解产品识别器的主要任务，让我们来看看下面的例子。

用户查询： 我想购买露营靴、木炭和谷歌像素 9 后盖。让我们了解一下产品标识符功能的理想工作流程。

首先，我们需要创建一个工具，用于分析用户查询并识别用户感兴趣的产品。我们可以使用下面的代码片段创建这样一个工具。

构建用户查询分析器

template = """
Understand the user query and list of products the user is interested in and return product names as list.
You should always return a Python list. Do not return any explanation.
Examples:
Question: I am interested in camping boots, charcoal and disposable rain jacket.
Answer: ["camping_boots","charcoal","disposable_rain_jacket"]
Question: Need a laptop, wireless mouse, and noise-cancelling headphones for work.
Answer: ["laptop","wireless_mouse","noise_cancelling_headphones"]
Question: {{ question }}
Answer:
"""
product_identifier = Pipeline()
product_identifier.add_component("prompt_builder", PromptBuilder(template=template))
product_identifier.add_component("llm", generator())
product_identifier.connect("prompt_builder", "llm")

好了，现在我们已经完成了第一个函数的一半，现在是时候通过添加 RAG 管道来完成这个函数了。

创建 RAG 管道

template = """
Return product name, price, and url as a python dictionary. 
You should always return a Python dictionary with keys price, name and url for single product.
You should always return a Python list of dictionaries with keys price, name and url for multiple products.
Do not return any explanation.
Legitimate Response Schema:
{"price": "float", "name": "string", "url": "string"}
Legitimate Response Schema for multiple products:
[{"price": "float", "name": "string", "url": "string"},{"price": "float", "name": "string", "url": "string"}]
Context:
{% for document in documents %}
    product_price: {{ document.meta['price'] }}
    product_url: {{ document.meta['url'] }}
    product_id: {{ document.meta['id'] }}
    product_name: {{ document.content }}
{% endfor %}
Question: {{ question }}
Answer:
"""
rag_pipe = Pipeline()
rag_pipe.add_component("embedder", SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"))
rag_pipe.add_component("retriever", InMemoryEmbeddingRetriever(document_store=document_store, top_k=5))
rag_pipe.add_component("prompt_builder", PromptBuilder(template=template))
rag_pipe.add_component("llm", generator())
rag_pipe.connect("embedder.embedding", "retriever.query_embedding")
rag_pipe.connect("retriever", "prompt_builder.documents")
rag_pipe.connect("prompt_builder", "llm")

完成这一阶段后，我们就完成了 RAG 和查询分析器管道。现在是将其转换为工具的时候了。为此，我们可以使用常规函数声明，如下所示。为 Agent 创建工具就像创建 Python 函数一样。如果你有这样的问题：

Agent 如何调用这个函数？

答案很简单：利用特定于模型的工具模式，我们计划在以后的步骤中加入该模式。现在，是时候创建一个同时使用查询分析器和 RAG 管道的封装函数了。

让我们明确一下该函数的目标。

目标 1：识别用户感兴趣的所有产品，并以列表形式返回。

目标 2：对于每个识别出的产品，从数据库中检索最多五个产品及其元数据。

确定产品标识符功能

def product_identifier_func(query: str):
    """
    Identifies products based on a given query and retrieves relevant details for each identified product.
    Parameters:
    query (str): The query string used to identify products.
    Returns:
    dict: A dictionary where the keys are product names and the values are details of each product. If no products are found, returns "No product found".
    """
    product_understanding = product_identifier.run({"prompt_builder": {"question": query}})
    try:
        product_list = literal_eval(product_understanding["llm"]["replies"][0])
    except:
        return "No product found"
    results = {}
    for product in product_list:
        response = rag_pipe.run({"embedder": {"text": product}, "prompt_builder": {"question": product}})
        try:
            results[product] = literal_eval(response["llm"]["replies"][0])
        except:
            results[product] = {}
    
    return results

这样，我们就完成了代理的第一个工具。让我们看看它是否按预期运行。

query = "I want crossbow and woodstock puzzle"
#execute function
product_identifier_func(query)
# {'crossbow': {'name': 'DB Longboards CoreFlex Crossbow 41" Bamboo Fiberglass '
#                        'Longboard Complete',
#                'price': 237.68,
#                'url': 'https://www.amazon.com/DB-Longboards-CoreFlex-Fiberglass-Longboard/dp/B07KMVJJK7'},
#  'woodstock_puzzle': {'name': 'Woodstock- Collage 500 pc Puzzle',
#                       'price': 17.49,
#                       'url': 'https://www.amazon.com/Woodstock-Collage-500-pc-Puzzle/dp/B07MX21WWX'}}

成功了 不过，值得注意的是返回输出模式。一般模式如下。

{
    "product_key": {
        "name": "string",
        "price": "float",
        "url": "string"
    }
}

这正是我们建议模型在 RAG 管道中生成的结果。下一步，让我们创建一个名为 find_budget_friendly_option 的可选工具。

def find_budget_friendly_option(selected_product_details):
    """
    Finds the most budget-friendly option for each category of products.
    Parameters:
    selected_product_details (dict): A dictionary where the keys are product categories and the values are lists of product details. Each product detail is expected to be a dictionary containing a 'price' key.
    Returns:
    dict: A dictionary where the keys are product categories and the values are the most budget-friendly product details for each category.
    """
    budget_friendly_options = {}
    
    for category, items in selected_product_details.items():
        if isinstance(items, list):
            lowest_price_item = min(items, key=lambda x: x['price'])
        else:
            lowest_price_item = items
        
        budget_friendly_options[category] = lowest_price_item
    
    return budget_friendly_options

好了，让我们把重点放在这个应用最关键的方面，即让代理能够根据需要使用这些功能。正如我们之前谈到的，这可以通过特定于模型的工具模式来实现。因此，我们需要找到特定于所选模型的工具模式。幸运的是，这里的模型卡中提到了它。我们需要对其进行调整，以适应我们的使用情况。

确定聊天模板

chat_template = '''<|start_header_id|>system<|end_header_id|>
You are a function calling AI model. You are provided with function signatures within <tools></tools> XML tags. You may call one or more functions to assist with the user query. Don't make assumptions about what values to plug into functions. For each function call return a json object with function name and arguments within <tool_call></tool_call> XML tags as follows:
<tool_call>
{"name": <function-name>,"arguments": <args-dict>}
</tool_call>
Here are the available tools:
<tools>
    {
        "name": "product_identifier_func",
        "description": "To understand user interested products and its details",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The query to use in the search. Infer this from the user's message. It should be a question or a statement"
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "find_budget_friendly_option",
        "description": "Get the most cost-friendly option. If selected_product_details has morethan one key this should return most cost-friendly options",
        "parameters": {
            "type": "object",
            "properties": {
                "selected_product_details": {
                    "type": "dict",
                    "description": "Input data is a dictionary where each key is a category name, and its value is either a single dictionary with 'price', 'name', and 'url' keys or a list of such dictionaries; example: {'category1': [{'price': 10.5, 'name': 'item1', 'url': 'http://example.com/item1'}, {'price': 8.99, 'name': 'item2', 'url': 'http://example.com/item2'}], 'category2': {'price': 15.0, 'name': 'item3', 'url': 'http://example.com/item3'}}"
                }
            },
            "required": ["selected_product_details"]
        }
    }
</tools><|eot_id|><|start_header_id|>user<|end_header_id|>
I need to buy a crossbow<|eot_id|><|start_header_id|>assistant<|end_header_id|>
<tool_call>
{"id":"call_deok","name":"product_identifier_func","arguments":{"query":"I need to buy a crossbow"}}
</tool_call><|eot_id|><|start_header_id|>tool<|end_header_id|>
<tool_response>
{"id":"call_deok","result":{'crossbow': {'price': 237.68,'name': 'crossbow','url': 'https://www.amazon.com/crossbow/dp/B07KMVJJK7'}}}
</tool_response><|eot_id|><|start_header_id|>assistant<|end_header_id|>
'''

现在只剩下几个步骤了。在做任何事情之前，让我们先测试一下我们的代理。

## Testing agent
messages = [
    ChatMessage.from_system(
        chat_template
    ),
    ChatMessage.from_user("I need to buy a crossbow for my child and Pokémon for myself."),
]
chat_generator = get_chat_generator()
response = chat_generator.run(messages=messages)
pprint(response)
## response
{'replies': [ChatMessage(content='<tool_call>\n'
                                 '{"id": 0, "name": "product_identifier_func", '
                                 '"arguments": {"query": "I need to buy a '
                                 'crossbow for my child"}}\n'
                                 '</tool_call>\n'
                                 '<tool_call>\n'
                                 '{"id": 1, "name": "product_identifier_func", '
                                 '"arguments": {"query": "I need to buy a '
                                 'Pokemon for myself"}}\n'
                                 '</tool_call>',
                         role=<ChatRole.ASSISTANT: 'assistant'>,
                         name=None,
                         meta={'finish_reason': 'stop',
                               'index': 0,
                               'model': 'llama3-groq-70b-8192-tool-use-preview',
                               'usage': {'completion_time': 0.217823967,
                                         'completion_tokens': 70,
                                         'prompt_time': 0.041348261,
                                         'prompt_tokens': 561,
                                         'total_time': 0.259172228,
                                         'total_tokens': 631}})]}

至此，我们已经完成了约 90% 的工作。

你可能已经注意到，在上述响应中，XML 标记 <tool_call> 包含了工具调用。因此，我们需要开发一种机制来提取 tool_call 对象。

def extract_tool_calls(tool_calls_str):
    json_objects = re.findall(r'<tool_call>(.*?)</tool_call>', tool_calls_str, re.DOTALL)
    
    result_list = [json.loads(obj) for obj in json_objects]
    
    return result_list
available_functions = {
    "product_identifier_func": product_identifier_func, 
    "find_budget_friendly_option": find_budget_friendly_option
    }

完成这一步后，我们就可以在代理调用工具时直接访问代理的响应。现在唯一需要做的就是获取工具调用对象并执行相应的函数。让我们也来完成这一步。

messages.append(ChatMessage.from_user(message))
response = chat_generator.run(messages=messages)
if response and "<tool_call>" in response["replies"][0].content:
    function_calls = extract_tool_calls(response["replies"][0].content)
    for function_call in function_calls:
        # Parse function calling information
        function_name = function_call["name"]
        function_args = function_call["arguments"]
        # Find the corresponding function and call it with the given arguments
        function_to_call = available_functions[function_name]
        function_response = function_to_call(**function_args)
        # Append function response to the messages list using `ChatMessage.from_function`
        messages.append(ChatMessage.from_function(content=json.dumps(function_response), name=function_name))
        response = chat_generator.run(messages=messages)

现在是时候将每个组件连接在一起，构建一个合适的聊天应用程序了。为此，我将使用 Gradio。

import gradio as gr
messages = [ChatMessage.from_system(chat_template)]
chat_generator = get_chat_generator()
def chatbot_with_fc(message, messages):
    messages.append(ChatMessage.from_user(message))
    response = chat_generator.run(messages=messages)
    while True:
        if response and "<tool_call>" in response["replies"][0].content:
            function_calls = extract_tool_calls(response["replies"][0].content)
            for function_call in function_calls:
                # Parse function calling information
                function_name = function_call["name"]
                function_args = function_call["arguments"]
                # Find the corresponding function and call it with the given arguments
                function_to_call = available_functions[function_name]
                function_response = function_to_call(**function_args)
                # Append function response to the messages list using `ChatMessage.from_function`
                messages.append(ChatMessage.from_function(content=json.dumps(function_response), name=function_name))
                response = chat_generator.run(messages=messages)
        # Regular Conversation
        else:
            messages.append(response["replies"][0])
            break
    return response["replies"][0].content

def chatbot_interface(user_input, state):
    response_content = chatbot_with_fc(user_input, state)
    return response_content, state
with gr.Blocks() as demo:
    gr.Markdown("# AI Purchase Assistant")
    gr.Markdown("Ask me about products you want to buy!")
    
    state = gr.State(value=messages)
    
    with gr.Row():
        user_input = gr.Textbox(label="Your message:")
        response_output = gr.Markdown(label="Response:")
    
    user_input.submit(chatbot_interface, [user_input, state], [response_output, state])
    gr.Button("Send").click(chatbot_interface, [user_input, state], [response_output, state])

demo.launch()

就是这样！我们已经构建了具有函数调用功能的基于 Llama 3 的人工智能代理。

结论

在构建基于人工智能代理的系统时，考虑完成一项任务所需的时间和每项任务所使用的 API 调用（令牌）数量非常重要。其中一个主要挑战是减少系统中的幻觉，这是一个活跃的研究领域。因此，构建 LLM 和代理系统并没有固定的规则。有必要耐心地、有策略地工作，以确保人工智能代理，即 LLM 正常运行。