对于LLM代理的许多应用程序,环境是真实的(互联网、数据库、REPL等)。然而,我们也可以定义代理在模拟环境中进行交互,比如基于文本的游戏。这是一个使用Gymnasium(之前是OpenAI Gym)创建简单代理-环境交互循环的示例。
# 安装gym库
!pip install gym
# 导入必要的模块
import tenacity # 引入tenacity模块
from langchain.output_parsers import RegexParser # 从langchain.output_parsers模块中导入RegexParser类
from langchain.schema import ( # 从langchain.schema模块中导入HumanMessage和SystemMessage类
HumanMessage,
SystemMessage,
)
class GymnasiumAgent:
@classmethod
def get_docs(cls, env):
return env.unwrapped.__doc__ # 获取环境的文档字符串
def __init__(self, model, env):
self.model = model # 模型
self.env = env # 环境
self.docs = self.get_docs(env) # 获取环境的文档字符串
self.instructions = """
你的目标是最大化你的回报,即你所获得的奖励的总和。
我会给你一个观察值、奖励、终止标志、截断标志和迄今为止的回报,格式如下:
观察值: <observation>
奖励: <reward>
终止: <termination>
截断: <truncation>
回报: <sum_of_rewards>
你需要用一个动作来回应,格式如下:
动作: <action>
你需要用你的实际动作替换 <action>。
除此之外,不要做任何其他操作,只需返回动作。
"""
self.action_parser = RegexParser(
regex=r"Action: (.*)", output_keys=["action"], default_output_key="action"
) # 用于解析动作的正则表达式解析器
self.message_history = [] # 消息历史记录
self.ret = 0 # 回报
def random_action(self):
action = self.env.action_space.sample() # 随机选择一个动作
return action
def reset(self):
self.message_history = [
SystemMessage(content=self.docs), # 添加环境的文档字符串到消息历史记录中
SystemMessage(content=self.instructions), # 添加指令到消息历史记录中
]
def observe(self, obs, rew=0, term=False, trunc=False, info=None):
self.ret += rew # 更新回报
obs_message = f"""
观察值: {obs}
奖励: {rew}
终止: {term}
截断: {trunc}
回报: {self.ret}
""" # 构建观察消息
self.message_history.append(HumanMessage(content=obs_message)) # 添加观察消息到消息历史记录中
return obs_message
def _act(self):
act_message = self.model.invoke(self.message_history) # 使用模型进行动作选择
self.message_history.append(act_message) # 添加动作消息到消息历史记录中
action = int(self.action_parser.parse(act_message.content)["action"]) # 解析动作消息中的动作
return action
def act(self):
try:
for attempt in tenacity.Retrying(
stop=tenacity.stop_after_attempt(2), # 最多重试2次
wait=tenacity.wait_none(), # 重试之间没有等待时间
retry=tenacity.retry_if_exception_type(ValueError), # 当遇到 ValueError 异常时重试
before_sleep=lambda retry_state: print(
f"ValueError occurred: {retry_state.outcome.exception()}, retrying..."
), # 打印错误信息并重试
):
with attempt:
action = self._act() # 执行动作选择
except tenacity.RetryError:
action = self.random_action() # 如果重试失败,则随机选择一个动作
return action # 返回选择的动作
# 创建一个名为env的Blackjack-v1环境
env = gym.make("Blackjack-v1")
# 使用ChatOpenAI模型创建一个名为agent的GymnasiumAgent代理,并设置temperature参数为0.2,同时将env环境传入
agent = GymnasiumAgent(model=ChatOpenAI(temperature=0.2), env=env)
# 导入必要的库
import gym
# 创建环境
env = gym.make('CartPole-v0')
# 重置环境并获取初始观测和信息
observation, info = env.reset()
# 重置智能体
agent.reset()
# 使用智能体观测环境并获取观测信息
obs_message = agent.observe(observation)
print(obs_message)
# 进入主循环
while True:
# 智能体选择动作
action = agent.act()
# 执行动作并获取新的观测、奖励、终止、截断和信息
observation, reward, termination, truncation, info = env.step(action)
# 使用智能体观测环境并获取观测信息
obs_message = agent.observe(observation, reward, termination, truncation, info)
print(f"Action: {action}")
print(obs_message)
# 如果达到终止或截断条件,则跳出循环
if termination or truncation:
print("break", termination, truncation)
break
# 关闭环境
env.close()
Observation: (15, 4, 0) Reward: 0 Termination: False Truncation: False Return: 0 Action: 1 Observation: (25, 4, 0) Reward: -1.0 Termination: True Truncation: False Return: -1.0 break True False