【收藏必备】大模型Agent开发:从ReAct原理到完整代码实现
文章介绍了基于ReAct框架的大模型Agent实现方法,通过将大模型作为"大脑",结合推理和规划能力调用外部工具完成复杂任务。详细讲解了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. 网络搜索
"""defget_workdir_root():
workdir_root = os.environ.get("WORKDIR_ROOT",'./data/llm_result')return workdir_root
WORKDIR_ROOT = get_workdir_root()defread_file(filename):ifnot os.path.exists(filename):returnf"{filename} not exist, please check file exist before read"withopen(filename,"r")as f:return"\n".join(f.readline())defappend_to_file(filename, content):
filename = os.path.join(WORKDIR_ROOT, filename)ifnot os.path.exists(filename):returnf"{filename} not exist, please check file exist before read"withopen(filename,'a')as f:
f.write(content)return"append content to file success"defwrite_to_file(filename, content):
filename = os.path.join(WORKDIR_ROOT, filename)ifnot os.path.exists(WORKDIR_ROOT):
os.makedirs(WORKDIR_ROOT)withopen(filename,'w')as f:
f.write(content)return"write content to file success"defsearch(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
}defgen_tools_desc():
tools_desc =[]for idx, t inenumerate(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()# 解析大模型的响应defparse_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)defagent_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)ifnot response ornotisinstance(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("恭喜你,任务完成")defmain():# 支持用户多次交互 最大规划次数设置为10次 当超过10次时 则说明任务失败 用户可以根据自己的情况进行调整
max_request_time =10whileTrue:
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用户也可以自己去申请或者使用其它搜索工具,比如百度搜索等。
最后
为什么要学AI大模型
当下,⼈⼯智能市场迎来了爆发期,并逐渐进⼊以⼈⼯通⽤智能(AGI)为主导的新时代。企业纷纷官宣“ AI+ ”战略,为新兴技术⼈才创造丰富的就业机会,⼈才缺⼝将达 400 万!
DeepSeek问世以来,生成式AI和大模型技术爆发式增长,让很多岗位重新成了炙手可热的新星,岗位薪资远超很多后端岗位,在程序员中稳居前列。
与此同时AI与各行各业深度融合,飞速发展,成为炙手可热的新风口,企业非常需要了解AI、懂AI、会用AI的员工,纷纷开出高薪招聘AI大模型相关岗位。
最近很多程序员朋友都已经学习或者准备学习 AI 大模型,后台也经常会有小伙伴咨询学习路线和学习资料,我特别拜托北京清华大学学士和美国加州理工学院博士学位的鲁为民老师给大家这里给大家准备了一份涵盖了AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频 全系列的学习资料,这些学习资料不仅深入浅出,而且非常实用,让大家系统而高效地掌握AI大模型的各个知识点。
这份完整版的大模型 AI 学习资料已经上传CSDN,朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费
】

AI大模型系统学习路线
在面对AI大模型开发领域的复杂与深入,精准学习显得尤为重要。一份系统的技术路线图,不仅能够帮助开发者清晰地了解从入门到精通所需掌握的知识点,还能提供一条高效、有序的学习路径。
但知道是一回事,做又是另一回事,初学者最常遇到的问题主要是理论知识缺乏、资源和工具的限制、模型理解和调试的复杂性,在这基础上,找到高质量的学习资源,不浪费时间、不走弯路,又是重中之重。
AI大模型入门到实战的视频教程+项目包
看视频学习是一种高效、直观、灵活且富有吸引力的学习方式,可以更直观地展示过程,能有效提升学习兴趣和理解力,是现在获取知识的重要途径
光学理论是没用的,要学会跟着一起敲,要动手实操,才能将自己的所学运用到实际当中去,这时候可以搞点实战案例来学习。
海量AI大模型必读的经典书籍(PDF)
阅读AI大模型经典书籍可以帮助读者提高技术水平,开拓视野,掌握核心技术,提高解决问题的能力,同时也可以借鉴他人的经验。对于想要深入学习AI大模型开发的读者来说,阅读经典书籍是非常有必要的。
600+AI大模型报告(实时更新)
这套包含640份报告的合集,涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师,还是对AI大模型感兴趣的爱好者,这套报告合集都将为您提供宝贵的信息和启示。
AI大模型面试真题+答案解析
我们学习AI大模型必然是想找到高薪的工作,下面这些面试题都是总结当前最新、最热、最高频的面试题,并且每道题都有详细的答案,面试前刷完这套面试题资料,小小offer,不在话下
这份完整版的大模型 AI 学习资料已经上传CSDN,朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费
】

更多推荐
所有评论(0)