当前位置：首页 > news >正文

大模型及agent开发1——基础知识及实现具备Funcation Calling功能的智能电商客服

news 来源：原创 2025/6/14 7:34:41

路径：agent-rag-微调
一.大模型应用发展及Agent前沿技术趋势
大模型有两个关键概念：原生能力和涌现能力。
涌现能力：这一能力指的是，尽管大模型可能没有直接学习过某些信息，但在与它对话时提供相关信息后，它能够类比和推理出解决方案。
存在着两个主要问题：知识库更新不及时和大模型幻觉问题。

步骤：
Stage 1：提示工程
Stage 2：函数调用：解决了知识库更新不及时、无法获取实时信息及其带来的优势。
Stage 3. Retrieval-Augmented Generation
RAG的实现是包括两个阶段的：检索阶段和生成阶段。
RAG技术解决的两个关键的问题是：
1.删去无用的上下文
2.token长度有限制

后续：AI AGENTS,强化学习
AGENT架构：规划；记忆；行动；工具

二.AI Agent应用类型及Function Calling开发实战
分类：聊天机器人、人工智能助手（能管理家中的智能设备）和人工智能代理（能自主执行多种任务、做出决策和解决问题，并且能够与现实世界互动）

1.聊天机器人：可以构建非常简单的While循环来生成一个持续对话的交互程序，代码如下所示：
from openai import OpenAI
client = OpenAI()

while True:
prompt = input('\n用户提问：')
if prompt == "退出":
break # 如果输入的是“退出”，则结束循环
completion = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
max_tokens=1024,
temperature=0.8)
message = completion.choices[0].message.content
print(f"模型回复:{message}")

问题：单个大模型并不具备记忆和状态跟踪的能力，它仅将每个输入视为一个独立的事件来处理。
我们需要将每次的对话和大模型的回复，不断地追加到传递给 API 的消息字典列表中：
from openai import OpenAI
client = OpenAI()

# 创建一个会话列表来不断地追加历史对话纪律
messages = []

while True:
prompt = input('\n用户提问： ')
if prompt == "退出":
break # 如果输入的是“退出”，则结束循环
messages.append(
{
'role':'user',
'content':prompt
})
completion = client.chat.completions.create(
model="gpt-4o-mini",
messages = messages)

response = completion.choices[0].message.content
print(f"模型回复:{response}")
messages.append(
{
'role':'assistant',
'content':response
})

当然，为了解决大模型知识的局限性问题，我们经常采用RAG技术来增强其信息检索能力。
RAG解决的问题：输入较少的文本去prompt
ChatGPT 不记得您之前的问题。因此，为了与它进行有意义的对话，您需要将对话反馈给 API。还记得需要传递给 API 的字典列表吗？
messages = []

while True:
prompt = input('\n提出一个问题： ')
if prompt == "退出":
break # 如果输入的是“退出”，则结束循环
messages.append(
{
'role':'user',
'content':prompt
})
completion = client.chat.completions.create(
model="gpt-4o-mini",
messages = messages)

response = completion.choices[0].message.content
print(response)
messages.append(
{
'role':'assistant',
'content':response
})

存在的问题是：输出无法给到下游！！！！！！！要解决的根本两点问题就是：

如何让大模型产生结构化的输出
如何让大模型在一次对话中，自动执行多个步骤。

2.人工智能助手
起到的主要作用是替代手工执行特定的任务。
类似智能客服的整个问答流程涉及两个阶段，第一阶段是接收用户的输入，去查询数据库。第二阶段是根据数据库返回的结果，结合用户的问题，生成最终的答案。而这样的阶段性操作，其根本要解决的两点问题是：
如何让大模型产生结构化的输出
如何让大模型在一次对话中，自动执行多个步骤。

使用大语言模型（LLM）通过API进行功能调用的完整生命周期。具体过程如下：
a.API调用：应用程序向API发送调用请求，附带具体的提示信息和可供LLM调用的函数定义。(用的什么模型，传递什么信息，有哪些工具可以调用）
b.模型决策：LLM评估接收到的输入，并决定是否直接回应用户，或是需要调用一个或多个外部函数以提供更合适的回答。
c.API响应：API向应用程序返回响应，指明需要执行的函数以及执行这些函数所需的参数。
d.执行函数：应用程序根据API的指示执行指定的函数，使用提供的参数。
e.调用结果处理：完成函数执行后，应用程序再次调用API，传递先前的提示信息和函数执行的结果，以便LLM可以利用这些新数据生成最终的用户响应。

接下来实操一个智能客服的案例来详细介绍中间环节的详细处理流程。
首先，我们先构造一个商家后台商品信息的模拟数据。这里使用 Python 连接 sqlite3库进行存储。代码如下所示：
import sqlite3

def create_and_populate_database():
# 连接到本地数据库文件
conn = sqlite3.connect('SportsEquipment.db') # 指定文件名来保存数据库
cursor = conn.cursor()

# 检查是否存在名为 'products' 的表，如果不存在则创建
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='products';")
if cursor.fetchone() is None:
# 创建表
cursor.execute('''
CREATE TABLE products (
product_id TEXT,
product_name TEXT,
description TEXT,
specifications TEXT,
usage TEXT,
brand TEXT,
price REAL,
stock_quantity INTEGER
)
''')
# 数据列表，用于插入表中
products = [
('001', '足球', '高品质职业比赛用球，符合国际标准', '圆形，直径22 cm', '职业比赛、学校体育课', '耐克', 120, 50),
('002', '羽毛球拍', '轻量级，适合初中级选手，提供优秀的击球感受', '碳纤维材质，重量85 g', '业余比赛、家庭娱乐', '尤尼克斯', 300, 30),
('003', '篮球', '室内外可用，耐磨耐用，适合各种天气条件', '皮质，标准7号球', '学校、社区运动场', '斯伯丁', 200, 40),
('004', '跑步鞋', '适合长距离跑步，舒适透气，提供良好的足弓支撑', '多种尺码，透气网布', '长跑、日常训练', '阿迪达斯', 500, 20),
('005', '瑜伽垫', '防滑材料，厚度适中，易于携带和清洗', '长180cm，宽60cm，厚5mm', '瑜伽、普拉提', '曼达卡', 150, 25),
('006', '速干运动衫', '吸汗快干，适合各种户外运动，持久舒适', 'S/M/L/XL，多色可选', '运动、徒步、旅游', '诺斯脸', 180, 60),
('007', '电子计步器', '精确计步，带心率监测功能，蓝牙连接手机应用', '可充电，续航7天', '日常健康管理、运动', 'Fitbit', 250, 15),
('008', '乒乓球拍套装', '包括两只拍子和三个球，适合家庭娱乐和业余训练', '标准尺寸，拍面防滑处理', '家庭、社区', '双鱼', 160, 35),
('009', '健身手套', '抗滑耐磨，保护手部，适合各种健身活动', '多种尺码，通风设计', '健身房、户外运动', 'Under Armour', 120, 50),
('010', '膝盖护具', '减少运动伤害，提供良好的支撑和保护，适合篮球和足球运动', '弹性织物，可调节紧度', '篮球、足球及其他运动', '麦克戴维', 220, 40)
]

# 插入数据到表中
cursor.executemany('''
INSERT INTO products (product_id, product_name, description, specifications, usage, brand, price, stock_quantity)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
''', products)

# 提交更改以确保数据被保存在文件中
conn.commit()

# 检索并打印所有记录以验证插入
cursor.execute('SELECT * FROM products')
all_rows = cursor.fetchall()

conn.close() # 关闭连接以释放资源

return all_rows

# 执行函数并打印结果
create_and_populate_database()

第一步：构建大模型能够调用的函数

import sqlite3

def query_by_product_name(product_name):
# 连接 SQLite 数据库
conn = sqlite3.connect('SportsEquipment.db')
cursor = conn.cursor()

# 使用SQL查询按名称查找产品。'%'符号允许部分匹配。
cursor.execute("SELECT * FROM products WHERE product_name LIKE ?", ('%' + product_name + '%',))

# 获取所有查询到的数据
rows = cursor.fetchall()

# 关闭连接
conn.close()

return rows

第二步：向大模型描述这个函数，以便大模型知道如何调用它，以及在什么情况下需要调用它
需要我们进一步创建一个“函数定义”来向模型描述该函数。该定义描述了该函数的作用（以及可能何时调用该函数）以及调用该函数需要哪些参数。
{
"name": "query_by_product_name",
"description": "Query the database to retrieve a list of products that match or contain the specified product name. This function can be used to assist customers in finding products by name via an online platform or customer support interface.", # 查询数据库以检索匹配或包含指定产品名称的产品列表。此功能可用于帮助客户通过在线平台或客户支持界面按名称查找产品。
"parameters": {
"type": "object",
"properties": {
"product_name": {
"type": "string",
"description": "The name of the product to search for. The search is case-insensitive and allows partial matches."
}
},
"required": ["product_name"]
}
}

这段Json Schema 描述的关键要素如下：

name 字段：明确指示大模型需要调用的具体函数名称。
description 字段：向大模型说明在哪些用户需求下应当触发这个函数，即在何种场景下调用。
parameters：详细描述函数所接受的参数，使大模型能够从类似“你家都卖什么球”的自然语言查询中提取出“球”作为关键搜索词。其中的required字段指明哪些参数是必需的，确保函数能在缺少这些参数时提示或防止执行。

第三步：将函数定义作为可用工具以及消息传递给大模型
当涉及到函数调用的时候，我们需要在调用聊天完成 API 时，额外的传递一个 tools参数，以告知大模型：你在当前的会话过程中，可以调用query_by_product_name参数。
tools = [
{
"type": "function",
"function": {
"name": "query_by_product_name",
"description": "Query the database to retrieve a list of products that match or contain the specified product name. This function can be used to assist customers in finding products by name via an online platform or customer support interface.",
"parameters": {
"type": "object",
"properties": {
"product_name": {
"type": "string",
"description": "The name of the product to search for. The search is case-insensitive and allows partial matches."
}
},
"required": ["product_name"]
}
}

}
]
然后：
from openai import OpenAI
client = OpenAI()

response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools, # 这里是添加
)

第四步：接收并处理大模型响应
这里需要注意的是：尽管传递了 tools工具，但并不意味着大模型一定会进行调用
假设用户的问题中提及了关于商品的提问，才会触发工具调用的流程：
messages = [
{"role": "user", "content": "你好，你家都卖什么球？"}
]
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools,
)

response
返回值为：ChatCompletion(id='chatcmpl-A6aaHkwifCmRmxPOKVkVQc5yppiY9', choices=[Choice(finish_reason='tool_calls', index=0, logprobs=None, message=ChatCompletionMessage(content=None, refusal=None, role='assistant', function_call=None, tool_calls=[ChatCompletionMessageToolCall(id='call_6Axa00hfMAVSzSGEBMIjXFZw', function=Function(arguments='{"product_name":"球"}', name='query_by_product_name'), type='function')]))], created=1726133961, model='gpt-4o-mini-2024-07-18', object='chat.completion', service_tier=None, system_fingerprint='fp_483d39d857', usage=CompletionUsage(completion_tokens=17, prompt_tokens=109, total_tokens=126))
这里的区别是：当触发函数调用，content字段的对应的值将会是None，而tool_calls字段中的内容，将会按照我们对query_by_product_name的Json Schema描述中 required字段的要求来返回值。如上所示为：arguments='{"product_name":"球"}', name='query_by_product_name')。当拿到这样的参数后，按照我们期望的逻辑应该是把球作为关键词，执行数据库查询，拿到商品的详细信息，如下图所示：
tool_call = response.choices[0].message.tool_calls[0]
arguments = json.loads(tool_call.function.arguments)
product_name = arguments.get('product_name')
final_res = query_by_product_name(product_name)
得到含球字的那些字段

第五步：将函数调用结果提供回大模型
经过上面的手动流程，现在我们已经在本地执行了函数调用，我们需要将此函数调用的结果提供回聊天 API，以便大模型可以生成用户应该看到的实际响应。参考聊天机器人案例中构建大模型记忆的过程，我们只需要维护一个 message 列表，把函数调用的信息追加进去即可。整个过程如下所示：
这里我们需要在使用 json.dumps() 添加 ensure_ascii=False 参数。这样，JSON 库就会保留中文字符而不是将它们转换为 Unicode 转义序列，如下所示：
content = json.dumps({
"product_name": product_name,
"query_by_product_name":final_res
}, ensure_ascii=False),

# 创建 function calling 结果的消息历史
function_call_result_message = {"role": "tool", "content": str(content), "tool_call_id": response.choices[0].message.tool_calls[0].id}

接下来，我们把这个能够调用外部工具的助手，再次以聊天机器人的形式来实现。最终封装代码如下：
from openai import OpenAI
client = OpenAI()

messages = []

while True:
prompt = input('\n提出一个问题： ')
if prompt.lower() == "退出":
break # 如果输入的是“退出”，则结束循环

# 添加用户的提问到消息列表
messages.append({'role': 'user', 'content': prompt})

# 检查是否需要调用外部函数
completion = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools,
parallel_tool_calls=False # 这里需要格外注意
)

# 提取回答内容
response = completion.choices[0].message
tool_calls = completion.choices[0].message.tool_calls

# 处理外部函数调用
if tool_calls:
function_name = tool_calls[0].function.name
function_args = json.loads(tool_calls[0].function.arguments)

function_response = available_functions[function_name](**function_args)

messages.append(response)

messages.append(
{
"role": "tool",
"name": function_name,
"content": str(function_response),
"tool_call_id": tool_calls[0].id,
}
)

second_response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
)
# 获取最终结果

final_response = second_response.choices[0].message.content
messages.append({'role': 'assistant', 'content': final_response})
print(final_response)
else:
# 打印响应并添加到消息列表
print(response.content)
messages.append({'role': 'assistant', 'content': response.content})

存在的问题：无法接受并行输入（比如同时问几个东西），也就是parallel_tool_calls=False 这句。
3. 并行函数调用
默认情况下，在OpenAI 的大模型生态中，2023 年 11 月 6 日或之后发布的任何模型都可能在单个响应中生成多个函数调用，这说明这类模型可以并行调用某个函数。
我们可以先打印结果看一下
for tool_call in response.choices[0].message.tool_calls:
print(tool_call)

结果：ChatCompletionMessageToolCall(id='call_lMhDOkgreD5ogxG8byFgXRo9', function=Function(arguments='{"product_name": "球"}', name='query_by_product_name'), type='function')
ChatCompletionMessageToolCall(id='call_t9i9HNvhmNSZaa3U454uBWxS', function=Function(arguments='{"product_name": "衣服"}', name='query_by_product_name'), type='function')
ChatCompletionMessageToolCall(id='call_PxcU5AWYGEdUFUwy7iDs8Gfa', function=Function(arguments='{"product_name": "鞋"}', name='query_by_product_name'), type='function')
其中，tool_calls 数组中的每个函数调用都有一个唯一的id ：
而如果我们想在单次的对话中记录每个函数调用的结果，就可以通过向每个函数调用的对话添加一条新消息来将结果提供回模型，每条消息都包含一个函数调用的结果，并使用tool_call_id引用来自的id tool_calls ，如下所示：
product_info = {}

# 遍历工具调用处理每一个产品名称查询
for tool_call in response.choices[0].message.tool_calls:
# 解析调用参数
arguments = json.loads(tool_call.function.arguments)
product_name = arguments['product_name']

# 执行查询并获取结果
query_results = query_by_product_name(product_name)

messages.append({"role": "tool", "content": str(query_results), "tool_call_id": tool_call.id})

# 格式化输出到字典， query_results 返回的列表中包含完整的产品信息
# 提取所需信息，假设每个结果包含 'product_name', 'description', 'price' 等字段
if query_results:
for result in query_results:
product_id, name, description, specifications, usage, brand, price, stock = result
product_info[name] = {
"描述": description,
"规格": specifications,
"适用场合": usage,
"品牌": brand,
"价格": f"{price}元",
"库存数量": stock
}
else:
product_info[product_name] = "未找到相关产品数据"

# 打印整理好的产品信息字典
for product_name, details in product_info.items():
print(f"产品名称：{product_name}")
if isinstance(details, dict):
for detail_key, detail_value in details.items():
print(f"{detail_key}: {detail_value}")
else:
print(details)
print() # 用于在每个产品信息之后添加一个空行以提高可读性

4.多函数调用
多函数调用其实就不是很复杂了。我们只需要新增函数，并且编写具体的函数说明就可以了。比如我们现在接入智能电商客服的第二个功能：可以根据用户对商品的提问查询对应的优化政策，那么接下来我们定义一个read_store_promotions函数根据提供的产品名称来读取具体的优惠政策。代码如下：
def read_store_promotions(product_name):
# 指定优惠政策文档的文件路径
file_path = 'store_promotions.txt'

try:
# 打开文件并按行读取内容
with open(file_path, 'r', encoding='utf-8') as file:
promotions_content = file.readlines()

# 搜索包含产品名称的行
filtered_content = [line for line in promotions_content if product_name in line]

# 返回匹配的行，如果没有找到，返回一个默认消息
if filtered_content:
return ''.join(filtered_content)
else:
return "没有找到关于该产品的优惠政策。"
except FileNotFoundError:
# 文件不存在的错误处理
return "优惠政策文档未找到，请检查文件路径是否正确。"
except Exception as e:
# 其他潜在错误的处理
return f"读取优惠政策文档时发生错误: {str(e)}"