Spaces:
Sleeping
Sleeping
File size: 1,298 Bytes
509d1ca |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 |
# coding: utf-8
# Copyright (c) 2025 inclusionAI.
from typing import Any, Dict, Union, List
from examples.tools.common import Agents, Tools
from aworld.core.agent.base import AgentFactory
from aworld.config.conf import AgentConfig, ConfigDict
from aworld.agents.llm_agent import Agent
from aworld.core.common import Observation, ActionModel
class GymDemoAgent(Agent):
"""Example agent"""
def __init__(self, conf: Union[Dict[str, Any], ConfigDict, AgentConfig], **kwargs):
super(GymDemoAgent, self).__init__(conf, **kwargs)
def policy(self, observation: Observation, info: Dict[str, Any] = {}, **kwargs) -> Union[
List[ActionModel], None]:
import numpy as np
env_id = observation.info.get('env_id')
if env_id and env_id != 'CartPole-v1':
raise ValueError("Unsupported env")
res = np.random.randint(2)
action = [ActionModel(agent_name=self.id(), tool_name=Tools.GYM.value, action_name="play", params={"result": res})]
if observation.info.get("done"):
self._finished = True
return action
async def async_policy(self, observation: Observation, info: Dict[str, Any] = {}, **kwargs) -> Union[
List[ActionModel], None]:
return self.policy(observation, info, **kwargs)
|