帶你拆解、實現一套Agent底層框架
Agent 平臺底層框架是怎么樣的？
底層開發
我們應該將這一萬個 Agent 視為一個統一的 Agent 實體。這個統一的 Agent 可以通過提示詞進行重新定義。本質上，它就是一個基于大型模型的聊天應用程序。
關鍵在于，我們需要一種機制，將我們的業務執行邏輯嵌入到聊天的過程中。
圖片
有的人可能會想，讓大模型直接調用我們的業務邏輯不就行了嗎？從廣義上來說，這樣實現也沒問題，只不過我們這里說的大模型指的是只有文本聊天能力的大模型。所以還需要一個“助理”角色代理執行。
意圖識別
那怎么才能實現這個核心的代理層呢？我們先從一個簡單的例子開始說。用過 GPT 的朋友可能都有下面的體驗，當我們要求 GPT 基于互聯網已有知識回答問題，它會怎么做呢？
圖片
你看，我問了一個問題，GPT 搜索了 7 個網站，然后根據這些網站的內容做出了回答。但是顯然大模型是不具備網絡搜索能力的。那它是怎么做的呢？答案是我們提問里的 請你先查詢網絡資料再回答我 這個意圖必須被識別出來，并改變原有流程，它才能做到執行搜索操作。
圖片
當然，請你先查詢網絡資料再回答我 這句話還有別的表述方式，比如下面的這些。
請先通過網絡搜索相關資料，然后再回答我的問題。
先查閱一些網絡資源，再給我詳細解答。
請先在網上找一些資料，然后再為我解答。
先在互聯網搜集相關信息，然后再回答我。
請先查找網絡上的相關內容，再來回答我的問題。
請先從網上獲取相關信息，然后再給出答案。
先通過網絡查詢一下，再給我提供答案。
請先在網上找到一些相關資料，再來回答我的問題。
請先通過網絡查詢相關內容，然后再回答我。
請先利用網絡搜索相關資料，然后再給出解答。在設計意圖識別模塊時，有一個關鍵問題不容忽視，那就是系統必須能夠理解用戶可能使用的各種不同表述方式。與此同時，意圖識別也不僅僅是單輪輸入的匹配，更需要具備對上下文的感知能力。畢竟，用戶的真實意圖往往不會直接說出來，而是埋藏在多次對話的語境之中。
當用戶的搜索意圖逐漸被識別出來以后，代理便能夠據此調起合適的插件或工作流。實際上，不論是大模型所具備的插件功能，還是 Agent 平臺上的插件能力與工作流，本質上都沒有區別。它們的共同點，都是為了讓大模型能夠獲得額外的程序執行能力，從而完成更加復雜的任務。
不過，相比單純的意圖識別，還有一個更重要的環節，那就是參數的提取。就像在之前討論過的營銷 AI 場景里，當助理與用戶進行聊天時，代理層不僅要判斷出該調用哪條工作流，更要從對話的上下文中提煉出關鍵的參數。比如，在那個案例中，最重要的參數就是公司的名稱。
那上面例子里的網絡讀取插件，輸入參數是什么呢？你可以想想。
圖片
我想你也注意到了，要實現這個 Agent 的底層，最核心的就是意圖識別和參數識別。需要注意，這兩個能力都是大模型具備的，我們實現的 Agent 底層也是用大模型來做意圖識別和參數識別。
Agent 只是一個公司的客服角色，它的職責是滿足客戶的需求，并負責溝通，工作流則像公司內部的部門，他們并不直接對外。
技術實現
后，我們說技術實現。意圖識別和參數識別的具體實現方法有兩種，可以通過微調大模型實現，也可以通過提示詞實現，思想都是想通的。
下面是一個 Demo 級別的核心架構代碼演示，這個代碼就是上述 Agent 底層架構的核心邏輯實現例子。這個 Demo 以一個用餐業務為例，分為提示詞和核心代碼兩個部分。理解了這個例子，剩下的一萬個 Agent 也就理解了。
首先是核心提示詞。
現在有3個角色，sys 代表應用程序系統，user 代表用戶，你的角色是一名餐廳服務員，你的目的是服務用戶，請你根據上下文決定是否調用 sys應用程序系統。
調用系統則輸出 to_sys: 應用程序參數
回復用戶則輸出 to_user: 回復的消息
規則：
1，要一句一句的和我溝通，每次只能選擇和to_sys或to_user其中之一，我會代為轉達和處理消息，并且用sys 或 user開頭的消息回復你
2，調用程序的具體參數要在上下文中確定，比如我們應用程序有個參數a，上下文里用戶或系統沒有提供，你就不能擅自決定a的具體數值，調用程序用json格式
3，當我用sys：開頭給你回復的時候，表示本次應用程序返回，你要結合這個返回信息繼續和用戶溝通，你直接用to_user和用戶溝通，要注意影藏程序系統這個信息，讓用戶感知不到
反例：
1，類似下面的回復是錯誤的，錯誤原因是在消息里同時出現to_sys 和 to_user
"""
to_sys: {
‘app’ : '點餐應用',
'usernum' : 1,
'caiming' : '包子',
'cainum' : 3
}
to_user: 好的，三個包子已經為您下單了，稍等片刻就會上菜。是否還需要點其他菜品或者飲料呢？
"""
[應用程序列表如下]
1，點餐應用，當用戶點餐時調用，
請你和用戶溝通確定這些參數：用戶人數，菜名（必填），數量（必填)
輸出json格式例子：{
‘app’ : '點餐應用', #應用名字（必填）
'usernum' : 1, #用戶人數
'caiming' : '包子', #菜名
'cainum' : 5 #數量
}
2，菜單應用，在點餐之前調用，
參數：無
輸出json格式例子：{
‘app’ : '菜單應用', #應用名字（必填）
}
3，結賬應用，當用戶用完餐需要結賬時調用，
參數：無
輸出json格式例子：{
‘app’ : '結賬應用', #應用名字（必填）
}
接下來用to_user給用戶開始第一條消息
你的餐廳名字是: 成都小吃我們可分析一下這個核心提示詞，有三個層次。
其一是定義角色，提示詞里表明了代理、系統、用戶三者的關系，讓大模型根據上下文來決定具體的回復，這也就解決了意圖識別的問題。其二是對 sys 類參數的限定提示詞，讓大模型按插件和工作流的參數格式來回答。其三是將應用的工作流和具體參數注入提示詞，讓大模型理解應用。
這段提示詞的作用就是告訴大模型我們的具體場景以及助理的具體能力，將上下文交給它處理。需要注意的是，各個插件和工作流的參數都是大模型來組織的，我們給于引導即可。
和提示詞對應的 Demo 程序代碼如下。
import json
def deal_app(app, params):
print(">>> " + app)
if app == "菜單應用":
return '菜單 1包子 2餃子 3 可樂或雪碧'
if app == "結賬應用":
return '一共消費100元'
if app == "點餐應用":
if 'caiming' not in params or params['caiming'] == None:
return "請先選擇菜品和數量"
if params['caiming'] == '面包':
return '面包賣完了'
return "已下單"
return "沒有這個應用"
def deal_answer(answer):
if answer.find("to_user:") >= 0 and answer.find("to_sys:") >= 0:
print(answer)
return xf_ai('user', "請不要同時回復to_user 和 to_sys, 兩者只能選一個，也不要[等待系統回復]這樣的說明", "sys:")
if answer.find("to_sys:") >= 0 and answer.find("\nsys:") >= 0:
print(answer)
return xf_ai('user', "請不要同時回復to_sys 和 sys, 只回復to_sys的數據", "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('服務員：' + 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)這個程序就是代理層的核心邏輯代碼了。整體的思想就是圍繞核心提示詞構建處理流程，在一般情況下直接回答用戶的問題，在工作流狀態下調用工作流邏輯，根據返回信息再讓模型組織語言回復用戶，完成整個交互。
下面是它核心程序流程圖。
圖片
千萬不要小看這個提示詞和底層 Demo，實際上麻雀雖小，五臟俱全。我們整個系統都是在這個基礎代碼上構建的，你可以把這部分的實現單獨作為一個實驗來做，一定會對大模型開發有更深入的體會。
如果要更好地理解整套 Agent 系統，我想可以從我們最終的助理創建界面繼續給你分析。
圖片
我們的創建助理功能實際分為 5 步，每一步都必須完成，這是和 Coze 這類平臺不同的，其中最重要的一步就是給每個助理配一個知識庫。
應用開發
這里我還是會舉一個具體的應用例子，因為我們的底層 demo 是以餐館菜單為基礎實現的，所以這里我們就繼續擴展這個案例，開發一個黃燜雞點餐大師。
知識庫插件
那么顯然，菜單數據是它的核心。實際上大部分應用都會用到數據庫，這也是我們為什么給每個助理都配置知識庫的原因。
圖片
實現這套客戶自助配置，如何跟現有系統對接呢？需要分為兩部分。其一是數據層面的自助訓練，其二是業務邏輯層面的工作流移植。
在數據層面，我們提供了知識庫自助訓練，這樣每個客戶都可以擁有一個自己定制化的小模型，保證數據的私有屬性。
以一個黃燜雞餐館菜單的數據為例，我們看一下私有知識庫的搭建流程。
第一步，將菜單數據整理成固定的輸入格式，這一步比較簡單。
{
"1": {"name": "經典黃燜雞", "description": "雞肉鮮嫩，配以濃郁醬汁和土豆", "price": "12元", "category": "主菜"},
"2": {"name": "辣味黃燜雞", "description": "經典黃燜雞基礎上加入辣椒調味", "price": "13元", "category": "主菜"},
"3": {"name": "香菇黃燜雞", "description": "黃燜雞加入香菇，風味獨特", "price": "14元", "category": "主菜"}
}第二步，將菜單向量化，將類似 item["description"] 這樣的本字段轉化為向量表示，具體偽代碼如下。
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
# 讀取數據并向量化
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)第三步，存儲和讀取向量數據庫，這里以 Milvus向量數據庫 為例，注意要將菜單信息和向量信息一起存儲，具體偽代碼如下。
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}
)
# 創建集合
collection_name = "huangmenji_menu"
client.create_collection(
collection_name=collection_name,
schema=schema,
index_params=index_params
)
# 插入數據
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)。你可以把它理解為傳統數據庫里的索引。加上這個字段后，后續工作流中需要用到菜單時都可以用向量查詢，具體查詢類似下面的偽代碼。
# 查詢向量化輸入
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']
)
# 顯示結果
for result in res:
print(f"Found menu item: {result['name']} - {result['description']}")注意，用戶的輸入例如 “喜歡辣味的菜品” 可能每個人表述不一樣。但是同樣需求下，通過向量查詢就可以找到相似的菜品信息，這正是向量數據庫的核心作用。
如果單獨針對一個應用開發這樣一個知識庫是沒問題的，但是我們要做的是讓所有助理可以復用一套知識庫邏輯，因此在實戰中，我們會針對這類邏輯開發一個統一的插件。
下面是這個知識庫插件的偽代碼表示。
# 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 - 從知識庫中獲取的相關信息
"""
# 調用原系統函數獲取知識庫信息
return original_system_get_knowledge_base_info(query, knowledge_base_name, field)黃燜雞點餐大師
其實，點餐大師的應用邏輯就是工作流的編排邏輯，只不過在我們的營銷 Agent 項目里，沒有實現拖拽開發，而是通過一個簡單的 yaml 配置文件來編排工作流。
以黃燜雞餐館的菜單推薦需求為例，我們開發一個菜單推薦工作流，用來給門店點餐助理擴展能力。你只需要在 yaml 配置文件里將工作的前后關系和輸入輸出配置即可，下面是對應的偽代碼。
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: '根據用戶喜好查詢AI知識庫中的菜品推薦'
inputs:
query: '{user_preference}'
knowledge_base_name: 'huangmenji_menu'
# 在這里具體化字段的選擇
field: 'embedding' # 使用向量數據庫的嵌入向量字段進行查詢
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 則是我們內部針對菜單這個行業場景開發的菜品推薦邏輯，它可以是基于 LLM 開發的。
看到這個編排工作流的 yaml 配置，你可能意識到它沒辦法和我們的 Agent 底層 Demo 直接對接到一起。實際上我們還需要一個統一的、基于 yaml 配置調度具體工作流邏輯的框架。你也可以自己想一想這部分的邏輯。
這里只是借助應用的配置更好地說明 Agent 底層是怎么工作的。
好，我再來說一個更進一步的配置例子，黃燜雞營銷大師應用。它能根據客戶輸入的用戶標簽，自動篩選和發送營銷短信。實際上，這個工作流的開發更加簡單，只需要通過配置組合現有的插件能力就可以完成，不需要編寫代碼。其配置文件的偽代碼如下。
version: '1.0'
name: '客戶營銷工作流'
description: '根據用戶標簽和營銷需求進行個性化營銷的工作流。'
steps:
- id: 'input_step'
type: 'input'
description: '獲取用戶標簽和營銷需求'
input:
description: '用戶輸入的標簽和營銷需求'
example:
tag: '潛在客戶'
marketing_need: '推廣新產品'
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: '根據用戶標簽獲取用戶詳細信息'
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: '發送營銷短信給用戶'
inputs:
mobile: '{user_info[mobile]}'
subject: '最新產品推薦'
body: '{marketing_article}'
outputs:
name: 'sms_status'
type: 'bool'
- id: 'output_step'
type: 'output'
description: '輸出短信發送狀態'
inputs:
sms_status: '{sms_status}'
output:
description: '短信發送是否成功'
example: true需要注意，這里的 plugin_sms_send 也是原營銷平臺的短信功能直接移植的。插件能力足夠多的情況下，我們開發工作流的效率會很高。
