【智能体Agent】ReAct智能体的实现思路和关键技术

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

import re
from typing import List, Tuple

from 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 HumanMessagePromptTemplate

from Agent.Action import Action
from Utils.CallbackHandlers import *


class ReActAgent:
    """AutoGPT：基于Langchain实现"""

    @staticmethod
    def __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" + observation
        return ret

    @staticmethod
    def __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_str
        return None

    def __init__(
            self,
            llm: BaseChatModel,
            tools: List[BaseTool],
            work_dir: str,
            main_prompt_file: str,
            max_thought_steps: Optional[int] = 10,
    ):
        self.llm = llm
        self.tools = tools
        self.work_dir = work_dir
        self.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_file

        self.__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):
        # 主流程的chain
        self.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 tool
        return None

    def __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, response

    def __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 observation

    def run(
            self,
            task: str,
            chat_history: ChatMessageHistory,
            verbose=False
    ) -> str:
        """
        运行智能体
        :param task: 用户任务
        :param chat_history: 对话上下文（长时记忆）
        :param verbose: 是否显示详细信息
        """
        # 初始化短时记忆: 记录推理过程
        short_term_memory = []

        # 思考步数
        thought_step_count = 0

        reply = ""

        # 开始逐步思考
        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 += 1

        if 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
)