帶你拆解、實現(xiàn)一套Agent底層框架
Agent 平臺底層框架是怎么樣的?
底層開發(fā)
我們應(yīng)該將這一萬個 Agent 視為一個統(tǒng)一的 Agent 實體。這個統(tǒng)一的 Agent 可以通過提示詞進行重新定義。本質(zhì)上,它就是一個基于大型模型的聊天應(yīng)用程序。
關(guān)鍵在于,我們需要一種機制,將我們的業(yè)務(wù)執(zhí)行邏輯嵌入到聊天的過程中。
圖片
有的人可能會想,讓大模型直接調(diào)用我們的業(yè)務(wù)邏輯不就行了嗎?從廣義上來說,這樣實現(xiàn)也沒問題,只不過我們這里說的大模型指的是只有文本聊天能力的大模型。所以還需要一個“助理”角色代理執(zhí)行。
意圖識別
那怎么才能實現(xiàn)這個核心的代理層呢?我們先從一個簡單的例子開始說。用過 GPT 的朋友可能都有下面的體驗,當我們要求 GPT 基于互聯(lián)網(wǎng)已有知識回答問題,它會怎么做呢?
圖片
你看,我問了一個問題,GPT 搜索了 7 個網(wǎng)站,然后根據(jù)這些網(wǎng)站的內(nèi)容做出了回答。但是顯然大模型是不具備網(wǎng)絡(luò)搜索能力的。那它是怎么做的呢?答案是我們提問里的 請你先查詢網(wǎng)絡(luò)資料再回答我 這個意圖必須被識別出來,并改變原有流程,它才能做到執(zhí)行搜索操作。
圖片
當然,請你先查詢網(wǎng)絡(luò)資料再回答我 這句話還有別的表述方式,比如下面的這些。
請先通過網(wǎng)絡(luò)搜索相關(guān)資料,然后再回答我的問題。
先查閱一些網(wǎng)絡(luò)資源,再給我詳細解答。
請先在網(wǎng)上找一些資料,然后再為我解答。
先在互聯(lián)網(wǎng)搜集相關(guān)信息,然后再回答我。
請先查找網(wǎng)絡(luò)上的相關(guān)內(nèi)容,再來回答我的問題。
請先從網(wǎng)上獲取相關(guān)信息,然后再給出答案。
先通過網(wǎng)絡(luò)查詢一下,再給我提供答案。
請先在網(wǎng)上找到一些相關(guān)資料,再來回答我的問題。
請先通過網(wǎng)絡(luò)查詢相關(guān)內(nèi)容,然后再回答我。
請先利用網(wǎng)絡(luò)搜索相關(guān)資料,然后再給出解答。在設(shè)計意圖識別模塊時,有一個關(guān)鍵問題不容忽視,那就是系統(tǒng)必須能夠理解用戶可能使用的各種不同表述方式。與此同時,意圖識別也不僅僅是單輪輸入的匹配,更需要具備對上下文的感知能力。畢竟,用戶的真實意圖往往不會直接說出來,而是埋藏在多次對話的語境之中。
當用戶的搜索意圖逐漸被識別出來以后,代理便能夠據(jù)此調(diào)起合適的插件或工作流。實際上,不論是大模型所具備的插件功能,還是 Agent 平臺上的插件能力與工作流,本質(zhì)上都沒有區(qū)別。它們的共同點,都是為了讓大模型能夠獲得額外的程序執(zhí)行能力,從而完成更加復(fù)雜的任務(wù)。
不過,相比單純的意圖識別,還有一個更重要的環(huán)節(jié),那就是參數(shù)的提取。就像在之前討論過的營銷 AI 場景里,當助理與用戶進行聊天時,代理層不僅要判斷出該調(diào)用哪條工作流,更要從對話的上下文中提煉出關(guān)鍵的參數(shù)。比如,在那個案例中,最重要的參數(shù)就是公司的名稱。
那上面例子里的網(wǎng)絡(luò)讀取插件,輸入?yún)?shù)是什么呢?你可以想想。
圖片
我想你也注意到了,要實現(xiàn)這個 Agent 的底層,最核心的就是意圖識別和參數(shù)識別。需要注意,這兩個能力都是大模型具備的,我們實現(xiàn)的 Agent 底層也是用大模型來做意圖識別和參數(shù)識別。
Agent 只是一個公司的客服角色,它的職責是滿足客戶的需求,并負責溝通,工作流則像公司內(nèi)部的部門,他們并不直接對外。
技術(shù)實現(xiàn)
后,我們說技術(shù)實現(xiàn)。意圖識別和參數(shù)識別的具體實現(xiàn)方法有兩種,可以通過微調(diào)大模型實現(xiàn),也可以通過提示詞實現(xiàn),思想都是想通的。
下面是一個 Demo 級別的核心架構(gòu)代碼演示,這個代碼就是上述 Agent 底層架構(gòu)的核心邏輯實現(xiàn)例子。這個 Demo 以一個用餐業(yè)務(wù)為例,分為提示詞和核心代碼兩個部分。理解了這個例子,剩下的一萬個 Agent 也就理解了。
首先是核心提示詞。
現(xiàn)在有3個角色,sys 代表應(yīng)用程序系統(tǒng),user 代表用戶,你的角色是一名餐廳服務(wù)員,你的目的是服務(wù)用戶,請你根據(jù)上下文決定是否調(diào)用 sys應(yīng)用程序系統(tǒng)。
調(diào)用系統(tǒng)則輸出 to_sys: 應(yīng)用程序參數(shù)
回復(fù)用戶則輸出 to_user: 回復(fù)的消息
規(guī)則:
1,要一句一句的和我溝通,每次只能選擇和to_sys或to_user其中之一,我會代為轉(zhuǎn)達和處理消息,并且用sys 或 user開頭的消息回復(fù)你
2,調(diào)用程序的具體參數(shù)要在上下文中確定,比如我們應(yīng)用程序有個參數(shù)a,上下文里用戶或系統(tǒng)沒有提供,你就不能擅自決定a的具體數(shù)值,調(diào)用程序用json格式
3,當我用sys:開頭給你回復(fù)的時候,表示本次應(yīng)用程序返回,你要結(jié)合這個返回信息繼續(xù)和用戶溝通,你直接用to_user和用戶溝通,要注意影藏程序系統(tǒng)這個信息,讓用戶感知不到
反例:
1,類似下面的回復(fù)是錯誤的,錯誤原因是在消息里同時出現(xiàn)to_sys 和 to_user
"""
to_sys: {
‘a(chǎn)pp’ : '點餐應(yīng)用',
'usernum' : 1,
'caiming' : '包子',
'cainum' : 3
}
to_user: 好的,三個包子已經(jīng)為您下單了,稍等片刻就會上菜。是否還需要點其他菜品或者飲料呢?
"""
[應(yīng)用程序列表如下]
1,點餐應(yīng)用,當用戶點餐時調(diào)用,
請你和用戶溝通確定這些參數(shù):用戶人數(shù),菜名(必填),數(shù)量(必填)
輸出json格式例子:{
‘a(chǎn)pp’ : '點餐應(yīng)用', #應(yīng)用名字(必填)
'usernum' : 1, #用戶人數(shù)
'caiming' : '包子', #菜名
'cainum' : 5 #數(shù)量
}
2,菜單應(yīng)用,在點餐之前調(diào)用,
參數(shù):無
輸出json格式例子:{
‘a(chǎn)pp’ : '菜單應(yīng)用', #應(yīng)用名字(必填)
}
3,結(jié)賬應(yīng)用,當用戶用完餐需要結(jié)賬時調(diào)用,
參數(shù):無
輸出json格式例子:{
‘a(chǎn)pp’ : '結(jié)賬應(yīng)用', #應(yīng)用名字(必填)
}
接下來用to_user給用戶開始第一條消息
你的餐廳名字是: 成都小吃我們可分析一下這個核心提示詞,有三個層次。
其一是定義角色,提示詞里表明了代理、系統(tǒng)、用戶三者的關(guān)系,讓大模型根據(jù)上下文來決定具體的回復(fù),這也就解決了意圖識別的問題。其二是對 sys 類參數(shù)的限定提示詞,讓大模型按插件和工作流的參數(shù)格式來回答。其三是將應(yīng)用的工作流和具體參數(shù)注入提示詞,讓大模型理解應(yīng)用。
這段提示詞的作用就是告訴大模型我們的具體場景以及助理的具體能力,將上下文交給它處理。需要注意的是,各個插件和工作流的參數(shù)都是大模型來組織的,我們給于引導即可。
和提示詞對應(yīng)的 Demo 程序代碼如下。
import json
def deal_app(app, params):
print(">>> " + app)
if app == "菜單應(yīng)用":
return '菜單 1包子 2餃子 3 可樂或雪碧'
if app == "結(jié)賬應(yīng)用":
return '一共消費100元'
if app == "點餐應(yīng)用":
if 'caiming' not in params or params['caiming'] == None:
return "請先選擇菜品和數(shù)量"
if params['caiming'] == '面包':
return '面包賣完了'
return "已下單"
return "沒有這個應(yīng)用"
def deal_answer(answer):
if answer.find("to_user:") >= 0 and answer.find("to_sys:") >= 0:
print(answer)
return xf_ai('user', "請不要同時回復(fù)to_user 和 to_sys, 兩者只能選一個,也不要[等待系統(tǒng)回復(fù)]這樣的說明", "sys:")
if answer.find("to_sys:") >= 0 and answer.find("\nsys:") >= 0:
print(answer)
return xf_ai('user', "請不要同時回復(fù)to_sys 和 sys, 只回復(fù)to_sys的數(shù)據(jù)", "sys:")
if answer.find("to_user:") == 0:
return answer.replace("to_user:", "")
if answer.find("to_sys:") == 0:
# print(answer)
j = answer.replace("to_sys:", "")
j = j.strip()
end = j.find("}")
j = j[0:end+1]
print(j)
jj = json.loads(j)
app = jj['app']
s = deal_app(app, jj)
o = xf_ai('user', s, "sys:請告訴用戶:")
return o
# print(answer)
return ""
def get_input_from_command_line():
"""
Get input string from command line using input function.
"""
input_string = input("用戶: ")
return input_string
def out_user_message(s):
if s == False:
return
print('服務(wù)員:' + s)
if __name__ == '__main__':
o = xf_ai('user', prompt)
out_user_message(o)
while True:
s = get_input_from_command_line()
if s == 'quit':
exit()
o = xf_ai('user', s, "user:")
out_user_message(o)這個程序就是代理層的核心邏輯代碼了。整體的思想就是圍繞核心提示詞構(gòu)建處理流程,在一般情況下直接回答用戶的問題,在工作流狀態(tài)下調(diào)用工作流邏輯,根據(jù)返回信息再讓模型組織語言回復(fù)用戶,完成整個交互。
下面是它核心程序流程圖。
圖片
千萬不要小看這個提示詞和底層 Demo,實際上麻雀雖小,五臟俱全。我們整個系統(tǒng)都是在這個基礎(chǔ)代碼上構(gòu)建的,你可以把這部分的實現(xiàn)單獨作為一個實驗來做,一定會對大模型開發(fā)有更深入的體會。
如果要更好地理解整套 Agent 系統(tǒng),我想可以從我們最終的助理創(chuàng)建界面繼續(xù)給你分析。
圖片
我們的創(chuàng)建助理功能實際分為 5 步,每一步都必須完成,這是和 Coze 這類平臺不同的,其中最重要的一步就是給每個助理配一個知識庫。
應(yīng)用開發(fā)
這里我還是會舉一個具體的應(yīng)用例子,因為我們的底層 demo 是以餐館菜單為基礎(chǔ)實現(xiàn)的,所以這里我們就繼續(xù)擴展這個案例,開發(fā)一個黃燜雞點餐大師。
知識庫插件
那么顯然,菜單數(shù)據(jù)是它的核心。實際上大部分應(yīng)用都會用到數(shù)據(jù)庫,這也是我們?yōu)槭裁唇o每個助理都配置知識庫的原因。
圖片
實現(xiàn)這套客戶自助配置,如何跟現(xiàn)有系統(tǒng)對接呢?需要分為兩部分。其一是數(shù)據(jù)層面的自助訓練,其二是業(yè)務(wù)邏輯層面的工作流移植。
在數(shù)據(jù)層面,我們提供了知識庫自助訓練,這樣每個客戶都可以擁有一個自己定制化的小模型,保證數(shù)據(jù)的私有屬性。
以一個黃燜雞餐館菜單的數(shù)據(jù)為例,我們看一下私有知識庫的搭建流程。
第一步,將菜單數(shù)據(jù)整理成固定的輸入格式,這一步比較簡單。
{
"1": {"name": "經(jīng)典黃燜雞", "description": "雞肉鮮嫩,配以濃郁醬汁和土豆", "price": "12元", "category": "主菜"},
"2": {"name": "辣味黃燜雞", "description": "經(jīng)典黃燜雞基礎(chǔ)上加入辣椒調(diào)味", "price": "13元", "category": "主菜"},
"3": {"name": "香菇黃燜雞", "description": "黃燜雞加入香菇,風味獨特", "price": "14元", "category": "主菜"}
}第二步,將菜單向量化,將類似 item["description"] 這樣的本字段轉(zhuǎn)化為向量表示,具體偽代碼如下。
from transformers import BertModel, BertTokenizer
import torch
# 使用BERT模型
tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
model = BertModel.from_pretrained('bert-base-uncased')
def get_embedding(texts):
embeddings = []
for text in texts:
inputs = tokenizer(text, return_tensors='pt')
outputs = model(**inputs)
vector = outputs.last_hidden_state.mean(dim=1).detach().numpy()
embeddings.append(vector)
return embeddings
# 讀取數(shù)據(jù)并向量化
with open('huangmenji_menu.json', 'r') as f:
menu_data = json.loads(f.read())
menu_descriptions = [item["description"] for item in menu_data.values()]
menu_embeddings = get_embedding(menu_descriptions)第三步,存儲和讀取向量數(shù)據(jù)庫,這里以 Milvus向量數(shù)據(jù)庫 為例,注意要將菜單信息和向量信息一起存儲,具體偽代碼如下。
from pymilvus import MilvusClient, FieldSchema, CollectionSchema, DataType, Collection
import numpy as np
# 連接到Milvus
client = MilvusClient(uri="http://localhost:19530", db_name="default")
# 定義集合schema
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=False),
FieldSchema(name="name", dtype=DataType.VARCHAR, max_length=100),
FieldSchema(name="description", dtype=DataType.VARCHAR, max_length=1000),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768)
]
schema = CollectionSchema(fields, "黃燜雞餐館菜單")
index_params = client.prepare_index_params()
index_params.add_index(
field_name="embedding",
index_type="IVF_FLAT",
metric_type="IP",
params={"nlist": 128}
)
# 創(chuàng)建集合
collection_name = "huangmenji_menu"
client.create_collection(
collection_name=collection_name,
schema=schema,
index_params=index_params
)
# 插入數(shù)據(jù)
entities = [
{"id": int(item_id),
"name": menu_data[item_id]["name"],
"description": menu_data[item_id]["description"],
"embedding": menu_embeddings[int(item_id)-1].tolist()}
for item_id in menu_data
]
client.insert(collection_name=collection_name, data=entities)整個模塊最核心的一行代碼其實就是這句 FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768)。你可以把它理解為傳統(tǒng)數(shù)據(jù)庫里的索引。加上這個字段后,后續(xù)工作流中需要用到菜單時都可以用向量查詢,具體查詢類似下面的偽代碼。
# 查詢向量化輸入
query = "我喜歡吃辣,有什么菜品推薦"
query_embedding = get_embedding([query])
# 搜索相似的菜單項
res = client.search(
collection_name=collection_name,
data=query_embedding,
limit=3,
search_params={"metric_type": "IP", "params": {}},
output_fields=['name', 'description']
)
# 顯示結(jié)果
for result in res:
print(f"Found menu item: {result['name']} - {result['description']}")注意,用戶的輸入例如 “喜歡辣味的菜品” 可能每個人表述不一樣。但是同樣需求下,通過向量查詢就可以找到相似的菜品信息,這正是向量數(shù)據(jù)庫的核心作用。
如果單獨針對一個應(yīng)用開發(fā)這樣一個知識庫是沒問題的,但是我們要做的是讓所有助理可以復(fù)用一套知識庫邏輯,因此在實戰(zhàn)中,我們會針對這類邏輯開發(fā)一個統(tǒng)一的插件。
下面是這個知識庫插件的偽代碼表示。
# AI知識庫插件
def plugin_ai_knowledge_base_query(query: str, knowledge_base_name: str, field: str) -> str:
"""
輸入:
query (str) - 用戶查詢的問題或主題
knowledge_base_name (str) - 知識庫的名稱
field (str) - 需要獲取的具體字段名稱
輸出:str - 從知識庫中獲取的相關(guān)信息
"""
# 調(diào)用原系統(tǒng)函數(shù)獲取知識庫信息
return original_system_get_knowledge_base_info(query, knowledge_base_name, field)黃燜雞點餐大師
其實,點餐大師的應(yīng)用邏輯就是工作流的編排邏輯,只不過在我們的營銷 Agent 項目里,沒有實現(xiàn)拖拽開發(fā),而是通過一個簡單的 yaml 配置文件來編排工作流。
以黃燜雞餐館的菜單推薦需求為例,我們開發(fā)一個菜單推薦工作流,用來給門店點餐助理擴展能力。你只需要在 yaml 配置文件里將工作的前后關(guān)系和輸入輸出配置即可,下面是對應(yīng)的偽代碼。
version: '1.0'
name: '菜單推薦工作流'
description: '一個基于用戶喜好使用AI知識庫和自定義邏輯推薦菜品的工作流。'
steps:
- id: 'input_step'
type: 'input'
description: '獲取用戶對菜品的喜好'
input:
description: '用戶輸入的菜品喜好'
example: '我喜歡吃辣,有什么菜品推薦'
output:
name: 'user_preference'
type: 'str'
- id: 'query_knowledge_base'
type: 'plugin'
plugin: 'plugin_ai_knowledge_base_query'
description: '根據(jù)用戶喜好查詢AI知識庫中的菜品推薦'
inputs:
query: '{user_preference}'
knowledge_base_name: 'huangmenji_menu'
# 在這里具體化字段的選擇
field: 'embedding' # 使用向量數(shù)據(jù)庫的嵌入向量字段進行查詢
outputs:
name: 'dish_recommendations'
type: 'str'
- id: 'custom_logic_filter_spicy'
type: 'custom'
function: 'filter_spicy_dishes'
description: '過濾推薦的菜品,僅保留辣味菜品'
inputs:
recommendations: '{dish_recommendations}'
outputs:
name: 'spicy_dish_recommendations'
type: 'list'
- id: 'output_step'
type: 'output'
description: '向用戶提供最終的辣味菜品推薦列表'
inputs:
spicy_dishes: '{spicy_dish_recommendations}'
output:
description: '菜品推薦列表'
example: ['麻婆豆腐', '辣子雞', '四川火鍋']在這個工作流配置中,query_knowledge_base 是剛才說的插件能力,在用戶上傳自己的菜單之后可以直接使用,filter_spicy_dishes 則是我們內(nèi)部針對菜單這個行業(yè)場景開發(fā)的菜品推薦邏輯,它可以是基于 LLM 開發(fā)的。
看到這個編排工作流的 yaml 配置,你可能意識到它沒辦法和我們的 Agent 底層 Demo 直接對接到一起。實際上我們還需要一個統(tǒng)一的、基于 yaml 配置調(diào)度具體工作流邏輯的框架。你也可以自己想一想這部分的邏輯。
這里只是借助應(yīng)用的配置更好地說明 Agent 底層是怎么工作的。
好,我再來說一個更進一步的配置例子,黃燜雞營銷大師應(yīng)用。它能根據(jù)客戶輸入的用戶標簽,自動篩選和發(fā)送營銷短信。實際上,這個工作流的開發(fā)更加簡單,只需要通過配置組合現(xiàn)有的插件能力就可以完成,不需要編寫代碼。其配置文件的偽代碼如下。
version: '1.0'
name: '客戶營銷工作流'
description: '根據(jù)用戶標簽和營銷需求進行個性化營銷的工作流。'
steps:
- id: 'input_step'
type: 'input'
description: '獲取用戶標簽和營銷需求'
input:
description: '用戶輸入的標簽和營銷需求'
example:
tag: '潛在客戶'
marketing_need: '推廣新產(chǎn)品'
output:
name: 'user_tag'
type: 'str'
name: 'marketing_need'
type: 'str'
- id: 'get_user_info'
type: 'plugin'
plugin: 'plugin_user_info_by_tag_get'
description: '根據(jù)用戶標簽獲取用戶詳細信息'
inputs:
tag: '{user_tag}'
outputs:
name: 'user_info'
type: 'dict'
- id: 'generate_marketing_article'
type: 'plugin'
plugin: 'plugin_marketing_article_generate'
description: '生成適合目標用戶的營銷文章'
inputs:
topic: '{marketing_need}'
audience: '{user_info[preferences]}'
outputs:
name: 'marketing_article'
type: 'str'
- id: 'send_sms'
type: 'plugin'
plugin: 'plugin_sms_send'
description: '發(fā)送營銷短信給用戶'
inputs:
mobile: '{user_info[mobile]}'
subject: '最新產(chǎn)品推薦'
body: '{marketing_article}'
outputs:
name: 'sms_status'
type: 'bool'
- id: 'output_step'
type: 'output'
description: '輸出短信發(fā)送狀態(tài)'
inputs:
sms_status: '{sms_status}'
output:
description: '短信發(fā)送是否成功'
example: true需要注意,這里的 plugin_sms_send 也是原營銷平臺的短信功能直接移植的。插件能力足夠多的情況下,我們開發(fā)工作流的效率會很高。




























