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 "达到最大迭代次数，任务未完成"