大模型ReAct框架——打造AI Agent的代碼實現——基于LLM + Function Call構建Agent 原創
“ Agent通過大模型的推理和規劃，使得大模型真正實現類似人類的能力”
AI Agent也就是AI智能體，是通過把大模型作為“大腦”，通過利用大模型的推理和規劃能力，然后調用外部工具來完成復雜任務的一種方式。
簡單來說，Agent就是一種讓大模型自己思考和分析問題，選擇合適的工具，最終解決問題的一種方法，其背后原理就來自于ReAct。
ReAct是Reasoning And Acting的縮寫，意思是LLM可以根據邏輯推理(Reson)，構建完整系列行動(Act)，從而達到期望目標。
LLM的靈感來源于人類和推理之間的協同關系，人類根據這種協同關系學習新的知識，做出決策，然后執行。
什么是ReAct框架？
從本質上來說，智能體的作用就是模仿人類的思維和處理復雜問題的方式。
基于LLM 和 Function Call實現Agent
ReAct的作用就是協同LLM和外部的信息獲取，與其它功能交互，如果說LLM模型是大腦，那么ReAct框架就是這個大腦的手腳和五官。
下面我們就用代碼來實現一個簡單的具有自主規劃功能的Agent，需要的東西也很簡單：
Python開發環境 python 版本用到3.12.1 版本沒有強制要求
支持Function Call 工具的大模型(可以是自己部署的大模型或者第三方模型)。
使用第三方模型需要自己申請并獲取其API-KEY，代碼中還用到了tavily搜索，這個也需要自己去申請。下圖是Agent根據任務要求輸出的結果，任務要求是
請幫我制定一份理財計劃，你可以通過網絡搜索的方式來收集一定的參考資料，并把最終的計劃內容寫入到理財計劃.txt文件中
Agent實現的核心有三點
- 大模型的質量
- 外部工具集
- 提示詞的質量
大模型的質量問題直接影響到Agent表現的好壞，推理能力強，知識豐富的大模型會表現更好。
而外部工具集就是提供給大模型使用的工具可以根據不同的業務場景提供不同的工具集(API)。如果使用一些第三方API可能需要自己申請，比如百度或谷歌搜索，高德和百度的地圖接口等。
提示詞是最重要的一個環節，我們知道大模型的能力是一方面，但怎么發揮大模型的能力是由提示詞的質量決定的。
# 約束
constraints = [
"僅使用下面列出的動作",
"你只能主動行動，在計劃行動時需要考慮到這一點",
"你無法與物理對象交互，如果對于完成任務或目標是絕對必要的，則必須要求用戶為你完成，如果用戶拒絕，并且沒有其它方法實現目標，則直接終止，避免浪費時間和精力"
]
# 資源
resources = [
"提供搜索和信息搜集的互聯網接入",
"讀取和寫入文件的能力",
"你是一個大語言模型，接受了大量的文本訓練，包括大量的事實知識，利用這些知識來避免不必要的信息收集"
]
# 最佳實踐說明
best_practices = [
"不斷地回顧和分析你的行為，確保發揮出你最大的能力",
"不斷地進行建設性的自我批評",
"反思過去的決策和策略，完善你的方案",
"每個動作執行部分代價，所以要聰明高效，目的是用最少的步驟完成任務"
]
prompt_template = """
你是一個問答專家，你必須始終獨立做出決策，無需尋求用戶的幫助，發揮你作為LLM的優勢，追求簡單的策略，不要涉及法律問題
任務：
{query}
限制條件說明：
{constraints}
動作說明：這是你唯一可以使用的作用，你的任何操作都必須通過以下操作實現:
{actions}
資源說明:
{resources}
最佳實踐的說明：
{best_practices}
agent_scratch:
{agent_scratch}
你應該只以json格式響應，響應格式如下：
{response_format_prompt}
確保響應結果可以由python json.loads解析
"""
response_format_prompt = """
{
"action":{
"name": "action name",
"args": {
"answer": "任務的最終結果"
}
},
"thoughts": {
"plan": "簡短的描述短期和長期的計劃列表",
"criticism": "建設性的自我批評",
"speak": "當前步驟，返回給用戶的總結",
"reasoning": "推理"
},
"observation": "觀察當前任務的整體進度"
}
"""在這個提示詞中加入了工具列表，資源說明，任務需求等；并且約定了大模型的輸出格式，以便于進行解析，大模型就可以根據這提示詞對任務進行思考和推理。
并且根據推理結果，選擇合適的工具來完成對應的任務，比如調用搜索工具完成信息收集，調用文件寫入工具把結果寫入到文件中。
下圖是大模型的思考，推理和工具調用的過程，從圖中可以看出大模型經過多次規劃才完成的任務。
其次，就是工具集的構建，簡單來說就是一些python函數，用來給大模型進行調用：
"""
1. 寫文件
2. 讀文件
3. 追加
4. 網絡搜索
"""
def get_workdir_root():
workdir_root = os.environ.get("WORKDIR_ROOT", './data/llm_result')
return workdir_root
WORKDIR_ROOT = get_workdir_root()
def read_file(filename):
if not os.path.exists(filename):
return f"{filename} not exist, please check file exist before read"
with open(filename, "r") as f:
return "\n".join(f.readline())
def append_to_file(filename, content):
filename = os.path.join(WORKDIR_ROOT, filename)
if not os.path.exists(filename):
return f"{filename} not exist, please check file exist before read"
with open(filename, 'a') as f:
f.write(content)
return "append content to file success"
def write_to_file(filename, content):
filename = os.path.join(WORKDIR_ROOT, filename)
if not os.path.exists(WORKDIR_ROOT):
os.makedirs(WORKDIR_ROOT)
with open(filename, 'w') as f:
f.write(content)
return "write content to file success"
def search(query):
tavily = TavilySearchResults(max_results=5)
try:
ret = tavily.invoke(input=query)
"""
ret:
[{
"content": "",
"url": ""
}]
"""
print("搜索結果", ret)
content_list = [obj["content"] for obj in ret]
return "\n".join(content_list)
except Exception as err:
return "search err: {}".format(err)
tools_info = [
{
"name": "read_file",
"description": "read file from agent generate, should write file before read.",
"args": [
{
"name": "filename",
"type": "string",
"description": "read file name"
}
]
},
{
"name": "append_to_file",
"description": "append llm content to file, should write file before read.",
"args": [
{
"name": "filename",
"type": "string",
"description": "file name"
},
{
"name": "filename",
"type": "string",
"description": "append to file content"
}
]
},
{
"name": "write_to_file",
"description": "write llm content to file",
"args": [
{
"name": "filename",
"type": "string",
"description": "file name"
},
{
"name": "filename",
"type": "string",
"description": "write to file content"
}
]
},
{
"name": "search",
"description": "this is a search engine, you can gain additional knowledge though this search engine when you are unsure of what large model return",
"args": [
{
"name": "query",
"type": "string",
"description": "search query to lookup"
}
]
},
{
"name": "finish",
"description": "完成用戶目標",
"args": [{
"name": "answer",
"type": "string",
"description": "最后的目標結果"
}]
}
]
tools_map = {
"read_file": read_file,
"append_to_file": append_to_file,
"write_to_file": write_to_file,
"search": search
}
def gen_tools_desc():
tools_desc = []
for idx, t in enumerate(tools_info):
args_desc = []
for info in t['args']:
args_desc.append({
"name": info['name'],
"description": info["description"],
"type": info['type']
})
args_desc = json.dumps(args_desc, ensure_ascii=False)
tool_desc = f"{idx + 1}. {t['name']}: {t['description']}, args: {args_desc}"
tools_desc.append(tool_desc)
tools_prompt = "\n".join(tools_desc)
return tools_prompt最后兩個就是大模型的調用模塊和業務的解析模塊，大模型的調用模塊相對比較簡單，這里就不仔細說了，感興趣的可以直接看代碼。
解析模塊說簡單也簡單，說復雜也復雜；因為當前的功能比較簡單，因此只需要使用大模型本身的能力即可，然后完成對大模型每次思考和規劃數據的解析即可。
而如果后續需要開發更加復雜的業務功能，比如說金融行業的投資分析，需要非常復雜的業務分析等環節，這時只依靠大模型本身的能力就不行了。
比如說，由于大模型在垂直領域的表現不佳，直接使用可能會帶來幻覺等問題；還有就是Agent的記憶模塊，在大量的復雜業務分析中，需要增加外部存儲模塊來記錄歷史記憶功能，這樣才能更好地完成復雜的任務處理。
"""
todo：
環境變量的設置
工具的引入
prompt模板
模型的初始化
"""
# 初始化模型
mp = ModelProvider()
# 解析大模型的響應
def parse_thoughts(response):
try:
thoughts = response.get("thoughts")
observation = response.get("observation")
plan = thoughts.get("plan")
reasoning = thoughts.get("reasoning")
criticism = thoughts.get("criticism")
prompt = f"plan: {plan}\n reasoning: {reasoning}\n criticism: {criticism}\nobservation: {observation}"
print("thoughts: ", prompt)
return prompt
except Exception as err:
print("parse thoughts err: {}".format(err))
return "".format(err)
def agent_execute(query, max_request_time=10):
cur_request_time = 0
# 大模型記憶 包括短期記憶和長期記憶
chat_history = []
# agent 反思 規劃等
agent_scratch = ''
while cur_request_time < max_request_time:
cur_request_time += 1
"""
如果返回結果達到預期，則直接返回
"""
# 提示詞模板
"""
prompt包含的功能：
1. 任務描述
2. 工具描述
3. 用戶的輸入user_msg
4. assistant_msg
5. 限制
6. 給出更好實踐的描述
"""
prompt = gen_prompt(query, agent_scratch)
start_time = time.time()
print("--------------------------{}, 開始調用大模型LLM------------".format(cur_request_time), flush=True)
# 調用大模型 直接返回json格式數據
"""
sys_prompt:
user_msg, assistant, history
"""
if cur_request_time < 3:
print("prompt: ", prompt)
# response = call_llm()
response = mp.chat(prompt, chat_history=chat_history)
end_time = time.time()
print("--------------------------{}, 調用大模型結束，耗時：{}---------------".format(cur_request_time, end_time-start_time), flush=True)
if not response or not isinstance(response, dict):
print("調用大模型錯誤, 即將重試: ", response)
continue
"""
大模型返回格式約定
response:
{
"action":{
"name": "action name",
"args": {
"args name": "args values"
}
},
"thoughts": {
"text": "thought",
"plan": "plan",
"criticism": "criticism",
"speak": "當前步驟，返回給用戶的總結",
"reasoning": ""
}
}
"""
action_info = response.get("action")
action_name = action_info.get("name")
action_args = action_info.get("args")
print("當前action name: ", action_name, action_args)
if action_name == "finish":
final_answer = action_args.get("answer")
print("final_answer: ", final_answer)
break
observation = response.get("observation")
try:
"""
action_name 到函數的映射, map -> { action_name: func }
"""
# todo: tools_map 的實現
# tools_map = {}
func = tools_map.get(action_name)
call_func_result = func(**action_args)
# { action_name: func }
except Exception as err:
print("調用工具異常: ", err)
call_func_result = "{}".format(err)
agent_scratch = agent_scratch + "\n:observation:{}\n execute action result: {}".format(observation, call_func_result)
# 由于大模型沒有記憶功能，因此需要把大模型之前的輸入和輸出加入到history中，這也是Agent四大塊中的記憶模塊 由于目前的業務并不復雜，因此不需要外部記憶模塊 如果業務比較復雜可能還需要外部模塊進行存儲
assistant_msg = parse_thoughts(response)
chat_history.append([user_prompt, assistant_msg])
if cur_request_time == max_request_time:
print("很遺憾，本次任務失敗")
else:
print("恭喜你，任務完成")
def main():
# 支持用戶多次交互 最大規劃次數設置為10次 當超過10次時 則說明任務失敗 用戶可以根據自己的情況進行調整
max_request_time = 10
while True:
query = input("請輸入你的目標：")
if query == "exit":
return
agent_execute(query, max_request_time=max_request_time)
if __name__ == "__main__":
main()最后，這里只是為了實現Agent的實現流程，并且由于大模型的質量問題，也不能保證每次都能成功。
受限于大模型的性能問題，大模型在推理和規劃方面表現還不盡人意，因此怎么才能讓大模型更好實現復雜推理規劃是一個值得研究的問題。
里面主要涉及兩個第三方模塊，一個是第三方大模型的申請，個人使用的是阿里的通義千問；第二個是Travily的搜索接口，官網地址：https://app.tavily.com用戶也可以自己去申請或者使用其它搜索工具，比如百度搜索等。
Agent學習參考
??https://www.bilibili.com/video/BV1Sz421m7Rr?p=1&vd_source=f2f15e671adda1cc1e5512694f310ebd??
用戶可以在公眾號回復：Agent源碼 獲取 用戶只需要在env文件中把參數改成自己的參數，然后啟動agent_main.py文件即可。
此代碼只是用來學習使用，并不能完成復雜的業務邏輯，用戶如果想實現更加復雜的業務功能，則需要對提示詞和工具進行添加和調整，比如做旅行規劃就需要添加地圖工具和酒店預定的API等。
本文轉載自公眾號AI探索時代 作者：DFires
原文鏈接：??https://mp.weixin.qq.com/s/KV1hp5SmdOAoyFoJX9vmmQ??
