【万字长文】手把手带你从0到1实现大模型agent

本文主要是介绍【万字长文】手把手带你从0到1实现大模型agent，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

前言

最近出了《手把手带你从0到1实现大模型agent》视频系列课程，从无到有的实现一个完整的大模型agent。其实市面上已经有很多agent的开源框架，实际的应用在框架基础上开发即可。出这个系列的原因主要还是希望能让大家对agent的底层原理和逻辑有一个清晰的认识，所谓知其然，更知其所以然。

目录

该系列课程的视频内容包括的核心内容，下文介绍核心模块及代码:

• - agent核心逻辑实现及讲解

• - agent各模块功能分析和梳理

• - tools的定义及与action的映射

• - prompt模板定义

• - prompt模板定义及工具prompt集成

• - 通义千问大模型作为基座模型接入

• - agent完整链路调试

• - 针对不足进行agent优化升级

核心逻辑实现

核心逻辑模块也是整个项目的入口，集成了

1. 环境变量的设置 ;

2. 工具(action)的引入;

3. prompt模板；

4)模型的初始化；

5. 整个逻辑的交互。

代码如下

核心逻辑模块也是整个项目的入口utf-8 -*-  
"""  
@author: acedar  @time: 2024/5/12 10:25  
@file: cli_main.py """  import time  
from tools import tools_map  
from prompt_cn import gen_prompt, user_prompt  
from model_provider import ModelProvider  
from dotenv import load_dotenv  load_dotenv()  # agent入口  """  
todo:  1. 环境变量的设置    2. 工具的引入    3. prompt模板    4. 模型的初始化"""  mp = ModelProvider()  def parse_thoughts(response):  """  response:        {            "action": {                "name": "action name",                "args": {                    "args name": "args value"                }            },            "thoughts":            {                "text": "thought",                "plan": "plan",                "criticism": "criticism",                "speak": "当前步骤，返回给用户的总结",                "reasoning": ""            }        }    """  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}\nreasoning:{reasoning}\ncriticism: {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_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)  # call llm  """  sys_prompt:        user_msg, assistant, history  """  if cur_request_time < 3:  print("prompt:", prompt)  response = mp.chat(prompt, 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 value"                }            },            "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}            """  # tools_map的实现  func = tools_map.get(action_name)  call_func_result = func(**action_args)  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)  assistant_msg = parse_thoughts(response)  chat_history.append([user_prompt, assistant_msg])  if cur_request_time == max_request_time:  print("很遗憾，本次任务失败")  else:  print("恭喜你，任务完成")  def main():  # 需求: 支持用户的多次交互  max_request_time = 30  while True:  query = input("请输入您的目标:")  if query == "exit":  return  agent_execute(query, max_request_time=max_request_time)  if __name__ == "__main__":  main()  

tools的定义及与action的映射

agent中交互式让模型给出下一步需要执行的动作(action，有点像判别模型)， 每个动作(action)的执行内容是需要提前定义好的，每个动作执行的定义我们称为工具，每个工具和动作是一一对应的，这里通过函数定义的方式定义工具。

下文包括了读文件、写文件、增加内容及调用搜索功能。tools_map定义了action到tools的映射。

import os  
import json  
from langchain_community.tools.tavily_search import TavilySearchResults  """  
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):  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, 'r', encoding='utf-8') as f:  return "\n".join(f.readlines())  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', encoding='utf-8') 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', encoding='utf-8') 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": "content",  "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": "content",  "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 look up"  }]  },  {  "name": "finish",  "description": "return finish when you get exactly the right answer",  "args": [{  "name": "answer",  "type": "string",  "description": "the final answer"  }]  }  
]  tools_map = {  "read_file": read_file,  "append_to_file": append_to_file,  "write_to_file": write_to_file,  "search": search  
}

prompt模板定义

prompt模板的定义需要包括工具的描述，返回结果及如何引导模型如何优化效果的描述等，当然具体的内容可以根据自身情况修改，并不是一成不变的，我实现的模板如下：

其中“{}”中的内容为需要补充填写的，每个定义也不一样，需要拼接成字符串的形式加到prompt中。

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": "观察当前任务的整体进度"  
}  
"""  # todo: query, agent_scratch, actions  
action_prompt = gen_tools_desc()  
constraints_prompt = "\n".join([f"{idx+1}. {con}" for idx, con in enumerate(constraints)])  
resources_prompt = "\n".join([f"{idx+1}. {con}" for idx, con in enumerate(resources)])  
best_practices_prompt = "\n".join([f"{idx+1}. {con}" for idx, con in enumerate(best_practices)])  def gen_prompt(query, agent_scratch):  prompt = prompt_template.format(  query=query,  constraints=constraints_prompt,  actions=action_prompt,  resources=resources_prompt,  best_practices=best_practices_prompt,  agent_scratch=agent_scratch,  response_format_prompt=response_format_prompt  )  return prompt  user_prompt = "根据给定的目标和迄今为止取得的进展，确定下一个要执行的action，并使用前面指定的JSON模式进行响应："  

通义千问大模型作为基座模型接入

经过调研，在开源的模型中通义千问的指令遵循效果较好，尤其是格式输出的遵循比较好。

一个能按格式输出的需求很重要，这便于模型结果的解析并提升模型调用的成功率（结果解析失败需要重新调用模型），本项目采用客户端的方式通义千问模型。

import os  
import json  
import dashscope  
from dashscope.api_entities.dashscope_response import Message  
from prompt_cn import user_prompt  class ModelProvider(object):  def __init__(self):  self.api_key = os.environ.get("API_KEY")  self.model_name = os.environ.get("MODEL_NAME")  self._client = dashscope.Generation()  print("model_name:", self.model_name)  self.max_retry_time = 3  def chat(self, prompt, chat_history):  cur_retry_time = 0  while cur_retry_time < self.max_retry_time:  cur_retry_time += 1  try:  messages = [Message(role='system', content=prompt)]  for his in chat_history:  messages.append(Message(role='user', content=his[0]))  messages.append(Message(role='assistant', content=his[1]))  messages.append(Message(role='user', content=user_prompt))  response = self._client.call(  model=self.model_name,  api_key=self.api_key,  messages=messages  )  """  {  "status_code": 200,  "request_id": "c965bd27-c89c-9b5c-924d-2f1688e8041e",   "code": "",   "message": "",   "output": {  "text": null, "finish_reason": null,  "choices": [{  "finish_reason": "null", "message":   {"role": "assistant", "content": "当然可以，这里有一个简单又美味"}  }]  },   "usage": {  "input_tokens": 31,   "output_tokens": 8,   "total_tokens": 39,   "plugins": {}  }  }  """  print("response:", response)  content = json.loads(response['output']['text'])  return content  except Exception as err:  print("调用大模型出错：{}".format(err))  return {}  

其他说明

本项目中，调用大模型的api_key，模型名及调用搜索的key均通过环境变量的方式获取，只需要配置.env文件即可。

# bailian  
MODEL_NAME=  
API_KEY=sk-  # search  
TAVILY_API_KEY=tvly-  

让环境变量生效，只需要在项目启动入口执行如下语句，注意.env需要放到当前执行的目录下，或者指定load_dotenv的参数dotenv_path：

from model_provider import ModelProvider  # load env  
load_dotenv()

大模型&AI产品经理如何学习

求大家的点赞和收藏，我花2万买的大模型学习资料免费共享给你们，来看看有哪些东西。

1.学习路线图

第一阶段： 从大模型系统设计入手，讲解大模型的主要方法；

第二阶段： 在通过大模型提示词工程从Prompts角度入手更好发挥模型的作用；

第三阶段： 大模型平台应用开发借助阿里云PAI平台构建电商领域虚拟试衣系统；

第四阶段： 大模型知识库应用开发以LangChain框架为例，构建物流行业咨询智能问答系统；

第五阶段： 大模型微调开发借助以大健康、新零售、新媒体领域构建适合当前领域大模型；

第六阶段： 以SD多模态大模型为主，搭建了文生图小程序案例；

第七阶段： 以大模型平台应用与开发为主，通过星火大模型，文心大模型等成熟大模型构建大模型行业应用。

2.视频教程

网上虽然也有很多的学习资源，但基本上都残缺不全的，这是我自己整理的大模型视频教程，上面路线图的每一个知识点，我都有配套的视频讲解。

（都打包成一块的了，不能一一展开，总共300多集）

因篇幅有限，仅展示部分资料，需要点击下方图片前往获取

3.技术文档和电子书

这里主要整理了大模型相关PDF书籍、行业报告、文档，有几百本，都是目前行业最新的。

4.LLM面试题和面经合集

这里主要整理了行业目前最新的大模型面试题和各种大厂offer面经合集。

👉学会后的收获：👈
• 基于大模型全栈工程实现（前端、后端、产品经理、设计、数据分析等），通过这门课可获得不同能力；

• 能够利用大模型解决相关实际项目需求： 大数据时代，越来越多的企业和机构需要处理海量数据，利用大模型技术可以更好地处理这些数据，提高数据分析和决策的准确性。因此，掌握大模型应用开发技能，可以让程序员更好地应对实际项目需求；

• 基于大模型和企业数据AI应用开发，实现大模型理论、掌握GPU算力、硬件、LangChain开发框架和项目实战技能， 学会Fine-tuning垂直训练大模型（数据准备、数据蒸馏、大模型部署）一站式掌握；

• 能够完成时下热门大模型垂直领域模型训练能力，提高程序员的编码能力： 大模型应用开发需要掌握机器学习算法、深度学习框架等技术，这些技术的掌握可以提高程序员的编码能力和分析能力，让程序员更加熟练地编写高质量的代码。

1.AI大模型学习路线图
2.100套AI大模型商业化落地方案
3.100集大模型视频教程
4.200本大模型PDF书籍
5.LLM面试题合集
6.AI产品经理资源合集

👉获取方式：
😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓

这篇关于【万字长文】手把手带你从0到1实现大模型agent的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---