Ollama-Qwen2，輕鬆搭建支持函數調用的聊天系統
本文介紹如何通過 Ollama 結合 Qwen2，搭建 OpenAI 格式的聊天 API，並與外部函數結合來拓展模型的更多功能。
tools 是 OpenAI 的 Chat Completion API 中的一個可選參數，可用於提供函數調用規範（function specifications）。這樣做的目的是使模型能夠生成符合所提供的規範的函數參數格式。同時，API 實際上不會執行任何函數調用。開發人員需要使用模型輸出來執行函數調用。
Ollama 支持 OpenAI 格式 API 的 tool 參數，在 tool 參數中，如果 functions 提供了參數，Qwen 將會決定何時調用什麼樣的函數，不過 Ollama 目前還不支持強制使用特定函數的參數 tool_choice。
注：本文測試用例參考 OpenAI cookbook：https://cookbook.openai.com/examples/how_to_call_functions_with_chat_models
本文主要包含以下三個部分：
-
模型部署： 使用 Ollama 和千問，通過設置 template，部署支持 Function call 的聊天 API 接口。
-
生成函數參數： 指定一組函數並使用 API 生成函數參數。
-
調用具有模型生成的參數的函數： 通過實際執行具有模型生成的參數的函數來閉合循環。
01 模型部署
單模型文件下載
使用 ModelScope 命令行工具下載單個模型，本文使用 Qwen2-7B 的 GGUF 格式：
modelscope download --model=qwen/Qwen2-7B-Instruct-GGUF --local_dir . qwen2-7b-instruct-q5_k_m.ggufLinux 環境使用
Liunx 用戶可使用魔搭鏡像環境安裝【推薦】
modelscope download --model=modelscope/ollama-linux --local_dir ./ollama-linux
cd ollama-linux
sudo chmod 777 ./ollama-modelscope-install.sh
./ollama-modelscope-install.sh啓動 Ollama 服務
ollama serve創建 ModelFile
複製模型路徑，創建名爲 “ModelFile” 的 meta 文件，其中設置 template，使之支持 function call，內容如下：
FROM /mnt/workspace/qwen2-7b-instruct-q5_k_m.gguf
# set the temperature to 0.7 [higher is more creative, lower is more coherent]
PARAMETER temperature 0.7
PARAMETER top_p 0.8
PARAMETER repeat_penalty 1.05
TEMPLATE """{{ if .Messages }}
{{- if or .System .Tools }}<|im_start|>system
{{ .System }}
{{- if .Tools }}
# Tools
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. Here are the available tools:
<tools>{{- range .Tools }}{{ .Function }}{{- end }}</tools>
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-json-object>}
</tool_call>{{- end }}<|im_end|>{{- end }}
{{- range .Messages }}
{{- if eq .Role "user" }}
<|im_start|>{{ .Role }}
{{ .Content }}<|im_end|>
{{- else if eq .Role "assistant" }}
<|im_start|>{{ .Role }}
{{- if .Content }}
{{ .Content }}
{{- end }}
{{- if .ToolCalls }}
<tool_call>
{{ range .ToolCalls }}{"name": "{{ .Function.Name }}", "arguments": {{ .Function.Arguments }}}
{{ end }}</tool_call>
{{- end }}<|im_end|>
{{- else if eq .Role "tool" }}
<|im_start|>user
<tool_response>
{{ .Content }}
</tool_response><|im_end|>
{{- end }}
{{- end }}
<|im_start|>assistant
{{ else }}{{ if .System }}<|im_start|>system
{{ .System }}<|im_end|>
{{ end }}{{ if .Prompt }}<|im_start|>user
{{ .Prompt }}<|im_end|>
{{ end }}<|im_start|>assistant
{{ end }}
"""創建自定義模型
使用 ollama create 命令創建自定義模型
ollama create myqwen2 --file ./ModelFile運行模型：
ollama run myqwen202 生成函數參數
安裝依賴
!pip install scipy --quiet
!pip install tenacity --quiet
!pip install tiktoken --quiet
!pip install termcolor --quiet
!pip install openai --quiet使用 OpenAI 的 API 格式調用本地部署的 qwen2 模型
import json
import openai
from tenacity import retry, wait_random_exponential, stop_after_attempt
from termcolor import colored
MODEL = "myqwen2"
client = openai.OpenAI(
base_url="http://127.0.0.1:11434/v1",
api_key = "None"
)實用工具
首先，讓我們定義一些實用工具，用於調用聊天完成 API 以及維護和跟蹤對話狀態。
@retry(wait=wait_random_exponential(multiplier=1, max=40), stop=stop_after_attempt(3))
def chat_completion_request(messages, tools=None, tool_choice=None, model=MODEL):
try:
response = client.chat.completions.create(
model=model,
messages=messages,
tools=tools,
tool_choice=tool_choice,
)
return response
except Exception as e:
print("Unable to generate ChatCompletion response")
print(f"Exception: {e}")
return edef pretty_print_conversation(messages):
role_to_color = {
"system": "red",
"user": "green",
"assistant": "blue",
"function": "magenta",
}
for message in messages:
if message["role"] == "system":
print(colored(f"system: {message['content']}\n", role_to_color[message["role"]]))
elif message["role"] == "user":
print(colored(f"user: {message['content']}\n", role_to_color[message["role"]]))
elif message["role"] == "assistant" and message.get("function_call"):
print(colored(f"assistant: {message['function_call']}\n", role_to_color[message["role"]]))
elif message["role"] == "assistant" and not message.get("function_call"):
print(colored(f"assistant: {message['content']}\n", role_to_color[message["role"]]))
elif message["role"] == "function":
print(colored(f"function ({message['name']}): {message['content']}\n", role_to_color[message["role"]]))基本概念（https://cookbook.openai.com/examples/how_to_call_functions_with_chat_models#basic-concepts）
這裏假設了一個天氣 API，並設置了一些函數規範和它進行交互。將這些函數規範傳遞給 Chat API，以便模型可以生成符合規範的函數參數。
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "Get the current weather",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use. Infer this from the users location.",
},
},
"required": ["location", "format"],
},
}
},
{
"type": "function",
"function": {
"name": "get_n_day_weather_forecast",
"description": "Get an N-day weather forecast",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use. Infer this from the users location.",
},
"num_days": {
"type": "integer",
"description": "The number of days to forecast",
}
},
"required": ["location", "format", "num_days"]
},
}
},
]如果我們向模型詢問當前的天氣情況，它將會反問，希望獲取到進一步的更多的參數信息。
messages = []
messages.append({"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."})
messages.append({"role": "user", "content": "hi ，can you tell me what's the weather like today"})
chat_response = chat_completion_request(
messages, tools=tools
)
assistant_message = chat_response.choices[0].message
messages.append(assistant_message)
assistant_messageChatCompletionMessage(content='Of course, I can help with that. To provide accurate information, could you please specify the city and state you are interested in?', role='assistant', function_call=None, tool_calls=None)
一旦我們通過對話提供缺失的參數信息，模型就會爲我們生成適當的函數參數。
messages.append({"role": "user", "content": "I'm in Glasgow, Scotland."})
chat_response = chat_completion_request(
messages, tools=tools
)
assistant_message = chat_response.choices[0].message
messages.append(assistant_message)
assistant_messageChatCompletionMessage(content='', role='assistant', function_call=None, tool_calls=[ChatCompletionMessageToolCall(id='call_qq8e5z9w', function=Function(arguments='{"location":"Glasgow, Scotland"}', name='get_current_weather'), type='function')])
通過不同的提示詞，我們可以讓它反問不同的問題以獲取函數參數信息。
messages = []
messages.append({"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."})
messages.append({"role": "user", "content": "can you tell me, what is the weather going to be like in Glasgow, Scotland in next x days"})
chat_response = chat_completion_request(
messages, tools=tools
)
assistant_message = chat_response.choices[0].message
messages.append(assistant_message)
assistant_messageChatCompletionMessage(content='Sure, I can help with that. Could you please specify how many days ahead you want to know the weather forecast for Glasgow, Scotland?', role='assistant', function_call=None, tool_calls=None)
messages.append({"role": "user", "content": "5 days"})
chat_response = chat_completion_request(
messages, tools=tools
)
chat_response.choices[0]Choice(finish_reason='stop', index=0, logprobs=None, message=ChatCompletionMessage(content='', role='assistant', function_call=None, tool_calls=[ChatCompletionMessageToolCall(id='call_b7f3j7im', function=Function(arguments='{"location":"Glasgow, Scotland","num_days":5}', name='get_n_day_weather_forecast'), type='function')]))
並行函數調用（https://cookbook.openai.com/examples/how_to_call_functions_with_chat_models#parallel-function-calling）
支持一次提問中，並行調用多次函數
messages = []
messages.append({"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."})
messages.append({"role": "user", "content": "what is the weather going to be like in San Francisco and Glasgow over the next 4 days"})
chat_response = chat_completion_request(
messages, tools=tools, model=MODEL
)
assistant_message = chat_response.choices[0].message.tool_calls
assistant_message[ChatCompletionMessageToolCall(id='call_vei89rz3', function=Function(arguments='{"location":"San Francisco, CA","num_days":4}', name='get_n_day_weather_forecast'), type='function'),
ChatCompletionMessageToolCall(id='call_4lgoubee', function=Function(arguments='{"location":"Glasgow, UK","num_days":4}', name='get_n_day_weather_forecast'), type='function')]
使用模型生成函數（https://cookbook.openai.com/examples/how_to_call_functions_with_chat_models#how-to-call-functions-with-model-generated-arguments）
在這個示例中，演示如何執行輸入由模型生成的函數，並使用它來實現可以爲我們解答有關數據庫的問題的代理。
本文使用 Chinook 示例數據庫（https://www.sqlitetutorial.net/sqlite-sample-database/）。
指定執行 SQL 查詢的函數（https://cookbook.openai.com/examples/how_to_call_functions_with_chat_models#specifying-a-function-to-execute-sql-queries）
首先，讓我們定義一些有用的函數來從 SQLite 數據庫中提取數據。
import sqlite3
conn = sqlite3.connect("data/Chinook.db")
print("Opened database successfully")def get_table_names(conn):
"""Return a list of table names."""
table_names = []
tables = conn.execute("SELECT name FROM sqlite_master WHERE type='table';")
for table in tables.fetchall():
table_names.append(table[0])
return table_names
def get_column_names(conn, table_name):
"""Return a list of column names."""
column_names = []
columns = conn.execute(f"PRAGMA table_info('{table_name}');").fetchall()
for col in columns:
column_names.append(col[1])
return column_names
def get_database_info(conn):
"""Return a list of dicts containing the table name and columns for each table in the database."""
table_dicts = []
for table_name in get_table_names(conn):
columns_names = get_column_names(conn, table_name)
table_dicts.append({"table_name": table_name, "column_names": columns_names})
return table_dicts現在可以使用這些實用函數來提取數據庫模式的表示。
database_schema_dict = get_database_info(conn)
database_schema_string = "\n".join(
[
f"Table: {table['table_name']}\nColumns: {', '.join(table['column_names'])}"
for table in database_schema_dict
]
)與之前一樣，我們將爲希望 API 爲其生成參數的函數定義一個函數規範。請注意，我們正在將數據庫模式插入到函數規範中。這對於模型瞭解這一點很重要。
tools = [
{
"type": "function",
"function": {
"name": "ask_database",
"description": "Use this function to answer user questions about music. Input should be a fully formed SQL query.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": f"""
SQL query extracting info to answer the user's question.
SQL should be written using this database schema:
{database_schema_string}
The query should be returned in plain text, not in JSON.
""",
}
},
"required": ["query"],
},
}
}
]執行 SQL 查詢（https://cookbook.openai.com/examples/how_to_call_functions_with_chat_models#executing-sql-queries）
現在讓我們實現實際執行數據庫查詢的函數。
def ask_database(conn, query):
"""Function to query SQLite database with a provided SQL query."""
try:
results = str(conn.execute(query).fetchall())
except Exception as e:
results = f"query failed with error: {e}"
return results使用 Chat Completions API 調用函數的步驟：
（https://cookbook.openai.com/examples/how_to_call_functions_with_chat_models#steps-to-invoke-a-function-call-using-chat-completions-api）
步驟 1：向模型提示可能導致模型選擇要使用的工具的內容。工具的描述（例如函數名稱和簽名）在 “工具” 列表中定義，並在 API 調用中傳遞給模型。如果選擇，函數名稱和參數將包含在響應中。
步驟 2：通過編程檢查模型是否想要調用函數。如果是，則繼續執行步驟 3。
步驟 3：從響應中提取函數名稱和參數，使用參數調用該函數。將結果附加到消息中。
步驟 4：使用消息列表調用聊天完成 API 以獲取響應。
messages = [{
"role":"user",
"content": "What is the name of the album with the most tracks?"
}]
response = client.chat.completions.create(
model='myqwen2',
messages=messages,
tools= tools,
tool_choice="auto"
)
# Append the message to messages list
response_message = response.choices[0].message
messages.append(response_message)
print(response_message)ChatCompletionMessage(content='', role='assistant', function_call=None, tool_calls=[ChatCompletionMessageToolCall(id='call_23nnhlv6', function=Function(arguments='{"query":"SELECT Album.Title FROM Album JOIN Track ON Album.AlbumId = Track.AlbumId GROUP BY Album.Title ORDER BY COUNT(*) DESC LIMIT 1"}', name='ask_database'), type='function')])
# Step 2: determine if the response from the model includes a tool call.
tool_calls = response_message.tool_calls
if tool_calls:
# If true the model will return the name of the tool / function to call and the argument(s)
tool_call_id = tool_calls[0].id
tool_function_name = tool_calls[0].function.name
tool_query_string = json.loads(tool_calls[0].function.arguments)['query']
# Step 3: Call the function and retrieve results. Append the results to the messages list.
if tool_function_name == 'ask_database':
results = ask_database(conn, tool_query_string)
messages.append({
"role":"tool",
"tool_call_id":tool_call_id,
"name": tool_function_name,
"content":results
})
# Step 4: Invoke the chat completions API with the function response appended to the messages list
# Note that messages with role 'tool' must be a response to a preceding message with 'tool_calls'
model_response_with_function_call = client.chat.completions.create(
model="myqwen2",
messages=messages,
) # get a new response from the model where it can see the function response
print(model_response_with_function_call.choices[0].message.content)
else:
print(f"Error: function {tool_function_name} does not exist")
else:
# Model did not identify a function to call, result can be returned to the user
print(response_message.content)The album "Greatest Hits" contains the most tracks
本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/d82jUnXldJw_UPVPngZjDQ