手撕 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]:
命令詳解：tar)
