import os import re from typing import List, Optional, Tuple import pandas as pd 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,JsonOutputToolsParser 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 class ReActAgent: @staticmethod def extract_json_from_action_rst(text: str) -> str | None: """ 从RST格式文本中提取JSON代码块内容 该函数使用正则表达式匹配文本中的json代码块，并返回最后一个匹配到的JSON字符串。 参数: text (str): 包含RST格式的输入文本，可能包含代码块 返回值: str | None: 如果找到JSON代码块则返回其中的内容（不包括json和标记）， 如果没有找到则返回None """ # 使用正则表达式匹配json代码块 pattern = re.compile(r'```json(.*?)```', flags=re.DOTALL) matches = pattern.findall(text) if matches: json_str = matches[-1] return json_str return None @staticmethod def format_observations(response: str, action: Action, observation: str) -> str: """ 格式化观察结果，去除响应中的JSON代码块，并将动作和观察结果追加到响应末尾 参数: response (str): 原始响应字符串，可能包含JSON代码块 action (Action): 执行的动作对象 observation (str): 动作执行后的观察结果 返回值: str: 格式化后的字符串，包含清理后的响应内容、动作信息和观察结果 """ rst = re.sub(r"```json(.*?)```", "", response, flags=re.DOTALL) rst += "\n" + str(action) + "\n返回结果为：\n" + observation return rst def __init__(self, llm: BaseChatModel, tools: List[BaseTool], main_prompt_file_path: str): """ 初始化智能体构造器 参数: llm (BaseChatModel): 基础聊天模型实例 tools (List[BaseTool]): 可用工具列表 main_prompt_file_path (str): 主提示词文件路径 """ self.llm = llm self.tools = tools self.output_parser = PydanticOutputParser(pydantic_object=Action) try: self.robust_parser = OutputFixingParser.from_llm( parser=self.output_parser, llm=llm ) except Exception as e: print(f"不能修复的输出. 错误: {e}") print("请尝试提升提示词质量或者使用一个更加合适的大模型") self.main_prompt_file_path = main_prompt_file_path self.__init_prompts() self.__init_chains() def __init_prompts(self): """ 初始化提示模板 该方法从指定文件路径读取主提示内容，并构建聊天提示模板。 模板包含聊天历史占位符和人类消息模板，同时将工具信息、工具名称列表 和输出格式指令作为部分参数绑定到模板中。 """ with open(os.path.join(os.getcwd(), self.main_prompt_file_path), 'r', encoding='utf-8') as f: self.prompt = ChatPromptTemplate.from_messages([ MessagesPlaceholder(variable_name="chat_history"), HumanMessagePromptTemplate.from_template(f.read()), ]).partial( 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): """ 初始化主链对象 该方法创建一个由提示词、语言模型和字符串输出解析器组成的处理链， 用于处理输入并生成相应的文本输出。 """ self.main_chains = (self.prompt | self.llm | StrOutputParser()) def confirm_tool(self, tool_name: str) -> Optional[BaseTool]: """ 根据工具名称查找并返回对应的工具对象 参数: tool_name (str): 要查找的工具名称 返回值: Optional[BaseTool]: 如果找到匹配的工具则返回该工具对象，否则返回None """ # print(f'confirm_tool:{tool_name}') for tool in self.tools: if tool.name == tool_name: return tool return None def step_by_step(self, task, short_term_memory=None, chat_history=None, verbose=False) -> Tuple[Action, str]: """ 执行逐步任务处理，将任务分解为多个步骤并生成相应的动作和响应 参数: task (str): 需要处理的任务描述 short_term_memory (list, optional): 短期记忆列表，包含之前的执行步骤 chat_history (object, optional): 聊天历史对象，包含对话记录 verbose (bool): 是否启用详细输出模式 返回值: Tuple[Action, str]: 返回一个元组，包含解析后的动作对象和完整的响应字符串 """ inputs = { "input": task, "agent_scratchpad": "\n".join(short_term_memory), "chat_history": chat_history.messages if chat_history else chat_history, } response = "" for s in self.main_chains.stream(inputs): response += s print(f'response:{response}') action_rst_json = self.extract_json_from_action_rst(response) print(f'json_action:{action_rst_json}') action = self.robust_parser.parse( action_rst_json if isinstance(action_rst_json, dict) else response ) print(f'action:{action}') return action, response def exec_actions(self, action: Action) -> str: """ 执行指定的动作并返回观察结果 参数: action (Action): 要执行的动作对象，包含工具名称和参数 返回值: str: 动作执行后的观察结果或错误信息 """ # print(f'action in exec_actions:{action}, action tool_name in exec_actions:{action.tool_name}') tool = self.confirm_tool(action.tool_name) # print(f'tool in exec_actions:{tool}') if tool is None: observation = ( f"Error:没找到待执行的工具或者指令 '{action.tool_name}'." f"请从提供的工具或者指令列表中选择，注意按对顶格式输出内容。" ) else: try: print(f'tool 执行：{action.args}') observation = tool.run(action.args) print(f'tool执行结果：{observation}') except ValidationError as e: observation = ( f"执行动作时参数有非法错误 in exec_actions：{str(e)},args:{action.args}" ) print(f'action in exec_actions:{action.args}') except Exception as e: observation = f"Error in exec_actions:{str(e)},{type(e).__name__},args:{action.args}" return observation # 基础版 # def execute(self, task: str, uuid: int, chat_history: ChatMessageHistory, verbose=False) -> str: # """ # 执行任务的主要方法 # 参数: # task (str): 需要执行的任务描述 # uuid (int): 任务的唯一标识符 # chat_history (ChatMessageHistory): 聊天历史记录对象 # verbose (bool): 是否启用详细输出模式，默认为False # 返回值: # str: 执行任务后得到的答案 # """ # short_term_memory = [] # action, response = self.step_by_step(task=task, short_term_memory=short_term_memory, # chat_history=chat_history, verbose=verbose) # print(f'execute方法:action:{action},response:{response}') # # answer = self.exec_actions(action) # print(f'answer:{answer}') # # short_term_memory.append(self.format_observations(response, action, answer)) # # chat_history.add_user_message(task) # chat_history.add_ai_message(answer) # return answer # 改进版 def execute(self, task: str, uuid: int, chat_history: ChatMessageHistory, verbose=False) -> str: """执行任务的主循环""" short_term_memory = [] max_iterations = 10 # 防止无限循环 iteration = 0 while iteration < max_iterations: iteration += 1 # 1. 思考：让LLM决定下一步行动 action, response = self.step_by_step( task=task, short_term_memory=short_term_memory, chat_history=chat_history, verbose=verbose ) # 2. 判断是否应该终止（当action为"finish"时） if action.tool_name == "finish" or action.tool_name == "Final Answer": return action.args.get("answer", "任务完成") # 3. 执行行动 observation = self.exec_actions(action) # 4. 观察：记录结果到短期记忆 formatted_obs = self.format_observations(response, action, observation) short_term_memory.append(formatted_obs) if verbose: print(f"Step {iteration}: {action.tool_name} -> {observation[:100]}...") return "达到最大迭代次数，任务未完成"