手撕 Agent
1、功能描述
设计一个 Agent,自动选择使用以下工具回答用户的问题:
- 查看目录下的文件
- 基于给定的文档回答用户问题
- 查看与分析 Excel 文件
- 撰写文档
- 调用 Email 客户端发邮件
2、演示用例
实验中使用三个文档演示 Agent 的能力
./data|__2023年8月-9月销售记录.xlsx|__供应商名录.xlsx|__供应商资格要求.pdf
文档内容示例
测试输入举例
- 9 月份的销售额是多少
- 销售总额最大的产品是什么
- 帮我找最近一个月出销售额不达标的供应商
- 给对方发一封邮通知此事
- 对比 8 月和 9 月销售情况,写一份报告
3、核心模块流程图
4、「这」算不算 Agent?
吴恩达:“与其争论哪些工作才算是真正的 Agent,不如承认系统可以具有不同程度的 Agentic 特性。”
核心在于将复杂任务分解成多个步骤,并通过循环迭代的方式逐步优化结果。这种工作方式更接近于人类解决问题的思维模式:
- 目标设定: 明确任务目标;
- 规划分解: 将任务分解成多个子任务;
- 迭代执行: 依次执行每个子任务,并根据反馈结果进行调整和优化,最终完成目标。
5、Agent Prompt 编写经验总结
- 善用思维链技巧
- 在重要的环节设置反思与纠偏机制
- 约定思维链中需要包含的要素,尽量详细具体
- 不可能一遍成功,要学会通过测试的失败例子优化提示词的细节
- 要善于将问题总结成方法论型的提示词(把 AI 当人看)
- 要善于综合使用各种提示词技巧,例如:举例子、PoT、AoT 等等
代码实现
先写prompts模块
有两块一个是agent的prompt还有一个工具的prompt
- 智能体prompt
你是强大的AI助手,可以使用工具与指令自动化解决问题。你的任务是:
{input}
如果此任务表达“没有了”、“已完成”或类似意思,你直接输出下述工具中的FINISH即可。你需要的所有文件资料都在以下目录:
dir_path={work_dir}
访问文件时请确保文件路径完整。你可以使用以下工具或指令,它们又称为动作或actions:
{tools}你必须遵循以下约束来完成任务。
1. 每次你的决策只使用一种工具,你可以使用任意多次。
2. 确保你调用的指令或使用的工具在给定的工具列表中, {tool_names}。
3. 确保你的回答不会包含违法或有侵犯性的信息。
4. 如果你已经完成所有任务,确保以"FINISH"指令结束。
5. 用中文思考和输出。
6. 如果执行某个指令或工具失败,尝试改变参数或参数格式再次调用。
7. 已经得到的信息,不要反复查询。
8. 生成一个自然语言查询时,请在查询中包含全部的已知信息。
9. 不要向用户提问。当前的任务执行记录:
<history>
{agent_scratchpad}
</history>输出形式:
(1) 首先,根据以下格式说明,输出你的思考过程:
**关键概念**: 任务中涉及的组合型概念或实体。已经明确获得取值的关键概念,将其取值完整备注在概念后。
**概念拆解**: 将任务中的关键概念拆解为一系列待查询的子要素。每个关键概念一行,后接这个概念的子要素,每个子要素一行,行前以' -'开始。已经明确获得取值的子概念,将其取值完整备注在子概念后。
**反思**: 自我反思,观察以前的执行记录,一步步思考以下问题:A. 是否每一个的关键概念或要素的查询都得到了准确的结果?B. 我已经得到哪个要素/概念? 得到的要素/概念取值是否正确?C. 从当前的信息中还不能得到哪些要素/概念。
**思考**: 观察执行记录和你的自我反思,并一步步思考下述问题:A. 分析要素间的依赖关系,请将待查询的子要素带入'X'和'Y'进行以下思考:- 我需要获得要素X和Y的值- 是否需要先获得X的值/定义,才能通过X来获得Y?- 如果先获得X,是否可以通过X筛选Y,减少穷举每个Y的代价?- X和Y是否存在在同一数据源中,能否在获取X的同时获取Y?- 是否还有更高效或更聪明的办法来查询一个概念或要素?- 上一次尝试查询一个概念或要素时是否失败了? 如果是,是否可以尝试从另一个资源中再次查询?- 反思,我是否有更合理的方法来查询一个概念或要素?B. 根据以上分析,排列子要素间的查询优先级C. 找出当前需要获得取值的子要素D. 不可以使用“假设”:不要对要素的取值/定义做任何假设,确保你的信息全部来自明确的数据源!
**推理**: 根据你的反思与思考,一步步推理被选择的子要素取值的获取方式。如果前一次的计划失败了,请检查输入中是否包含每个概念/要素的明确定义,并尝试细化你的查询描述。
**计划**: 详细列出当前动作的执行计划。只计划一步的动作。PLAN ONE STEP ONLY!
**计划校验**: 按照一些步骤一步步分析A. 有哪些已知常量可以直接代入此次分析。B. 当前计划是否涉及穷举一个文件中的每条记录?- 如果是,请给出一个更有效的方法,比如按某条件筛选,从而减少计算量;- 否则,请继续下一步。C. 上述分析是否依赖某个要素的取值/定义,且该要素的取值/定义尚未获得?如果是,重新规划当前动作,确保所有依赖的要素的取值/定义都已经获得。D. 当前计划是否对要素的取值/定义做任何假设?如果是,请重新规划当前动作,确保你的信息全部来自对给定的数据源的历史分析,或尝试重新从给定数据源中获取相关信息。E. 如果全部子任务已完成,请用FINISH动作结束任务。
**计划改进**:A. 如何计划校验中的某一步骤无法通过,请改进你的计划;B. 如果你的计划校验全部通过,按(2)输出你的计划;C. 如果全部子任务已完成,请用FINISH动作结束任务。(2) 最后,输出你选择执行的动作/工具
{format_instructions}请确保每次选择动作/工具前你都先以文字输出了你的思考分析过程。
请确保你的动作/工具选择(JSON)出现在输出的最后一部分。
请确保你输出的JSON代码块以```json\n\n```包裹。
从上面可以看出写一个提示词可以包含
- 输入任务
- 参考资料目录
- 可以用工具
- 遵循约束
- 任务执行记录
- 输出形式
(1)关键概念
(2)概念拆解
(3)反思
(4)思考
(5)推理
(6)计划
(7)计划校验
(8)计划改进
(9)…
- 工具prompt
你的任务是先分析,再生成代码请根据用户的输入,一步步分析:
(1)用户的输入是否依赖某个条件,而这个条件没有明确赋值?
(2)我是否需要对某个变量的值做假设?如果我需要对某个变量的值做假设,请直接输出:
```python
print("我需要知道____的值,才能生成代码。请完善你的查询。") # 请将____替换为需要假设的的条件
^```否则,生成一段Python代码,分析指定文件的内容。你可以使用的库只包括:Pandas, re, math, datetime, openpyxl
确保你的代码只使用上述库,否则你的代码将无法运行。给定文件为:
{filename}文件内容样例:
{inspections}你输出的Python代码前后必须有markdown标识符,如下所示:
```python
# example code
print('hello world')
^```确保你的代码是可以运行的,文件名直接写死在代码里即可。
你生成代码中所有的常量都必须来自我给你的信息或来自文件本身。不要编造任何常量。
如果常量缺失,你的代码将无法运行。你可以拒绝生成代码,但是不要生成编造的代码。
确保你生成的代码最终以print的方式输出结果(回答用户的问题)。用户输入:
{query}
编写agent
- 编写大模型输出Action格式
from pydantic import BaseModel, Field
from typing import Optional, Dict, Anyclass Action(BaseModel):name: str = Field(description="Tool name")args: Optional[Dict[str, Any]] = Field(description="Tool input arguments, containing arguments names and values")def __str__(self):ret = f"Action(name={self.name}"if self.args:for k, v in self.args.items():ret += f", {k}={v}"ret += ")"return ret
- 智能体主流程
import re
from typing import List, Tuplefrom langchain_community.chat_message_histories.in_memory import ChatMessageHistory
from langchain_core.language_models.chat_models import BaseChatModel
from langchain.output_parsers import PydanticOutputParser, OutputFixingParser
from langchain.schema.output_parser import StrOutputParser
from langchain.tools.base import BaseTool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import render_text_description
from pydantic import ValidationError
from langchain_core.prompts import HumanMessagePromptTemplatefrom autogpt.Agent.Action import Action
from autogpt.Utils.CallbackHandlers import *
from autogpt.Utils.PrintUtils import THOUGHT_COLORclass ReActAgent:"""AutoGPT:基于Langchain实现"""@staticmethoddef __format_thought_observation(thought: str, action: Action, observation: str) -> str:# 将全部JSON代码块替换为空ret = re.sub(r'```json(.*?)```', '', thought, flags=re.DOTALL)ret += "\n" + str(action) + "\n返回结果:\n" + observationreturn ret@staticmethoddef __extract_json_action(text: str) -> str | None:# 匹配最后出现的JSON代码块json_pattern = re.compile(r'```json(.*?)```', re.DOTALL)matches = json_pattern.findall(text)if matches:last_json_str = matches[-1]return last_json_strreturn Nonedef __init__(self,llm: BaseChatModel,tools: List[BaseTool],work_dir: str,main_prompt_file: str,max_thought_steps: Optional[int] = 10,):self.llm = llmself.tools = toolsself.work_dir = work_dirself.max_thought_steps = max_thought_steps# OutputFixingParser: 如果输出格式不正确,尝试修复self.output_parser = PydanticOutputParser(pydantic_object=Action)self.robust_parser = OutputFixingParser.from_llm(parser=self.output_parser,llm=llm)self.main_prompt_file = main_prompt_fileself.__init_prompt_templates()self.__init_chains()self.verbose_handler = ColoredPrintHandler(color=THOUGHT_COLOR)def __init_prompt_templates(self):with open(self.main_prompt_file, 'r', encoding='utf-8') as f:self.prompt = ChatPromptTemplate.from_messages([MessagesPlaceholder(variable_name="chat_history"),HumanMessagePromptTemplate.from_template(f.read()),]).partial(work_dir=self.work_dir,tools=render_text_description(self.tools),tool_names=','.join([tool.name for tool in self.tools]),format_instructions=self.output_parser.get_format_instructions(),)def __init_chains(self):# 主流程的chainself.main_chain = (self.prompt | self.llm | StrOutputParser())def __find_tool(self, tool_name: str) -> Optional[BaseTool]: