目标
了解 function 调用、工具调用 和 Agent
function:别用了(OpenAI API has deprecated functions
in favor of tools
)
Tool :能用,但是需要我们自己调用函数,并实现后续逻辑
Agent: 推荐,自动调用我们本地定义的函数
核心思想:通过 llms 来识别用户意图并调用本地函数(Agent)
https://python.langchain.com/docs/modules/agents/agenttypes/openaifunctions_agent
- function call + tools
大模型 的函数调用功能,最早来自 OpenAI .
OpenAI 函数是 OpenAI API 中的一个功能,允许用户通过 API 调用自定义函数。这些函数可以是在 Python 中定义的任何函数,可以接受任何类型的输入,并返回任何类型的输出。这使得用户可以扩展 OpenAI 模型的功能,使其能够执行特定的任务或操作。
OpenAI 函数的接收方式如下:
- 首先,用户需要定义一个函数,并将其作为参数传递给 OpenAI API。这个函数可以接受任何类型的输入,并返回任何类型的输出。
- 然后,用户需要将这个函数传递给 OpenAI API 的 functions 参数。这个参数是一个字典,其中键是函数的名称,值是函数本身。
- 当用户调用 OpenAI API 时,他们可以在请求中指定要调用的函数的名称和参数。OpenAI API 会将这个请求转发给相应的函数,并返回函数的输出。
核心思路
【1】用户描述一个函数调用(通常通过Json Schema 格式的数据构造)
- 函数名称:传给大模型,告诉模型这个函数的名字
- 函数参数:函数有哪些参数,分别是什么意思
function = {
"name": "get_flight_number",
"description": "根据始发地、目的地和日期,查询对应日期的航班号",
"parameters": {
"type": "object",
"properties": {
"departure": {
"description": "出发地",
"type": "string"
},
"destination": {
"description": "目的地",
"type": "string"
},
"date": {
"description": "日期",
"type": "string",
}
},
"required": ["departure", "destination", "date"]
},
}
【2】问题 + 函数描述 传递给模型
#定义模型,通过bind 方法传递给模型
model = AzureChatOpenAI(
openai_api_version="2024-02-15-preview",
azure_deployment=os.getenv('DEPLOYMENT_NAME_GPT3p5'),
temperature=0,
).bind(
function_call={"name": "get_flight_number"}, functions=[function]
)
runnable = {"equation_statement": RunnablePassthrough()} | prompt | model
res = runnable.invoke("后天从北京到南极的飞机票")
【3】大模型进行识别,告诉用户需要执行函数,并返回输入参数
content='' additional_kwargs={'function_call':
{'arguments': '{\n "departure": "北京",\n "destination": "南极",\n "date": "2022-12-10"\n}', 'name': 'get_flight_number'}}
response_metadata={'finish_reason': 'stop', 'logprobs': None, 'content_filter_results': {}}
【4】用户执行函数,并将结果返回给大模型
需要我们自己在代码里处理 如何调用 get flight number, 并继续
函数名
res.additional_kwargs['function_call']['name']
#函数参数
res.additional_kwargs['function_call']['arguments']
get_flight_number(xxx,xxx,xxx)
【5】大模型给出最后回答
需要我们自己在代码里处理 ,如何继续处理get flight number 结果
问题来了
无论是 Function call 还是 Tool, 用户都要自己调用本地函数处理(硬编码在代码里)。
智谱AI举例 function call 举例
定义一个函数:获得航班号,参数:出发地,目的地,日期
{
"type": "function",
"function": {
"name": "get_flight_number",
"description": "根据始发地、目的地和日期,查询对应日期的航班号",
"parameters": {
"type": "object",
"properties": {
"departure": {
"description": "出发地",
"type": "string"
},
"destination": {
"description": "目的地",
"type": "string"
},
"date": {
"description": "日期",
"type": "string",
}
},
"required": ["departure", "destination", "date"]
},
}
},
实现这个函数
def get_flight_number(date:str , departure:str , destination:str):
flight_number = {
"北京":{
"上海" : "1234",
"广州" : "8321",
},
"上海":{
"北京" : "1233",
"广州" : "8123",
}
}
return { "flight_number":flight_number[departure][destination] }
将 函数 + 用户输入 传给模型
messages = []
messages.append({"role": "user", "content": "帮我查询从2024年1月20日,从北京出发前往上海的航班"})
response = client.chat.completions.create(
model="glm-4", # 填写需要调用的模型名称
messages=messages,
tools=tools,
)
print(response.choices[0].message)
messages.append(response.choices[0].message.model_dump())
模型返回
Function(arguments='{"date":"2024-01-20","departure":"北京","destination":"上海"}', name='getflightnumber')
content=None role='assistant' tool_calls=[CompletionMessageToolCall(id='call_8513536572832694738', function=Function(arguments='{"date":"2024-01-20","departure":"北京","destination":"上海"}', name='get_flight_number'), type='function')]
写一个函数,分析这个返回,如果需要执行函数,就执行
def parse_function_call(model_response,messages):
# 处理函数调用结果,根据模型返回参数,调用对应的函数。
# 调用函数返回结果后构造tool message,再次调用模型,将函数结果输入模型
# 模型会将函数调用结果以自然语言格式返回给用户。
if model_response.choices[0].message.tool_calls:
tool_call = model_response.choices[0].message.tool_calls[0]
args = tool_call.function.arguments
function_result = {}
if tool_call.function.name == "get_flight_number":
function_result = get_flight_number(**json.loads(args))
if tool_call.function.name == "get_ticket_price":
function_result = get_ticket_price(**json.loads(args))
messages.append({
"role": "tool",
"content": f"{json.dumps(function_result)}",
"tool_call_id":tool_call.id
})
response = client.chat.completions.create(
model="glm-4", # 填写需要调用的模型名称
messages=messages,
tools=tools,
)
print(response.choices[0].message)
messages.append(response.choices[0].message.model_dump())
执行完函数之后,需要把结果反馈给模型。最后,模型给出结果
content='根据您的查询,我已经帮您找到了2024年1月20日从北京出发前往上海的航班,航班号为1234。' role='assistant' tool_calls=None
然后我们再把结构,保存在对话列表
messages.append(response.choices[0].message.model_dump())
在Langchian中调用 function
function = {
"name": "get_flight_number",
"description": "根据始发地、目的地和日期,查询对应日期的航班号",
"parameters": {
"type": "object",
"properties": {
"departure": {
"description": "出发地",
"type": "string"
},
"destination": {
"description": "目的地",
"type": "string"
},
"date": {
"description": "日期",
"type": "string",
}
},
"required": ["departure", "destination", "date"]
},
}
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"你是飞机票务信息提供商",
),
("human", "{equation_statement}"),
]
)
# 函数调用
model = AzureChatOpenAI(
openai_api_version="2024-02-15-preview",
azure_deployment=os.getenv('DEPLOYMENT_NAME_GPT3_4K_JP'),
temperature=0,
).bind(
function_call={"name": "get_flight_number"}, functions=[function]
)
runnable = {"equation_statement": RunnablePassthrough()} | prompt | model
res = runnable.invoke("后天从北京到南极的飞机票")
print(res)
输出
content='' additional_kwargs={'function_call': {'arguments': '{\n "departure": "北京",\n "destination": "南极",\n "date": "2022-03-06"\n}', 'name': 'get_flight_number'}} response_metadata={'finish_reason': 'stop', 'logprobs': None}
在Langchian中调用 Tool
工具调用
tools = [
{
"type": "function",
"function": {
"name": "get_flight_number",
"description": "根据始发地、目的地和日期,查询对应日期的航班号",
"parameters": {
"type": "object",
"properties": {
"departure": {
"description": "出发地",
"type": "string"
},
"destination": {
"description": "目的地",
"type": "string"
},
"date": {
"description": "日期",
"type": "string",
}
},
"required": ["departure", "destination", "date"]
},
},
},
]
# Azure Openai
os.environ["AZURE_OPENAI_API_KEY"] = os.getenv('MY_AZURE_OPENAI_API_KEY')
os.environ["AZURE_OPENAI_ENDPOINT"] = os.getenv('MY_AZURE_OPENAI_ENDPOINT')
DEPLOYMENT_NAME_GPT3P5 = os.getenv('MY_DEPLOYMENT_NAME_GPT3P5')
model = AzureChatOpenAI(
openai_api_version="2023-05-15",
azure_deployment=DEPLOYMENT_NAME_GPT3P5,
).bind(
tools=tools
)
runnable = {"equation_statement": RunnablePassthrough()} | prompt | model
res = runnable.invoke("后天从北京到南极的飞机票")
# res = runnable.invoke("后天从北京到上海的火车票")
pass
输出
content='' additional_kwargs={'tool_calls': [{'id': 'call_1lP2S9SFi9FDdhYZNkXVxpqt', 'function': {'arguments': '{\n"departure": "北京",\n"destination": "南极",\n"date": "2022-11-18"\n}', 'name': 'get_flight_number'}, 'type': 'function'}]} response_metadata={'finish_reason': 'tool_calls', 'logprobs': None}
Agent (全自动)
代理的核心思想是使用语言模型来选择要采取的一系列行动。在链式中,一系列行动是硬编码在代码中的。而在代理中,语言模型被用作推理引擎,以确定要采取哪些行动以及它们的顺序。
函数声明: @tool
from langchain.agents import tool,
@tool
def get_flight_number(departure, destination, date) -> str:
"""根据输入的出发地(departure),目的地(destination),时间(date),给出航班号"""
return "[茉卷航空666]"
构建chain
user_template = '{input}'
prompt = ChatPromptTemplate.from_messages(
[
("system", "You are a helpful assistant"),
MessagesPlaceholder("chat_history", optional=True),
("human", user_template),
MessagesPlaceholder("agent_scratchpad"),
]
)
model = AzureChatOpenAI(
openai_api_version="2024-02-15-preview",
azure_deployment=os.getenv('DEPLOYMENT_NAME_GPT3P5'),
)
调用 Agent
tool_list = [get_flight_number]
agent = create_openai_functions_agent(model, tool_list, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tool_list, verbose=True)
res = agent_executor.invoke({"input": "后天从北京到南极的飞机票"})
结果
Entering new AgentExecutor chain...
Invoking:
get_flight_number
with
{'departure': '北京', 'destination': '南极', 'date': '后天'}
[茉卷航空666]根据查询,后天从北京到南极的飞机票的航班号是[茉卷航空666]。
Finished chain.
完整代码
zhipuai function call
import json
import os
from zhipuai import ZhipuAI
def get_flight_number(date:str , departure:str , destination:str):
flight_number = {
"北京":{
"上海" : "1234",
"广州" : "8321",
},
"上海":{
"北京" : "1233",
"广州" : "8123",
}
}
return { "flight_number":flight_number[departure][destination] }
def get_ticket_price(date:str , flight_number:str):
return {"ticket_price": "1000"}
def parse_function_call(model_response,messages):
# 处理函数调用结果,根据模型返回参数,调用对应的函数。
# 调用函数返回结果后构造tool message,再次调用模型,将函数结果输入模型
# 模型会将函数调用结果以自然语言格式返回给用户。
if model_response.choices[0].message.tool_calls:
tool_call = model_response.choices[0].message.tool_calls[0]
args = tool_call.function.arguments
function_result = {}
if tool_call.function.name == "get_flight_number":
function_result = get_flight_number(**json.loads(args))
if tool_call.function.name == "get_ticket_price":
function_result = get_ticket_price(**json.loads(args))
messages.append({
"role": "tool",
"content": f"{json.dumps(function_result)}",
"tool_call_id":tool_call.id
})
response = client.chat.completions.create(
model="glm-4", # 填写需要调用的模型名称
messages=messages,
tools=tools,
)
print(response.choices[0].message)
messages.append(response.choices[0].message.model_dump())
if
name
== '
__main__
':
client = ZhipuAI(api_key=os.getenv('MY_ZHIPUAI_API_KEY'))
messages = []
tools = [
{
"type": "function",
"function": {
"name": "get_flight_number",
"description": "根据始发地、目的地和日期,查询对应日期的航班号",
"parameters": {
"type": "object",
"properties": {
"departure": {
"description": "出发地",
"type": "string"
},
"destination": {
"description": "目的地",
"type": "string"
},
"date": {
"description": "日期",
"type": "string",
}
},
"required": ["departure", "destination", "date"]
},
}
},
{
"type": "function",
"function": {
"name": "get_ticket_price",
"description": "查询某航班在某日的票价",
"parameters": {
"type": "object",
"properties": {
"flight_number": {
"description": "航班号",
"type": "string"
},
"date": {
"description": "日期",
"type": "string",
}
},
"required": ["flight_number", "date"]
},
}
},
]
messages = []
messages.append({"role": "user", "content": "帮我查询从2024年1月20日,从北京出发前往上海的航班"})
response = client.chat.completions.create(
model="glm-4", # 填写需要调用的模型名称
messages=messages,
tools=tools,
)
print(response.choices[0].message)
messages.append(response.choices[0].message.model_dump())
pass
parse_function_call(response, messages)
pass
langchain: function call, tool, agent
import json
import os
import uuid
from langchain.agents.output_parsers import XMLAgentOutputParser
from langchain.chains.llm import LLMChain
from langchain_community.chat_models.azure_openai import AzureChatOpenAI
from langchain_community.chat_models.baidu_qianfan_endpoint import QianfanChatEndpoint
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough
from langchain import hub
from langchain.agents import AgentExecutor, tool, create_openai_functions_agent
run_uid = uuid.uuid4().hex[:6]
os.environ["LANGCHAIN_PROJECT"] = f" [Agent call locat func] Tracing Walkthrough - {run_uid}"
os.environ["LANGCHAIN_TRACING_V2"] = 'true'
os.environ["LANGCHAIN_API_KEY"] = os.getenv('MY_LANGCHAIN_API_KEY')
if
name
== '
__main__
':
function = {
"name": "get_flight_number",
"description": "根据始发地、目的地和日期,查询对应日期的航班号",
"parameters": {
"type": "object",
"properties": {
"departure": {
"description": "出发地",
"type": "string"
},
"destination": {
"description": "目的地",
"type": "string"
},
"date": {
"description": "日期",
"type": "string",
}
},
"required": ["departure", "destination", "date"]
},
}
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"你是飞机票务信息提供商",
),
("human", "{equation_statement}"),
]
)
# 函数调用
# Azure Openai
os.environ["AZURE_OPENAI_API_KEY"] = os.getenv('MY_AZURE_OPENAI_API_KEY')
os.environ["AZURE_OPENAI_ENDPOINT"] = os.getenv('MY_AZURE_OPENAI_ENDPOINT')
DEPLOYMENT_NAME_GPT3P5 = os.getenv('MY_DEPLOYMENT_NAME_GPT3P5')
model = AzureChatOpenAI(
openai_api_version="2023-05-15",
azure_deployment=DEPLOYMENT_NAME_GPT3P5,
).bind(
function_call={"name": "get_flight_number"}, functions=[function]
)
runnable = {"equation_statement": RunnablePassthrough()} | prompt | model
# res = runnable.invoke("后天从北京到南极的飞机票")
# print(res)
pass
# 工具调用
tools = [
{
"type": "function",
"function": {
"name": "get_flight_number",
"description": "根据始发地、目的地和日期,查询对应日期的航班号",
"parameters": {
"type": "object",
"properties": {
"departure": {
"description": "出发地",
"type": "string"
},
"destination": {
"description": "目的地",
"type": "string"
},
"date": {
"description": "日期",
"type": "string",
}
},
"required": ["departure", "destination", "date"]
},
},
},
{
"type": "function",
"function": {
"name": "get_train_number",
"description": "根据始发地、目的地和日期,查询对应日期的火车票",
"parameters": {
"type": "object",
"properties": {
"departure": {
"description": "出发地",
"type": "string"
},
"destination": {
"description": "目的地",
"type": "string"
},
"date": {
"description": "日期",
"type": "string",
}
},
"required": ["departure", "destination", "date"]
},
},
}
]
# Azure Openai
os.environ["AZURE_OPENAI_API_KEY"] = os.getenv('MY_AZURE_OPENAI_API_KEY')
os.environ["AZURE_OPENAI_ENDPOINT"] = os.getenv('MY_AZURE_OPENAI_ENDPOINT')
DEPLOYMENT_NAME_GPT3P5 = os.getenv('MY_DEPLOYMENT_NAME_GPT3P5')
model = AzureChatOpenAI(
openai_api_version="2023-05-15",
azure_deployment=DEPLOYMENT_NAME_GPT3P5,
).bind(
tools=tools
)
runnable = {"equation_statement": RunnablePassthrough()} | prompt | model
# res = runnable.invoke("后天从北京到南极的飞机票")
# res = runnable.invoke("后天从北京到上海的火车票")
pass
# 如果需要执行函数
# if len(res.additional_kwargs) > 0:
#
# func_name = res.additional_kwargs['tool_calls'][0]['function']['name']
# if func_name == 'get_flight_number':
# data_str = res.additional_kwargs['tool_calls'][0]['function']['arguments']
# # 自己加
# pass
# Agent
user_template = '{input}'
prompt = ChatPromptTemplate.from_messages(
[
("system", "You are a helpful assistant"),
MessagesPlaceholder("chat_history", optional=True),
("human", user_template),
MessagesPlaceholder("agent_scratchpad"),
]
)
# Azure Openai
os.environ["AZURE_OPENAI_API_KEY"] = os.getenv('MY_AZURE_OPENAI_API_KEY')
os.environ["AZURE_OPENAI_ENDPOINT"] = os.getenv('MY_AZURE_OPENAI_ENDPOINT')
DEPLOYMENT_NAME_GPT3P5 = os.getenv('MY_DEPLOYMENT_NAME_GPT3P5')
model = AzureChatOpenAI(
openai_api_version="2023-05-15",
azure_deployment=DEPLOYMENT_NAME_GPT3P5,
)
@tool
def get_flight_number(departure, destination, date) -> str:
"""根据输入的出发地(departure),目的地(destination),时间(date),给出航班号"""
return "[茉卷航空666]"
tool_list = [get_flight_number]
agent = create_openai_functions_agent(model, tool_list, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tool_list, verbose=True)
res = agent_executor.invoke({"input": "后天从北京到南极的飞机票"})
pass