【智能體Agent】ReAct智能體的實現思路和關鍵技術

基于ReAct（Reasoning + Acting）框架的自主智能體

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 Agent.Action import Action
from Utils.CallbackHandlers import *class 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]:for tool in self.tools:if tool.name == tool_name:return toolreturn Nonedef __step(self,task,short_term_memory,chat_history,verbose=False) -> Tuple[Action, str]:"""執行一步思考"""inputs = {"input": task,"agent_scratchpad": "\n".join(short_term_memory),"chat_history": chat_history.messages,}config = {"callbacks": [self.verbose_handler]if verbose else []}response = ""for s in self.main_chain.stream(inputs, config=config):response += s# 提取JSON代碼塊json_action = self.__extract_json_action(response)# 帶容錯的解析action = self.robust_parser.parse(json_action if json_action else response)return action, responsedef __exec_action(self, action: Action) -> str:# 查找工具tool = self.__find_tool(action.name)if tool is None:observation = (f"Error: 找不到工具或指令 '{action.name}'. "f"請從提供的工具/指令列表中選擇，請確保按對頂格式輸出。")else:try:# 執行工具observation = tool.run(action.args)except ValidationError as e:# 工具的入參異常observation = (f"Validation Error in args: {str(e)}, args: {action.args}")except Exception as e:# 工具執行異常observation = f"Error: {str(e)}, {type(e).__name__}, args: {action.args}"return observationdef run(self,task: str,chat_history: ChatMessageHistory,verbose=False) -> str:"""運行智能體:param task: 用戶任務:param chat_history: 對話上下文（長時記憶）:param verbose: 是否顯示詳細信息"""# 初始化短時記憶: 記錄推理過程short_term_memory = []# 思考步數thought_step_count = 0reply = ""# 開始逐步思考while thought_step_count < self.max_thought_steps:if verbose:self.verbose_handler.on_thought_start(thought_step_count)# 執行一步思考action, response = self.__step(task=task,short_term_memory=short_term_memory,chat_history=chat_history,verbose=verbose,)# 如果是結束指令，執行最后一步if action.name == "FINISH":reply = self.__exec_action(action)break# 執行動作observation = self.__exec_action(action)if verbose:self.verbose_handler.on_tool_end(observation)# 更新短時記憶short_term_memory.append(self.__format_thought_observation(response, action, observation))thought_step_count += 1if thought_step_count >= self.max_thought_steps:# 如果思考步數達到上限，返回錯誤信息reply = "抱歉，我沒能完成您的任務。"# 更新長時記憶chat_history.add_user_message(task)chat_history.add_ai_message(reply)return reply

結合LangChain框架和工具調用能力來逐步解決用戶任務。

以下是代碼的逐模塊解析：

1. 核心結構

class ReActAgent:"""AutoGPT：基于Langchain實現"""

核心類：實現了ReAct范式（推理+行動循環）的自主智能體
核心能力：
- 多步思考推理
- 工具調用
- 長短期記憶管理
- 異常處理與自我修正

2. 關鍵靜態方法

@staticmethod
def __format_thought_observation(...):  # 格式化思考記錄
@staticmethod
def __extract_json_action(...):         # 提取JSON動作

功能：
- __format_thought_observation：將思考過程、動作執行和觀察結果格式化為可讀文本，存入短期記憶
- __extract_json_action：用正則表達式提取模型輸出中的最后一個JSON代碼塊（確保獲取最新動作）

3. 初始化模塊

def __init__(...):# 核心組件初始化self.llm = llm                    # 大語言模型self.tools = tools                # 可用工具列表self.work_dir = work_dir          # 工作目錄self.max_thought_steps = ...      # 最大思考步數# 輸出解析系統self.output_parser = PydanticOutputParser(pydantic_object=Action)self.robust_parser = OutputFixingParser.from_llm(...)# 提示工程self.__init_prompt_templates()self.__init_chains()

關鍵技術點：
- 雙解析器機制：OutputFixingParser可在格式錯誤時自動修復輸出
- Pydantic驗證：確保動作符合預定義結構（Action模型）
- 工具描述渲染：render_text_description將工具轉化為自然語言描述

4. 提示工程系統

def __init_prompt_templates(self):with open(self.main_prompt_file) as f:self.prompt = ChatPromptTemplate.from_messages(...).partial(tools=...,             # 工具描述tool_names=...,        # 工具名稱列表format_instructions=..., # 格式說明)

核心要素：
- 動態加載提示模板文件
- 包含：
  - 聊天歷史占位符
  - 工具使用說明
  - 輸出格式要求
  - 工作目錄上下文

5. 執行流程控制

def run(...):while thought_step_count < self.max_thought_steps:# 單步思考action, response = self.__step(...)if action.name == "FINISH":break# 執行動作observation = self.__exec_action(action)# 記憶更新short_term_memory.append(...)

ReAct循環：
1. Reasoning：生成思考與動作（__step）
2. Acting：執行工具調用（__exec_action）
3. Observing：記錄執行結果
4. Loop：直到達到終止條件

6. 關鍵技術實現

6.1 單步推理 (`__step`)

def __step(...):inputs = {"input": task,"agent_scratchpad": "\n".join(short_term_memory),"chat_history": chat_history.messages,}# 流式處理LLM輸出for s in self.main_chain.stream(inputs):response += s# 提取并解析動作json_action = self.__extract_json_action(response)action = self.robust_parser.parse(...)

輸入組成：
- 任務目標
- 短期記憶（推理過程）
- 長期記憶（聊天歷史）
流式處理：實時顯示思考過程
錯誤恢復：自動修復格式錯誤的JSON輸出

6.2 動作執行 (`__exec_action`)

def __exec_action(...):tool = self.__find_tool(action.name)try:observation = tool.run(action.args)except ValidationError:# 參數驗證錯誤處理except Exception:# 通用錯誤處理

異常處理機制：
- 工具不存在
- 參數驗證錯誤
- 運行時異常
觀察反饋：將錯誤信息轉化為自然語言，供后續推理使用

7. 記憶系統

# 短期記憶
short_term_memory = []  # 存儲格式化的推理過程# 長期記憶
chat_history = ChatMessageHistory()  # 保存完整對話記錄

記憶類型：
- 短期記憶：當前任務的推理過程（最多保留max_thought_steps步）
- 長期記憶：跨會話的完整對話歷史

8. 關鍵設計亮點

自愈式輸出解析：
- 通過OutputFixingParser實現格式錯誤自動修復
- 示例場景：當LLM返回非法JSON時，自動嘗試修正

漸進式推理：

# 示例輸出格式
Thought: 我需要先查找用戶信息
Action: {"name": "user_search", "args": {"id": 123}}
Observation: 用戶張三，年齡30歲

通過agent_scratchpad維護推理上下文

工具發現機制：
- 動態渲染工具描述到提示詞
- 支持工具的熱插拔
多級異常處理：
- 工具不存在
- 參數驗證錯誤
- 執行時異常
- 最大步數限制

9. 使用示例

# 初始化組件
llm = ChatOpenAI()
tools = [SearchTool(), Calculator()]
agent = ReActAgent(llm, tools, work_dir="/data")# 執行任務
result = agent.run(task="計算馬云當前年齡的平方根",chat_history=ChatMessageHistory(),verbose=True
)