安全的關(guān)鍵.png)
使用NestJS和Prisma構(gòu)建REST API：身份驗(yàn)證
這可能看起來(lái)有些復(fù)雜，但本文將通過(guò)示例對(duì)此進(jìn)行詳細(xì)說(shuō)明
在深入研究代碼之前，先簡(jiǎn)單介紹一下演示應(yīng)用程序的架構(gòu)
這里我們?yōu)樵L問(wèn)酒店的游客構(gòu)建一個(gè)助手。助手可以訪問(wèn)以下工具,從而與外部應(yīng)用程序交互:
get_items、purchase_item: 通過(guò)API連接到存儲(chǔ)在數(shù)據(jù)庫(kù)中的產(chǎn)品目錄,用于檢索物品列表和進(jìn)行購(gòu)買rag_pipeline_func: 連接到文檔庫(kù),使用檢索增強(qiáng)生成(RAG)從非結(jié)構(gòu)化文本(如酒店手冊(cè))中獲取信息現(xiàn)在讓我們開(kāi)始吧!
前往 Github 克隆我的代碼。下面的內(nèi)容可以在 function_calling_demo 筆記本中找到。
請(qǐng)創(chuàng)建并激活一個(gè)虛擬環(huán)境,然后運(yùn)行 pip install -r requirements.txt 安裝所需的包。
我們首先連接到 OpenRouter?；蛘?不覆蓋?api_base_url?而直接使用原始的?OpenAIChatGenerator?也可以,前提是你擁有一個(gè) OpenAI API 密鑰。
import os
from dotenv import load_dotenv
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.utils import Secret
from haystack.dataclasses import ChatMessage
from haystack.components.generators.utils import print_streaming_chunk
load_dotenv()
OPENROUTER_API_KEY = os.environ.get('OPENROUTER_API_KEY')
chat_generator = OpenAIChatGenerator(api_key=Secret.from_env_var("OPENROUTER_API_KEY"),
api_base_url="https://openrouter.ai/api/v1",
model="openai/gpt-4-turbo-preview",
streaming_callback=print_streaming_chunk)然后測(cè)試?chat_generator?是否可以成功調(diào)用
chat_generator.run(messages=[ChatMessage.from_user("Return this text: 'test'")])———-輸出應(yīng)如下所示———-
{'replies': [ChatMessage(content="'test'", role=<ChatRole.ASSISTANT: 'assistant'>, name=None, meta={'model': 'openai/gpt-4-turbo-preview', 'index': 0, 'finish_reason': 'stop', 'usage': {}})]}在這里,我們建立應(yīng)用程序與兩個(gè)數(shù)據(jù)源之間的連接:?文檔庫(kù)用于非結(jié)構(gòu)化文本,應(yīng)用程序數(shù)據(jù)庫(kù)通過(guò)API連接。
使用管道索引文檔
我們提供了?documents?中的示例文本,供模型執(zhí)行檢索增強(qiáng)生成(RAG)。這些文本被轉(zhuǎn)換為嵌入并存儲(chǔ)在內(nèi)存中的文檔庫(kù)中。
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
documents = [
Document(content="Coffee shop opens at 9am and closes at 5pm."),
Document(content="Gym room opens at 6am and closes at 10pm.")
]
document_store = InMemoryDocumentStore()
indexing_pipeline = Pipeline()
indexing_pipeline.add_component(
"doc_embedder", SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
)
indexing_pipeline.add_component("doc_writer", DocumentWriter(document_store=document_store))
indexing_pipeline.connect("doc_embedder.documents", "doc_writer.documents")
indexing_pipeline.run({"doc_embedder": {"documents": documents}})它應(yīng)該輸出以下內(nèi)容,對(duì)應(yīng)于我們之前創(chuàng)建的?documents
{'doc_writer': {'documents_written': 2}}啟動(dòng)API服務(wù)器
使用Flask創(chuàng)建了一個(gè)API服務(wù)器?db_api.py?以連接到SQLite。請(qǐng)?jiān)诮K端中運(yùn)行?python db_api.py?啟動(dòng)它
另請(qǐng)注意,?db_api.py?中已添加了一些初始數(shù)據(jù)：
在這里,我們準(zhǔn)備了模型在執(zhí)行函數(shù)調(diào)用后將調(diào)用的實(shí)際函數(shù)(步驟4-5,如函數(shù)調(diào)用結(jié)構(gòu)中所述)
RAG函數(shù)
即?rag_pipeline_func。這用于模型通過(guò)搜索存儲(chǔ)在文檔庫(kù)中的文本來(lái)提供答案。我們首先將RAG檢索定義為一個(gè)Haystack管道。
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
template = """
Answer the questions based on the given context.
Context:
{% for document in documents %}
{{ 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))
rag_pipe.add_component("prompt_builder", PromptBuilder(template=template))
rag_pipe.add_component("llm", OpenAIGenerator(api_key=Secret.from_env_var("OPENROUTER_API_KEY"),
api_base_url="https://openrouter.ai/api/v1",
model="openai/gpt-4-turbo-preview"))
rag_pipe.connect("embedder.embedding", "retriever.query_embedding")
rag_pipe.connect("retriever", "prompt_builder.documents")
rag_pipe.connect("prompt_builder", "llm")測(cè)試函數(shù)是否工作
query = "When does the coffee shop open?"
rag_pipe.run({"embedder": {"text": query}, "prompt_builder": {"question": query}})這應(yīng)該會(huì)產(chǎn)生以下輸出。注意模型給出的?replies?來(lái)自我們之前提供的示例文檔。
{'llm': {'replies': ['The coffee shop opens at 9am.'],
'meta': [{'model': 'openai/gpt-4-turbo-preview',
'index': 0,
'finish_reason': 'stop',
'usage': {'completion_tokens': 9,
'prompt_tokens': 60,
'total_tokens': 69,
'total_cost': 0.00087}}]}}然后我們可以將?rag_pipe?轉(zhuǎn)換為一個(gè)函數(shù),只提供?replies?而不添加其他詳細(xì)信息。
def rag_pipeline_func(query: str):
result = rag_pipe.run({"embedder": {"text": query}, "prompt_builder": {"question": query}})
return {"reply": result["llm"]["replies"][0]}API調(diào)用
我們定義?get_items?和?purchase_item?函數(shù)與數(shù)據(jù)庫(kù)交互。
db_base_url = 'http://127.0.0.1:5000'
import requests
import json
def get_categories():
response = requests.get(f'{db_base_url}/category')
data = response.json()
return data
def get_items(ids=None,categories=None):
params = {
'id': ids,
'category': categories,
}
response = requests.get(f'{db_base_url}/item', params=params)
data = response.json()
return data
def purchase_item(id,quantity):
headers = {
'Content-type':'application/json',
'Accept':'application/json'
}
data = {
'id': id,
'quantity': quantity,
}
response = requests.post(f'{db_base_url}/item/purchase', json=data, headers=headers)
return response.json()定義工具列表
現(xiàn)在我們已經(jīng)定義了函數(shù),需要讓模型識(shí)別這些函數(shù),并指示它們?nèi)绾问褂?通過(guò)為它們提供描述。
由于我們?cè)谶@里使用的是OpenAI,因此?tools?的格式遵循?OpenAI 所需的格式
tools = [
{
"type": "function",
"function": {{
"name": "get_items",
"description": "Get a list of items from the database",
"parameters": {
"type": "object",
"properties": {
"ids": {
"type": "string",
"description": "Comma separated list of item ids to fetch",
},
"categories": {
"type": "string",
"description": "Comma separated list of item categories to fetch",
},
},
"required": [],
},
}
},
{
"type": "function",
"function": {
"name": "purchase_item",
"description": "Purchase a particular item",
"parameters": {
"type": "object",
"properties": {
"id": {
"type": "string",
"description": "The given product ID, product name is not accepted here. Please obtain the product ID from the database first.",
},
"quantity": {
"type": "integer",
"description": "Number of items to purchase",
},
},
"required": [],
},
}
},
{
"type": "function",
"function": {
"name": "rag_pipeline_func",
"description": "Get information from hotel brochure",
"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"],
},
},
}
]我們現(xiàn)在擁有了測(cè)試函數(shù)調(diào)用所需的所有輸入!在這里我們做了以下幾件事:
tools 中向聊天生成器傳遞了工具列表context = f"""You are an assistant to tourists visiting a hotel.
You have access to a database of items (which includes {get_categories()}) that tourists can buy, you also have access to the hotel's brochure.
If the tourist's question cannot be answered from the database, you can refer to the brochure.
If the tourist's question cannot be answered from the brochure, you can ask the tourist to ask the hotel staff.
"""
messages = [
ChatMessage.from_system(context),
ChatMessage.from_user("Can I buy a coffee?"),
]
response = chat_generator.run(messages=messages, generation_kwargs= {"tools": tools})
response———-輸出———-
{'replies': [ChatMessage(content='[{"index": 0, "id": "call_AkTWoiJzx5uJSgKW0WAI1yBB", "function": {"arguments": "{\\"categories\\":\\"Food and beverages\\"}", "name": "get_items"}, "type": "function"}]', role=<ChatRole.ASSISTANT: 'assistant'>, name=None, meta={'model': 'openai/gpt-4-turbo-preview', 'index': 0, 'finish_reason': 'tool_calls', 'usage': {}})]}現(xiàn)在讓我們檢查一下響應(yīng)。注意函數(shù)調(diào)用是如何同時(shí)返回模型選擇的函數(shù)和調(diào)用所選函數(shù)所需的參數(shù)的。
function_call = json.loads(response["replies"][0].content)[0]
function_name = function_call["function"]["name"]
function_args = json.loads(function_call["function"]["arguments"])
print("Function Name:", function_name)
print("Function Arguments:", function_args)———-輸出———-
Function Name: get_items
Function Arguments: {'categories': 'Food and beverages'}當(dāng)提出另一個(gè)問(wèn)題時(shí),模型會(huì)使用更相關(guān)的工具
messages.append(ChatMessage.from_user("Where's the coffee shop?"))
response = chat_generator.run(messages=messages, generation_kwargs= {"tools": tools})
function_call = json.loads(response["replies"][0].content)[0]
function_name = function_call["function"]["name"]
function_args = json.loads(function_call["function"]["arguments"])
print("Function Name:", function_name)
print("Function Arguments:", function_args)———-輸出———-
Function Name: rag_pipeline_func
Function Arguments: {'query': "Where's the coffee shop?"}同樣,請(qǐng)注意這里并沒(méi)有實(shí)際調(diào)用任何函數(shù),這就是我們接下來(lái)要做的!
調(diào)用函數(shù)
我們可以將參數(shù)輸入選擇的函數(shù)
available_functions = {"get_items": get_items, "purchase_item": purchase_item,"rag_pipeline_func": rag_pipeline_func}
function_to_call = available_functions[function_name]
function_response = function_to_call(**function_args)
print("Function Response:", function_response)———-輸出———-
Function Response: {'reply': 'The provided context does not specify a physical location for the coffee shop, only its operating hours. Therefore, I cannot determine where the coffee shop is located based on the given information.'}然后可以將?rag_pipeline_func?的響應(yīng)作為上下文追加到?messages?中,以便模型提供最終答復(fù)。
messages.append(ChatMessage.from_function(content=json.dumps(function_response), name=function_name))
response = chat_generator.run(messages=messages)
response_msg = response["replies"][0]
print(response_msg.content)———-輸出———-
For the location of the coffee shop within the hotel, I recommend asking the hotel staff directly. They will be able to guide you to it accurately.我們現(xiàn)在已經(jīng)完成了一個(gè)聊天周期!
上述代碼展示了如何進(jìn)行函數(shù)調(diào)用,但我們希望進(jìn)一步將其轉(zhuǎn)換為交互式聊天
這里我展示了兩種實(shí)現(xiàn)方式,從在筆記本本身中打印對(duì)話的基本 input() 方法,到通過(guò) Streamlit 渲染以提供類似于 ChatGPT 的 UI。
**input()** 循環(huán)
代碼借鑒自?Haystack 的教程,允許我們快速測(cè)試模型。注意:這個(gè)應(yīng)用程序旨在演示函數(shù)調(diào)用的思想,并非旨在完全健壯,例如不支持同時(shí)訂購(gòu)多個(gè)物品、無(wú)虛構(gòu)等。
import json
from haystack.dataclasses import ChatMessage, ChatRole
response = None
messages = [
ChatMessage.from_system(context)
]
while True:
if response and response["replies"][0].meta["finish_reason"] == "tool_calls":
function_calls = json.loads(response["replies"][0].content)
for function_call in function_calls:
function_name = function_call["function"]["name"]
function_args = json.loads(function_call["function"]["arguments"])```
function_to_call = available_functions[function_name]
function_response = function_to_call(**function_args)
messages.append(ChatMessage.from_function(content=json.dumps(function_response), name=function_name))
else:
if not messages[-1].is_from(ChatRole.SYSTEM):
messages.append(response["replies"][0])
user_input = input("ENTER YOUR MESSAGE ?? INFO: Type 'exit' or 'quit' to stop\n")
if user_input.lower() == "exit" or user_input.lower() == "quit":
break
else:
messages.append(ChatMessage.from_user(user_input))
response = chat_generator.run(messages=messages, generation_kwargs={"tools": tools})
雖然它可以工作,但我們可能希望有一個(gè)更好看的界面。
Streamlit 界面
Streamlit 可以將數(shù)據(jù)腳本轉(zhuǎn)換為可共享的 Web 應(yīng)用程序,為我們的應(yīng)用程序提供了一個(gè)簡(jiǎn)潔的 UI。上面顯示的代碼已經(jīng)適配到我的 repo 中 streamlit 文件夾下的 Streamlit 應(yīng)用程序中。
你可以通過(guò)以下步驟運(yùn)行它:
python db_api.py?啟動(dòng) API 服務(wù)器export OPENROUTER_API_KEY='@替換為你的API密鑰'(假設(shè)你在 Linux 上或使用 git bash 執(zhí)行)streamlit 文件夾,使用 cd streamlitstreamlit run app.py。你的瀏覽器中應(yīng)該會(huì)自動(dòng)打開(kāi)一個(gè)新標(biāo)簽頁(yè),運(yùn)行該應(yīng)用程序就是這樣!希望你喜歡這篇文章。
文章轉(zhuǎn)自微信公眾號(hào)@知覺(jué)之門
對(duì)比大模型API的內(nèi)容創(chuàng)意新穎性、情感共鳴力、商業(yè)轉(zhuǎn)化潛力
一鍵對(duì)比試用API 限時(shí)免費(fèi)
