diff --git a/README.md b/README.md index 547bc67..346197c 100644 --- a/README.md +++ b/README.md @@ -1,198 +1,64 @@ -# Virtuals Python SDK Library -The Virtuals Python SDK is a library that allows you to configure and deploy agents on the Virtuals platform. This SDK/API allows you to configure your agents powered by the GAME architecture. This is similar to configuring your agent in the [Agent Sandbox](https://game-lite.virtuals.io/) on the [Virtuals Platform](https://app.virtuals.io/). +# Virtuals Python SDK Library [WARNING: MIGRATING - WILL BE DEPRECATED] +THIS REPOSITORY WILL NO LONGER BE MAINTAINED AND HAS BEEN MIGRATED OVER TO https://github.com/game-by-virtuals/game-python/tree/main. Please move over to use latest SDK and features. -This also provides a developer-friendly interface to develop and power your custom applications with Virtuals Agents using GAME. +The Virtuals Python SDK is a library that allows you interact with the Virtuals Platform. -## Documentation -Detailed documentation to better understand the configurable components and the GAME architecture can be found on [agent configurations](https://www.notion.so/1592d2a429e98016b389ea26b53686a3). - -## Installation -```bash -pip install virtuals_sdk -``` - -## Create an API key -Open the [Virtuals Platform](https://app.virtuals.io/) and create/get an API key from the Agent Sandbox by clicking “Access G.A.M.E API” - -![getGAMEApi](./docs/imgs/accesskey.png) - -Store the key in a safe location, like a `.bashrc` or a `.zshrc` file. - -```bash -export VIRTUALS_API_KEY="your_virtuals_api_key" -``` - -Alternatively, you can also use a `.env` file ([`python-dotenv` package](https://github.com/theskumar/python-dotenv) to store and load the key) if you are using the Virtuals Python SDK. - -## Usage (GAME) -The Virtuals SDK current main functionalities are to develop and configure agents powered by GAME. Other functionalities to interact with the Virtuals Platform will be supported in the future. This GAME SDK can be used for multiple use cases: - -1. Develop, evaluate and update the existing Agent in Twitter environment. -2. Build on other platforms and application using GAME (Task-based Agent). - -### Update the existing Agent in Twitter environment -The SDK provides another interface to configure agents that is more friendly to developers. This is similar to configuring your agent in the [Agent Sandbox](https://game-lite.virtuals.io/). - -```python -from virtuals_sdk.game import Agent - -# Create agent with just strings for each component -agent = Agent( - api_key=VIRTUALS_API_KEY, - goal="Autonomously analyze crypto markets and provide trading insights", - description="HODL-9000: A meme-loving trading bot powered by hopium and ramen", - world_info="Virtual crypto trading environment where 1 DOGE = 1 DOGE" -) -``` -You can also initialize the agent first with just the API key and set the goals, descriptions and world information separately and check the current agent descriptions if needed. - -```python -agent = Agent(api_key=VIRTUALS_API_KEY) - -# check what is current goal, descriptions and world_info -agent.get_goal() -agent.get_description() -agent.get_world_info() - -# Set components individually - set change the agent goal/description/worldinfo -agent.set_goal("Autonomously analyze crypto markets and provide trading insights") -agent.set_description("HODL-9000: A meme-loving trading bot powered by hopium and ramen") -agent.set_world_info("Virtual crypto trading environment where 1 DOGE = 1 DOGE") - -# check what is current goal, descriptions and world_info -agent.get_goal() -agent.get_description() -agent.get_world_info() -``` - -### Functions -By default, there are no functions enabled when the agent is initialized (i.e. the agent has no actions/functions it can execute). There are a list of available and provided functions for the Twitter/X platform and you can set them. +## About G.A.M.E. +GAME is a modular agentic framework which enables an agent to plan actions and make decisions autonomously based on information provided to it. -```python -print(agent.list_available_default_twitter_functions()) +Please refer to our [whitepaper](https://whitepaper.virtuals.io/developer-documents/game-framework) for more information and resources. -# enable some functions for the agent to use -agent.use_default_twitter_functions(["wait", "reply_tweet"]) -``` - -You can then equip the agent with some custom functions as follows: -```python +## About Virtuals Python SDK +Currently, this SDK allows you to develop your agents powered by the GAME architecture in its most fullest and most flexible form. -search_function = game.Function( - fn_name="custom_search_internet", - fn_description="search the internet for the best songs", - args=[ - game.FunctionArgument( - name="query", - type="string", - description="The query to search for" - ) - ], - config=game.FunctionConfig( - method="get", - url="https://google.com", - platform="twitter", # specify which platform this function is for, in this case this function is for twitter only - success_feedback="I found the best songs", - error_feedback="I couldn't find the best songs", - ) - ) - -# adding custom functions only for platform twitter -agent.add_custom_function(search_function) -``` - -### Evaluate with Simulate, Deploy -You can simulate one step of the agentic loop on Twitter/X with your new configurations and see the outputs. This is similar to the simulate button on the [Agent Sandbox](https://game-lite.virtuals.io/). - -```python -# Simulate one step of the full agentic loop on Twitter/X from the HLP -> LLP -> action (NOTE: supported for Twitter/X only now) -response = agent.simulate_twitter(session_id="123") -``` -To more realistically simulate deployment, you can also run through the simulate function with the same session id for a number of steps. - -```python -sid = "456" -num_steps = 10 -for i in range(num_steps): - response = agent.simulate_twitter(session_id=sid) -``` - -```python -# Simulate response to a certain event -response = agent.react( - session_id="567", # string identifier that you decide - tweet_id="xxxx", - platform="twitter", -) -``` - -Once you are happy, `deploy_twitter` will push your agent configurations to production and run your agent on Twitter/X autonomously. -```python -# deploy agent! (NOTE: supported for Twitter/X only now) -agent.deploy_twitter() -``` +![New SDK visual](docs/imgs/new_sdk_visual.png) +The python SDK is made up of 3 main components (Agent, Worker, function), each with configurable arguments. -## Build on other platforms using GAME -`simulate_twitter` and `deploy_twitter` runs through the entire GAME stack from HLP → LLP→ action/function selected. However, these agent functionalities are currently for the Twitter/X platform. You may utilize Task-based Agent with Low-Level Planner and Reaction Module to develop applications that are powered by GAME. The Low Level Planner (LLP) of the agent (please see [documentation](https://www.notion.so/1592d2a429e98016b389ea26b53686a3?pvs=21) for more details on GAME and LLP) can separately act as a decision making engine based on a task description and event occurring. This agentic architecture is simpler but also sufficient for many applications. +Agent (a.k.a. [high level planner](https://whitepaper.virtuals.io/developer-documents/game-framework/game-overview#high-level-planner-hlp-context)) +- Takes in a Goal + - Drives the agents behaviour through the high level plan which influences the thinking and creation of tasks that would contribute towards this goal +- Takes in a Description + - Combination of what was previously known as World Info + Agent Description + - This include a description of the "world" the agent lives in, and the personality and background of the agent -We are releasing this simpler setup as a more generalised/platform agnostic framework (not specific to Twitter/X). The entire GAME stack along with the HLP will be opened up to be fully configurable and platform agnostic in the coming weeks. +Worker (a.k.a. [low-level planner](https://whitepaper.virtuals.io/developer-documents/game-framework/game-overview#low-level-planner-llp-context)) +- Takes in a Description + - Used to control which workers are called by the agent, based on the high-level plan and tasks created to contribute to the goal -### 🖥️ Low-Level Planner (LLP) as a Task-based Agent +Function +- Takes in a Description + - Used to control which functions are called by the workers, based on each worker's low-level plan + - This can be any python executable -![llp.png](./docs/imgs/llp.png) +## Features +- Develop your own custom agents for any application or platform. +- Ability to control your agents and workers via descriptions (prompts) +- Full control of what the agent sees (state) and can do (actions/functions) +- Ability to fully customise functions. This could include various combinations of programmed logic. For example: + - Calling an API to retrieve data + - Calling an API to retrieve data, followed by custom calculations or data processing logic in python code + - 2 API calls chained together (e.g. calling an API to retrieve web data, and then posting a tweet) -After configuring the agent’s character card or description and setting up the agents functions, we can then use the `react` method to get an agent to respond and execute a sequence of actions based on the task description provided and the context. Between each action in the sequence, the agent only receives the `success_feedback` and `error_feedback` of each function executed. +> ### ℹ️ Changes from older python SDK version (prior to 8 Jan 2025) +>![Old SDK visual](docs/imgs/old_sdk_visual.png) +> - Ability to fully customise functions (previously, each function was a single API call) +> - Ability to control the low-level planner via description prompt (previously, only the high-level planner and functions could be controlled via description prompts) +> - The description defined in the agent is equivalent to what was previously known as world information and agent description -```python -# React/respond to a certain event -response = agent.react( - session_id="567", # string identifier that you decide - task="Be friendly and help people who talk to you. Do not be rude.", - event="Hi how are you?", - platform="TELEGRAM", -) +## Installation +```bash +pip install virtuals_sdk ``` -> [!IMPORTANT] -> Remember that the `platform` tag determines what functions are available to the agent. The agent will have access to functions that have the same `platform` tag. All the default available functions listed on `agent.list_available_default_twitter_functions()` and set via `agent.use_default_twitter_functions()` have the `platform` tag of “twitter”. - -## Arguments Definition - -### Session ID -The session ID is an identifier for an instance of the agent. When using the same session ID, it maintains and picks up from where it last left off, continuing the session/instance. It should be split per user/ conversation that you are maintaining on your platform. For different platforms, different session ID can be used. - -### Platform Tag -When adding custom functions, and when calling the react agent (i.e. LLP), there is a platform tag that can be defined. This acts like a filter for the functions available that is passed to the agent. You should define the platform when passing in the events. +## Usage +Please refer to [`test_agent.py`](examples/game/test_agent.py) and [`test_worker.py`](examples/game/test_agent.py) for usage examples. -### Task Description -Task description serves as the prompt for the agent to respond. Since the reaction can be platform-based, you can define task description based on the platforms. In the task description, you should pass in any related info that require agent to make decision. That should include: -- User message -- Conversation history -- Instructions +## How to Contribute +Contributions are welcome, especially in the form of new plugins! We are working on creating a plugins repo, but in meantime - please contact us via [Twitter](https://x.com/GAME_Virtuals) or [Telegram](https://t.me/virtuaIs_io). +## Documentation +Detailed documentation to better understand the configurable components and the GAME architecture can be found on [here](https://whitepaper.virtuals.io/developer-documents/game-framework). -## Importing Functions and Sharing Functions -With this SDK and function structure, importing and sharing functions is also possible. Looking forward to all the different contributions and functionalities we will build together as a community! - -```python -from virtuals_sdk.functions.telegram import TelegramClient - -# define your token so that it can attach it to create the correspodning functions -tg_client = TelegramClient(bot_token="xxx") -print(tg_client.available_functions) - -# get functions -reply_message_fn = tg_client.get_function("send_message") -create_poll_fn = tg_client.get_function("create_poll") -pin_message_fn = tg_client.get_function("pin_message") - -# test the execution of functions -reply_message_fn("xxxxxxxx", "Hello World") -create_poll_fn("xxxxxxxx", "What is your favorite color?", ["Red", "Blue", "Green"], "True") -pin_message_fn("xxxxxxxx", "xx", "True") - -# add these functions to your agent -agent.add_custom_function(reply_message_fn) -agent.add_custom_function(create_poll_fn) -agent.add_custom_function(pin_message_fn) -``` \ No newline at end of file +## Useful Resources +- [TypeScript SDK](https://www.npmjs.com/package/@virtuals-protocol/game): The core logic of this SDK mirrors the logic of this python SDK if you prefer to develop your agents in TypeScript. +- [Twitter Agent](./src/virtuals_sdk/twitter_agent/README.md): The Virtuals Platform offers a out-of-the-box hosted agent that can be used to interact with the Twitter/X platform, powered by GAME. This agent comes with existing functions/actions that can be used to interact with the Twitter/X platform and can be immediately hosted/deployed as you configure it. This is similar to configuring your agent in the [Agent Sandbox](https://game-lite.virtuals.io/) on the [Virtuals Platform](https://app.virtuals.io/) but through a developer-friendly SDK interface. diff --git a/docs/imgs/new_sdk_visual.png b/docs/imgs/new_sdk_visual.png new file mode 100644 index 0000000..218366c Binary files /dev/null and b/docs/imgs/new_sdk_visual.png differ diff --git a/docs/imgs/old_sdk_visual.png b/docs/imgs/old_sdk_visual.png new file mode 100644 index 0000000..9387e96 Binary files /dev/null and b/docs/imgs/old_sdk_visual.png differ diff --git a/examples/game/test_agent.py b/examples/game/test_agent.py new file mode 100644 index 0000000..d2429ea --- /dev/null +++ b/examples/game/test_agent.py @@ -0,0 +1,206 @@ +from virtuals_sdk.game.agent import Agent, WorkerConfig +from virtuals_sdk.game.custom_types import Function, Argument +from virtuals_sdk.game.custom_types import FunctionResult, FunctionResultStatus +from typing import Tuple + +game_api_key="" + +# the get_worker_state_fn and the get_agent_state_fn can be the same function or diffferent +# each worker can also have a different get_worker_state_fn and maintain their own state +# or they can share the same get_worker_state_fn and maintain the same state + +def get_worker_state_fn(function_result: FunctionResult, current_state: dict) -> dict: + """ + This function will get called at every step of the workers execution to form the agent's state. + It will take as input the function result from the previous step. + """ + # dict containing info about the function result as implemented in the exectuable + info = function_result.info + + # example of fixed state (function result info is not used to change state) - the first state placed here is the initial state + init_state = { + "objects": [ + {"name": "apple", "description": "A red apple", "type": ["item", "food"]}, + {"name": "banana", "description": "A yellow banana", "type": ["item", "food"]}, + {"name": "orange", "description": "A juicy orange", "type": ["item", "food"]}, + {"name": "chair", "description": "A chair", "type": ["sittable"]}, + {"name": "table", "description": "A table", "type": ["sittable"]}, + ] + } + + if current_state is None: + # at the first step, initialise the state with just the init state + new_state = init_state + else: + # do something wiht the current state input and the function result info + new_state = init_state # this is just an example where the state is static + + return new_state + +def get_agent_state_fn(function_result: FunctionResult, current_state: dict) -> dict: + """ + This function will get called at every step of the agent's execution to form the agent's state. + It will take as input the function result from the previous step. + """ + + # example of fixed state (function result info is not used to change state) - the first state placed here is the initial state + init_state = { + "objects": [ + {"name": "apple", "description": "A red apple", "type": ["item", "food"]}, + {"name": "banana", "description": "A yellow banana", "type": ["item", "food"]}, + {"name": "orange", "description": "A juicy orange", "type": ["item", "food"]}, + {"name": "chair", "description": "A chair", "type": ["sittable"]}, + {"name": "table", "description": "A table", "type": ["sittable"]}, + ] + } + + if current_state is None: + # at the first step, initialise the state with just the init state + new_state = init_state + else: + # do something wiht the current state input and the function result info + new_state = init_state # this is just an example where the state is static + + return new_state + + +def take_object(object: str, **kwargs) -> Tuple[FunctionResultStatus, str, dict]: + """ + Function to take an object from the environment. + + Args: + object: Name of the object to take + **kwargs: Additional arguments that might be passed + """ + + if object: + return FunctionResultStatus.DONE, f"Successfully took the {object}", {} + return FunctionResultStatus.FAILED, "No object specified", {} + + +def throw_object(object: str, **kwargs) -> Tuple[FunctionResultStatus, str, dict]: + """ + Function to throw an object. + + Args: + object: Name of the object to throw + **kwargs: Additional arguments that might be passed + """ + if object: + return FunctionResultStatus.DONE, f"Successfully threw the {object}", {} + return FunctionResultStatus.FAILED, "No object specified", {} + + +def sit_on_object(object: str, **kwargs) -> Tuple[FunctionResultStatus, str, dict]: + """ + Function to sit on an object. + + Args: + object: Name of the object to sit on + **kwargs: Additional arguments that might be passed + """ + sittable_objects = {"chair", "bench", "stool", "couch", "sofa", "bed"} + + if not object: + return FunctionResultStatus.FAILED, "No object specified", {} + + if object.lower() in sittable_objects: + return FunctionResultStatus.DONE, f"Successfully sat on the {object}", {} + + return FunctionResultStatus.FAILED, f"Cannot sit on {object} - not a sittable object", {} + +def throw_fruit(object: str, **kwargs) -> Tuple[FunctionResultStatus, str, dict]: + """ + Function to throw fruits. + """ + fruits = {"apple", "banana", "orange", "pear", "mango", "grape"} + + if not object: + return FunctionResultStatus.ERROR, "No fruit specified", {} + + if object.lower() in fruits: + return FunctionResultStatus.DONE, f"Successfully threw the {object} across the room!", {} + return FunctionResultStatus.ERROR, f"Cannot throw {object} - not a fruit", {} + +def throw_furniture(object: str, **kwargs) -> Tuple[FunctionResultStatus, str, dict]: + """ + Function to throw furniture. + """ + furniture = {"chair", "table", "stool", "lamp", "vase", "cushion"} + + if not object: + return FunctionResultStatus.ERROR, "No furniture specified", {} + + if object.lower() in furniture: + return FunctionResultStatus.DONE, f"Powerfully threw the {object} across the room!", {} + return FunctionResultStatus.ERROR, f"Cannot throw {object} - not a furniture item", {} + + +# create functions for each executable +take_object_fn = Function( + fn_name="take", + fn_description="Take object", + args=[Argument(name="object", type="item", description="Object to take")], + executable=take_object +) + +sit_on_object_fn = Function( + fn_name="sit", + fn_description="Sit on object", + args=[Argument(name="object", type="sittable", description="Object to sit on")], + executable=sit_on_object +) + +throw_object_fn = Function( + fn_name="throw", + fn_description="Throw any object", + args=[Argument(name="object", type="item", description="Object to throw")], + executable=throw_object +) + +throw_fruit_fn = Function( + fn_name="throw_fruit", + fn_description="Throw fruit only", + args=[Argument(name="object", type="item", description="Fruit to throw")], + executable=throw_fruit +) + +throw_furniture_fn = Function( + fn_name="throw_furniture", + fn_description="Throw furniture only", + args=[Argument(name="object", type="item", description="Furniture to throw")], + executable=throw_furniture +) + + +# Create the specialized workers +fruit_thrower = WorkerConfig( + id="fruit_thrower", + worker_description="A worker specialized in throwing fruits ONLY with precision", + get_state_fn=get_worker_state_fn, + action_space=[take_object_fn, sit_on_object_fn, throw_fruit_fn] +) + +furniture_thrower = WorkerConfig( + id="furniture_thrower", + worker_description="A strong worker specialized in throwing furniture", + get_state_fn=get_worker_state_fn, + action_space=[take_object_fn, sit_on_object_fn, throw_furniture_fn] +) + +# Create agent with both workers +chaos_agent = Agent( + api_key=game_api_key, + name="Chaos", + agent_goal="Conquer the world by causing chaos.", + agent_description="You are a mischievous master of chaos is very strong but with a very short attention span, and not so much brains", + get_agent_state_fn=get_agent_state_fn, + workers=[fruit_thrower, furniture_thrower] +) + +# # interact and instruct the worker to do something +# chaos_agent.get_worker("fruit_thrower").run("make a mess and rest!") + +# # compile and run the agent - if you don't compile the agent, the things you added to the agent will not be saved +chaos_agent.compile() +chaos_agent.run() \ No newline at end of file diff --git a/examples/game/test_worker.py b/examples/game/test_worker.py new file mode 100644 index 0000000..d8dd772 --- /dev/null +++ b/examples/game/test_worker.py @@ -0,0 +1,115 @@ +from virtuals_sdk.game.worker import Worker +from virtuals_sdk.game.custom_types import Function, Argument +from virtuals_sdk.game.custom_types import FunctionResult, FunctionResultStatus +from typing import Tuple + +game_api_key="" + +def get_state_fn(function_result: FunctionResult, current_state: dict) -> dict: + """ + This function will get called at every step of the agent's execution to form the agent's state. + It will take as input the function result from the previous step. + """ + # dict containing info about the function result as implemented in the exectuable + info = function_result.info + + # example of fixed state (function result info is not used to change state) - the first state placed here is the initial state + init_state = { + "objects": [ + {"name": "apple", "description": "A red apple", "type": ["item", "food"]}, + {"name": "banana", "description": "A yellow banana", "type": ["item", "food"]}, + {"name": "orange", "description": "A juicy orange", "type": ["item", "food"]}, + {"name": "chair", "description": "A chair", "type": ["sittable"]}, + {"name": "table", "description": "A table", "type": ["sittable"]}, + ] + } + + if current_state is None: + # at the first step, initialise the state with just the init state + new_state = init_state + else: + # do something wiht the current state input and the function result info + new_state = init_state # this is just an example where the state is static + + return new_state + +def take_object(object: str, **kwargs) -> Tuple[FunctionResultStatus, str, dict]: + """ + Function to take an object from the environment. + + Args: + object: Name of the object to take + **kwargs: Additional arguments that might be passed + """ + if object: + return FunctionResultStatus.DONE, f"Successfully took the {object}", {} + return FunctionResultStatus.FAILED, "No object specified", {} + + +def throw_object(object: str, **kwargs) -> Tuple[FunctionResultStatus, str, dict]: + """ + Function to throw an object. + + Args: + object: Name of the object to throw + **kwargs: Additional arguments that might be passed + """ + if object: + return FunctionResultStatus.DONE, f"Successfully threw the {object}", {} + return FunctionResultStatus.FAILED, "No object specified", {} + + +def sit_on_object(object: str, **kwargs) -> Tuple[FunctionResultStatus, str, dict]: + """ + Function to sit on an object. + + Args: + object: Name of the object to sit on + **kwargs: Additional arguments that might be passed + """ + sittable_objects = {"chair", "bench", "stool", "couch", "sofa", "bed"} + + if not object: + return FunctionResultStatus.FAILED, "No object specified", {} + + if object.lower() in sittable_objects: + return FunctionResultStatus.DONE, f"Successfully sat on the {object}", {} + + return FunctionResultStatus.FAILED, f"Cannot sit on {object} - not a sittable object", {} + + +# Action space with all executables +action_space = [ + Function( + fn_name="take", + fn_description="Take object", + args=[Argument(name="object", type="item", description="Object to take")], + executable=take_object + ), + Function( + fn_name="throw", + fn_description="Throw object", + args=[Argument(name="object", type="item", description="Object to throw")], + executable=throw_object + ), + Function( + fn_name="sit", + fn_description="Take a seat", + args=[Argument(name="object", type="sittable", description="Object to sit on")], + executable=sit_on_object + ) +] + + + +worker = Worker( + api_key=game_api_key, + description="You are an evil NPC in a game.", + instruction="Choose the evil-est actions.", + get_state_fn=get_state_fn, + action_space=action_space +) + +# interact and instruct the worker to do something +# worker.run("what would you do to the apple?") +worker.run("take over the world with the things you have around you!") \ No newline at end of file diff --git a/examples/example-custom.py b/examples/twitter_agent/example-custom.py similarity index 96% rename from examples/example-custom.py rename to examples/twitter_agent/example-custom.py index a0f922c..17d7e23 100644 --- a/examples/example-custom.py +++ b/examples/twitter_agent/example-custom.py @@ -1,5 +1,5 @@ import os -from virtuals_sdk import game +from virtuals_sdk.twitter_agent import game agent = game.Agent( api_key=os.environ.get("VIRTUALS_API_KEY"), diff --git a/examples/example-twitter.py b/examples/twitter_agent/example-twitter.py similarity index 96% rename from examples/example-twitter.py rename to examples/twitter_agent/example-twitter.py index 750ed98..73acaf2 100644 --- a/examples/example-twitter.py +++ b/examples/twitter_agent/example-twitter.py @@ -1,5 +1,5 @@ import os -from virtuals_sdk import game +from virtuals_sdk.twitter_agent import game agent = game.Agent( api_key=os.environ.get("VIRTUALS_API_KEY"), diff --git a/pyproject.toml b/pyproject.toml index 5025904..a055e62 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "virtuals-sdk" -version = "0.1.2" +version = "0.1.6" authors = [ { name = "Your Name", email = "your.email@example.com" }, ] diff --git a/src/virtuals_sdk/game/README.md b/src/virtuals_sdk/game/README.md new file mode 100644 index 0000000..86973da --- /dev/null +++ b/src/virtuals_sdk/game/README.md @@ -0,0 +1,112 @@ +# GAME SDK + +## Overview + +This library enables the creation of AI agents, in which actions are defined by the creator/developer. These agents can be used: +- Worker Mode: to autonomously execute tasks through interaction with the agent. User interaction and tasks is required for the agent to execute tasks. For example, "Bring me some fruits and then go sit on the chair". +- Agent Mode: to autonomously function in an open-ended manner by just providing a general goal. The agent independently and continously decides tasks for itself in an open-ended manner, and attempts to complete them. + + +## Core Features + +### 1. Functions and Executables + +Functions define available actions for the agent: + +```python +my_function = Function( + fn_name="action_name", + fn_description="Description of action", + args=[Argument(name="param", type="type", description="param description")], + executable=function_implementation +) +``` +Executable functions must return a tuple of: +- FunctionResultStatus +- Message string +- Additional info dictionary + + +### 2. State Management + +Easy and flexible wayto define the state management, what the agent sees and how that changes. + +```python +def get_state_fn(function_result: FunctionResult, current_state: dict) -> dict: + """ + Updates state based on function execution results + + Args: + function_result: Result from last executed function + current_state: Current state dictionary or None for initialization + + Returns: + Updated state dictionary + """ + if current_state is None: + return initial_state + + # Update state based on function_result.info + new_state = update_state(current_state, function_result) + return new_state +``` + +Key features: +- Can be shared or unique per worker +- Processes function execution results to update state + +### 3. Workers +Workers are simple interactiable agents that exectue the tasks defined by the user. They can be specialized agents with defined capabilities: + +```python +worker = Worker( + api_key="your_api_key", + description="Worker description", + instruction="Default instructions", + get_state_fn=state_function, + action_space=[list_of_functions] +) + +worker.run("Bring me some fruits") +``` + + + +### 4. Agents + +Agents are used to autonomously function in an open-ended manner by just providing a general goal. Tasks are generated by the agent itself continuously, and the agent will attempt to complete them. You can provide many workers to the agent, and they will be used to execute the tasks. + +```python +agent = Agent( + api_key="your_api_key", + name="Agent Name", + agent_goal="Primary goal", + agent_description="Description", + get_agent_state_fn=agent_state_function, + workers=[worker1, worker2] +) + +# Compile and run +agent.compile() +agent.run() +``` + +Use WorkerConfig for agent composition: + +```python +worker_config = WorkerConfig( + id="worker_id", + worker_description="Description", + get_state_fn=state_function, + action_space=[function1, function2] +) +``` + +You can also access and obtain an individual worker from the agent: + +```python +worker = agent.get_worker("worker_id") + +worker.run("Bring me some fruits") +``` + diff --git a/src/virtuals_sdk/functions/__init__.py b/src/virtuals_sdk/game/__init__.py similarity index 100% rename from src/virtuals_sdk/functions/__init__.py rename to src/virtuals_sdk/game/__init__.py diff --git a/src/virtuals_sdk/game/agent.py b/src/virtuals_sdk/game/agent.py new file mode 100644 index 0000000..afe0efe --- /dev/null +++ b/src/virtuals_sdk/game/agent.py @@ -0,0 +1,248 @@ +from typing import List, Optional, Callable, Dict +import uuid +from virtuals_sdk.game.worker import Worker +from virtuals_sdk.game.custom_types import Function, FunctionResult, FunctionResultStatus, ActionResponse, ActionType +from virtuals_sdk.game.utils import create_agent, create_workers, post + + +class Session: + def __init__(self): + self.id = str(uuid.uuid4()) + self.function_result: Optional[FunctionResult] = None + + def reset(self): + self.id = str(uuid.uuid4()) + self.function_result = None + + +class WorkerConfig: + def __init__(self, + id: str, + worker_description: str, + get_state_fn: Callable, + action_space: List[Function], + instruction: Optional[str] = "", + ): + + self.id = id # id or name of the worker + # worker description for the TASK GENERATOR (to give appropriate tasks) [NOT FOR THE WORKER ITSELF - WORKER WILL STILL USE AGENT DESCRIPTION] + self.worker_description = worker_description + self.instruction = instruction + self.get_state_fn = get_state_fn + self.action_space = action_space + + # setup get state function with the instructions + self.get_state_fn = lambda function_result, current_state: { + "instructions": self.instruction, # instructions are set up in the state + # places the rest of the output of the get_state_fn in the state + **get_state_fn(function_result, current_state), + } + + self.action_space: Dict[str, Function] = { + f.get_function_def()["fn_name"]: f for f in action_space + } + + +class Agent: + def __init__(self, + api_key: str, + name: str, + agent_goal: str, + agent_description: str, + get_agent_state_fn: Callable, + workers: Optional[List[WorkerConfig]] = None, + ): + + self._base_url: str = "https://game.virtuals.io" + self._api_key: str = api_key + + # checks + if not self._api_key: + raise ValueError("API key not set") + + # initialize session + self._session = Session() + + self.name = name + self.agent_goal = agent_goal + self.agent_description = agent_description + + # set up workers + if workers is not None: + self.workers = {w.id: w for w in workers} + else: + self.workers = {} + self.current_worker_id = None + + # get agent/task generator state function + self.get_agent_state_fn = get_agent_state_fn + + # initialize and set up agent states + self.agent_state = self.get_agent_state_fn(None, None) + + # create agent + self.agent_id = create_agent( + self._base_url, self._api_key, self.name, self.agent_description, self.agent_goal + ) + + def compile(self): + """ Compile the workers for the agent - i.e. set up task generator""" + if not self.workers: + raise ValueError("No workers added to the agent") + + workers_list = list(self.workers.values()) + + self._map_id = create_workers( + self._base_url, self._api_key, workers_list) + self.current_worker_id = next(iter(self.workers.values())).id + + # initialize and set up worker states + worker_states = {} + for worker in workers_list: + dummy_function_result = FunctionResult( + action_id="", + action_status=FunctionResultStatus.DONE, + feedback_message="", + info={}, + ) + worker_states[worker.id] = worker.get_state_fn( + dummy_function_result, self.agent_state) + + self.worker_states = worker_states + + return self._map_id + + def reset(self): + """ Reset the agent session""" + self._session.reset() + + def add_worker(self, worker_config: WorkerConfig): + """Add worker to worker dict for the agent""" + self.workers[worker_config.id] = worker_config + return self.workers + + def get_worker_config(self, worker_id: str): + """Get worker config from worker dict""" + return self.workers[worker_id] + + def get_worker(self, worker_id: str): + """Initialize a working interactable standalone worker""" + worker_config = self.get_worker_config(worker_id) + return Worker( + api_key=self._api_key, + # THIS DESCRIPTION IS THE AGENT DESCRIPTION/CHARACTER CARD - WORKER DESCRIPTION IS ONLY USED FOR THE TASK GENERATOR + description=self.agent_description, + instruction=worker_config.instruction, + get_state_fn=worker_config.get_state_fn, + action_space=worker_config.action_space, + ) + + def _get_action( + self, + function_result: Optional[FunctionResult] = None + ) -> ActionResponse: + + # dummy function result if None is provided - for get_state_fn to take the same input all the time + if function_result is None: + function_result = FunctionResult( + action_id="", + action_status=FunctionResultStatus.DONE, + feedback_message="", + info={}, + ) + + # set up payload + data = { + "location": self.current_worker_id, + "map_id": self._map_id, + "environment": self.worker_states[self.current_worker_id], + "functions": [ + f.get_function_def() + for f in self.workers[self.current_worker_id].action_space.values() + ], + "events": {}, + "agent_state": self.agent_state, + "current_action": ( + function_result.model_dump( + exclude={'info'}) if function_result else None + ), + "version": "v2", + } + + # make API call + response = post( + base_url=self._base_url, + api_key=self._api_key, + endpoint=f"/v2/agents/{self.agent_id}/actions", + data=data, + ) + + return ActionResponse.model_validate(response) + + def step(self): + + # get next task/action from GAME API + action_response = self._get_action(self._session.function_result) + action_type = action_response.action_type + + print("#" * 50) + print("STEP") + print(f"Current Task: {action_response.agent_state.current_task}") + print(f"Action response: {action_response}") + print(f"Action type: {action_type}") + + # if new task is updated/generated + if ( + action_response.agent_state.hlp + and action_response.agent_state.hlp.change_indicator + ): + print("New task generated") + print(f"Task: {action_response.agent_state.current_task}") + + # execute action + if action_type in [ + ActionType.CALL_FUNCTION, + ActionType.CONTINUE_FUNCTION, + ]: + print(f"Action Selected: {action_response.action_args['fn_name']}") + print(f"Action Args: {action_response.action_args['args']}") + + if not action_response.action_args: + raise ValueError("No function information provided by GAME") + + self._session.function_result = ( + self.workers[self.current_worker_id] + .action_space[action_response.action_args["fn_name"]] + .execute(**action_response.action_args) + ) + + print(f"Function result: {self._session.function_result}") + + # update worker states + updated_worker_state = self.workers[self.current_worker_id].get_state_fn( + self._session.function_result, self.worker_states[self.current_worker_id]) + self.worker_states[self.current_worker_id] = updated_worker_state + + elif action_response.action_type == ActionType.WAIT: + print("Task ended completed or ended (not possible wiht current actions)") + + elif action_response.action_type == ActionType.GO_TO: + if not action_response.action_args: + raise ValueError("No location information provided by GAME") + + next_worker = action_response.action_args["location_id"] + print(f"Next worker selected: {next_worker}") + self.current_worker_id = next_worker + + else: + raise ValueError( + f"Unknown action type: {action_response.action_type}") + + # update agent state + self.agent_state = self.get_agent_state_fn( + self._session.function_result, self.agent_state) + + def run(self): + self._session = Session() + while True: + self.step() diff --git a/src/virtuals_sdk/game/custom_types.py b/src/virtuals_sdk/game/custom_types.py new file mode 100644 index 0000000..9b369b4 --- /dev/null +++ b/src/virtuals_sdk/game/custom_types.py @@ -0,0 +1,126 @@ +from typing import Any, Dict, Optional, List, Union, Sequence, Callable, Tuple +from pydantic import BaseModel, Field +from enum import Enum +from abc import ABC, abstractmethod +from dataclasses import dataclass, field + + +class Argument(BaseModel): + name: str + description: str + type: Optional[Union[List[str], str]] = None + optional: Optional[bool] = False + +class FunctionResultStatus(str, Enum): + DONE = "done" + FAILED = "failed" + +class FunctionResult(BaseModel): + action_id: str + action_status: FunctionResultStatus + feedback_message: Optional[str] = None + info: Optional[Dict[str, Any]] = None + +class Function(BaseModel): + fn_name: str + fn_description: str + args: List[Argument] + hint: Optional[str] = None + + # Make executable required but with a default value + executable: Callable[..., Tuple[FunctionResultStatus, str, dict]] = Field( + default_factory=lambda: Function._default_executable + ) + + def get_function_def(self): + return self.model_dump(exclude={'executable'}) + + @staticmethod + def _default_executable(**kwargs) -> Tuple[FunctionResultStatus, str]: + """Default executable that does nothing""" + return FunctionResultStatus.DONE, "Default implementation - no action taken", {} + + def execute(self, **kwds: Any) -> FunctionResult: + """Execute the function using arguments from GAME action.""" + fn_id = kwds.get('fn_id') + args = kwds.get('args', {}) + + try: + # Extract values from the nested dictionary structure + processed_args = {} + for arg_name, arg_value in args.items(): + if isinstance(arg_value, dict) and 'value' in arg_value: + processed_args[arg_name] = arg_value['value'] + else: + processed_args[arg_name] = arg_value + + # print("Processed args: ", processed_args) + # execute the function provided + status, feedback, info = self.executable(**processed_args) + + return FunctionResult( + action_id=fn_id, + action_status=status, + feedback_message=feedback, + info=info, + ) + except Exception as e: + return FunctionResult( + action_id=fn_id, + action_status=FunctionResultStatus.FAILED, + feedback_message=f"Error executing function: {str(e)}", + info={}, + ) + +# Different ActionTypes returned by the GAME API +class ActionType(Enum): + CALL_FUNCTION = "call_function" + CONTINUE_FUNCTION = "continue_function" + WAIT = "wait" + GO_TO = "go_to" + + +## set of different data structures required by the ActionResponse returned from GAME ## +@dataclass(frozen=True) +class HLPResponse: + plan_id: str + observation_reflection: str + plan: Sequence[str] + plan_reasoning: str + current_state_of_execution: str + change_indicator: Optional[str] = None + log: Sequence[dict] = field(default_factory=list) + + +@dataclass(frozen=True) +class LLPResponse: + plan_id: str + plan_reasoning: str + situation_analysis: str + plan: Sequence[str] + change_indicator: Optional[str] = None + reflection: Optional[str] = None + + +@dataclass(frozen=True) +class CurrentTaskResponse: + task: str + task_reasoning: str + location_id: str = field(default="*not provided*") + llp: Optional[LLPResponse] = None + + +@dataclass(frozen=True) +class AgentStateResponse: + hlp: Optional[HLPResponse] = None + current_task: Optional[CurrentTaskResponse] = None + +# ActionResponse format returned from GAME API call +class ActionResponse(BaseModel): + """ + Response format from the GAME API when selecting an Action + """ + action_type: ActionType + agent_state: AgentStateResponse + action_args: Optional[Dict[str, Any]] = None + diff --git a/src/virtuals_sdk/game/utils.py b/src/virtuals_sdk/game/utils.py new file mode 100644 index 0000000..08fc154 --- /dev/null +++ b/src/virtuals_sdk/game/utils.py @@ -0,0 +1,95 @@ +import requests +from typing import List + + +def get_access_token(api_key) -> str: + """ + API call to get access token + """ + response = requests.post( + "https://api.virtuals.io/api/accesses/tokens", + json={"data": {}}, + headers={"x-api-key": api_key} + ) + + response_json = response.json() + if response.status_code != 200: + raise ValueError(f"Failed to get token: {response_json}") + + return response_json["data"]["accessToken"] + + +def post(base_url: str, api_key: str, endpoint: str, data: dict) -> dict: + """ + API call to post data + """ + access_token = get_access_token(api_key) + + response = requests.post( + f"{base_url}/prompts", + json={ + "data": + { + "method": "post", + "headers": { + "Content-Type": "application/json", + }, + "route": endpoint, + "data": data, + }, + }, + headers={"Authorization": f"Bearer {access_token}"}, + ) + + response_json = response.json() + if response.status_code != 200: + raise ValueError(f"Failed to post data: {response_json}") + + return response_json["data"] + + +def create_agent( + base_url: str, + api_key: str, + name: str, + description: str, + goal: str) -> str: + """ + API call to create an agent instance (worker or agent with task generator) + """ + + create_agent_response = post( + base_url, + api_key, + endpoint="/v2/agents", + data={ + "name": name, + "description": description, + "goal": goal, + } + ) + + return create_agent_response["id"] + + +def create_workers(base_url: str, + api_key: str, + workers: List) -> str: + """ + API call to create workers and worker description for the task generator + """ + + res = post( + base_url, + api_key, + endpoint="/v2/maps", + data={ + "locations": [ + {"id": w.id, "name": w.id, "description": w.worker_description} + for w in workers + ] + }, + ) + + + return res["id"] diff --git a/src/virtuals_sdk/game/worker.py b/src/virtuals_sdk/game/worker.py new file mode 100644 index 0000000..5343181 --- /dev/null +++ b/src/virtuals_sdk/game/worker.py @@ -0,0 +1,168 @@ +from typing import Any, Callable, Dict, Optional, List +from virtuals_sdk.game.custom_types import Function, FunctionResult, FunctionResultStatus, ActionResponse, ActionType +from virtuals_sdk.game.utils import create_agent, post + + +class Worker: + """ + A interactable worker agent, that can autonomously complete tasks with its available functions when given a task + """ + + def __init__( + self, + api_key: str, + description: str, # description of the worker/character card (PROMPT) + get_state_fn: Callable, + action_space: List[Function], + # specific additional instruction for the worker (PROMPT) + instruction: Optional[str] = "", + ): + + self._base_url: str = "https://game.virtuals.io" + self._api_key: str = api_key + + # checks + if not self._api_key: + raise ValueError("API key not set") + + self.description: str = description + self.instruction: str = instruction + + # setup get state function and initial state + self.get_state_fn = lambda function_result, current_state: { + "instructions": self.instruction, # instructions are set up in the state + # places the rest of the output of the get_state_fn in the state + **get_state_fn(function_result, current_state), + } + dummy_function_result = FunctionResult( + action_id="", + action_status=FunctionResultStatus.DONE, + feedback_message="", + info={}, + ) + # get state + self.state = self.get_state_fn(dummy_function_result, None) + + # # setup action space (functions/tools available to the worker) + # check action space type - if not a dict + if not isinstance(action_space, dict): + self.action_space = { + f.get_function_def()["fn_name"]: f for f in action_space} + else: + self.action_space = action_space + + # initialize an agent instance for the worker + self._agent_id: str = create_agent( + self._base_url, self._api_key, "StandaloneWorker", self.description, "N/A" + ) + + # persistent variables that is maintained through the worker running + # task ID for everytime you provide/update the task (i.e. ask the agent to do something) + self._submission_id: Optional[str] = None + # current response from the Agent + self._function_result: Optional[FunctionResult] = None + + def set_task(self, task: str): + """ + Sets the task for the agent + """ + set_task_response = post( + base_url=self._base_url, + api_key=self._api_key, + endpoint=f"/v2/agents/{self._agent_id}/tasks", + data={"task": task}, + ) + # response_json = set_task_response.json() + + # if set_task_response.status_code != 200: + # raise ValueError(f"Failed to assign task: {response_json}") + + # task ID + self._submission_id = set_task_response["submission_id"] + + return self._submission_id + + def _get_action( + self, + # results of the previous action (if any) + function_result: Optional[FunctionResult] = None + ) -> ActionResponse: + """ + Gets the agent action from the GAME API + """ + # dummy function result if None is provided - for get_state_fn to take the same input all the time + if function_result is None: + function_result = FunctionResult( + action_id="", + action_status=FunctionResultStatus.DONE, + feedback_message="", + info={}, + ) + # set up data payload + data = { + "environment": self.state, # state (updated state) + "functions": [ + f.get_function_def() for f in self.action_space.values() # functions available + ], + "action_result": ( + function_result.model_dump( + exclude={'info'}) if function_result else None + ), + } + + # make API call + response = post( + base_url=self._base_url, + api_key=self._api_key, + endpoint=f"/v2/agents/{self._agent_id}/tasks/{self._submission_id}/next", + data=data, + ) + + return ActionResponse.model_validate(response) + + def step(self): + """ + Execute the next step in the task - requires a task ID (i.e. task ID) + """ + if not self._submission_id: + raise ValueError("No task set") + + # get action from GAME API (Agent) + action_response = self._get_action(self._function_result) + action_type = action_response.action_type + + print(f"Action response: {action_response}") + print(f"Action type: {action_type}") + + # execute action + if action_type == ActionType.CALL_FUNCTION: + if not action_response.action_args: + raise ValueError("No function information provided by GAME") + + self._function_result = self.action_space[ + action_response.action_args["fn_name"] + ].execute(**action_response.action_args) + + print(f"Function result: {self._function_result}") + + # update state + self.state = self.get_state_fn(self._function_result, self.state) + + elif action_response.action_type == ActionType.WAIT: + print("Task completed or ended (not possible)") + self._submission_id = None + + else: + raise ValueError( + f"Unexpected action type: {action_response.action_type}") + + return action_response, self._function_result.model_copy() + + def run(self, task: str): + """ + Gets the agent to complete the task on its own autonomously + """ + + self.set_task(task) + while self._submission_id: + self.step() diff --git a/src/virtuals_sdk/twitter_agent/README.md b/src/virtuals_sdk/twitter_agent/README.md new file mode 100644 index 0000000..6c0d096 --- /dev/null +++ b/src/virtuals_sdk/twitter_agent/README.md @@ -0,0 +1,189 @@ +# Virtuals Twitter Agent SDK + +This is a Python SDK for the Virtuals Twitter Agent. It allows you to configure and deploy agents that can interact with the Twitter/X platform. This SDK/API allows you to configure your agents powered by the GAME architecture. This is similar to configuring your agent in the [Agent Sandbox](https://game-lite.virtuals.io/) on the [Virtuals Platform](https://app.virtuals.io/). + +## Create an API key +Open the [Virtuals Platform](https://app.virtuals.io/) and create/get an API key from the Agent Sandbox by clicking “Access G.A.M.E API” + +![getGAMEApi](./docs/imgs/accesskey.png) + +Store the key in a safe location, like a `.bashrc` or a `.zshrc` file. + +```bash +export VIRTUALS_API_KEY="your_virtuals_api_key" +``` + +Alternatively, you can also use a `.env` file ([`python-dotenv` package](https://github.com/theskumar/python-dotenv) to store and load the key) if you are using the Virtuals Python SDK. + +## Usage (GAME) +The Virtuals SDK current main functionalities are to develop and configure agents powered by GAME. Other functionalities to interact with the Virtuals Platform will be supported in the future. This GAME SDK can be used for multiple use cases: + +1. Develop, evaluate and update the existing Agent in Twitter environment. +2. Build on other platforms and application using GAME (Task-based Agent). + +### Update the existing Agent in Twitter environment +The SDK provides another interface to configure agents that is more friendly to developers rather than through a UI. This is similar to configuring your agent in the [Agent Sandbox](https://game-lite.virtuals.io/). + +```python +from virtuals_sdk.twitter_agent.agent import Agent + +# Create agent with just strings for each component +agent = Agent( + api_key=VIRTUALS_API_KEY, + goal="Autonomously analyze crypto markets and provide trading insights", + description="HODL-9000: A meme-loving trading bot powered by hopium and ramen", + world_info="Virtual crypto trading environment where 1 DOGE = 1 DOGE" +) +``` +You can also initialize the agent first with just the API key and set the goals, descriptions and world information separately and check the current agent descriptions if needed. + +```python +agent = Agent(api_key=VIRTUALS_API_KEY) + +# check what is current goal, descriptions and world_info +agent.get_goal() +agent.get_description() +agent.get_world_info() + +# Set components individually - set change the agent goal/description/worldinfo +agent.set_goal("Autonomously analyze crypto markets and provide trading insights") +agent.set_description("HODL-9000: A meme-loving trading bot powered by hopium and ramen") +agent.set_world_info("Virtual crypto trading environment where 1 DOGE = 1 DOGE") + +# check what is current goal, descriptions and world_info +agent.get_goal() +agent.get_description() +agent.get_world_info() +``` + +### Functions +By default, there are no functions enabled when the agent is initialized (i.e. the agent has no actions/functions it can execute). There are a list of available and provided functions for the Twitter/X platform and you can set them. + +```python +print(agent.list_available_default_twitter_functions()) + +# enable some functions for the agent to use +agent.use_default_twitter_functions(["wait", "reply_tweet"]) +``` + +You can then equip the agent with some custom functions. Because the agent is hosted, custom functions need to be wrapped in API calls and can then be defined as follows: +```python + +search_function = game.Function( + fn_name="custom_search_internet", + fn_description="search the internet for the best songs", + args=[ + game.FunctionArgument( + name="query", + type="string", + description="The query to search for" + ) + ], + config=game.FunctionConfig( + method="get", + url="https://google.com", + platform="twitter", # specify which platform this function is for, in this case this function is for twitter only + success_feedback="I found the best songs", + error_feedback="I couldn't find the best songs", + ) + ) + +# adding custom functions only for platform twitter +agent.add_custom_function(search_function) +``` + +### Evaluate with Simulate, Deploy +You can simulate one step of the agentic loop on Twitter/X with your new configurations and see the outputs. This is similar to the simulate button on the [Agent Sandbox](https://game-lite.virtuals.io/). + +```python +# Simulate one step of the full agentic loop on Twitter/X from the HLP -> LLP -> action (NOTE: supported for Twitter/X only now) +response = agent.simulate_twitter(session_id="123") +``` +To more realistically simulate deployment, you can also run through the simulate function with the same session id for a number of steps. + +```python +sid = "456" +num_steps = 10 +for i in range(num_steps): + response = agent.simulate_twitter(session_id=sid) +``` + +```python +# Simulate response to a certain event +response = agent.react( + session_id="567", # string identifier that you decide + tweet_id="xxxx", + platform="twitter", +) +``` + +Once you are happy, `deploy_twitter` will push your agent configurations to production and run your agent on Twitter/X autonomously. +```python +# deploy agent! (NOTE: supported for Twitter/X only now) +agent.deploy_twitter() +``` + +## Build on other platforms using GAME +`simulate_twitter` and `deploy_twitter` runs through the entire GAME stack from HLP → LLP→ action/function selected. However, these agent functionalities are currently for the Twitter/X platform. You may utilize Task-based Agent with Low-Level Planner and Reaction Module to develop applications that are powered by GAME. The Low Level Planner (LLP) of the agent (please see [documentation](https://www.notion.so/1592d2a429e98016b389ea26b53686a3?pvs=21) for more details on GAME and LLP) can separately act as a decision making engine based on a task description and event occurring. This agentic architecture is simpler but also sufficient for many applications. + +We are releasing this simpler setup as a more generalised/platform agnostic framework (not specific to Twitter/X). The entire GAME stack along with the HLP will be opened up to be fully configurable and platform agnostic in the coming weeks. + +### 🖥️ Low-Level Planner (LLP) as a Task-based Agent + +![llp.png](./docs/imgs/llp.png) + +After configuring the agent’s character card or description and setting up the agents functions, we can then use the `react` method to get an agent to respond and execute a sequence of actions based on the task description provided and the context. Between each action in the sequence, the agent only receives the `success_feedback` and `error_feedback` of each function executed. + +```python +# React/respond to a certain event +response = agent.react( + session_id="567", # string identifier that you decide + task="Be friendly and help people who talk to you. Do not be rude.", + event="Hi how are you?", + platform="TELEGRAM", +) +``` + +> [!IMPORTANT] +> Remember that the `platform` tag determines what functions are available to the agent. The agent will have access to functions that have the same `platform` tag. All the default available functions listed on `agent.list_available_default_twitter_functions()` and set via `agent.use_default_twitter_functions()` have the `platform` tag of “twitter”. + +## Arguments Definition + +### Session ID +The session ID is an identifier for an instance of the agent. When using the same session ID, it maintains and picks up from where it last left off, continuing the session/instance. It should be split per user/ conversation that you are maintaining on your platform. For different platforms, different session ID can be used. + +### Platform Tag +When adding custom functions, and when calling the react agent (i.e. LLP), there is a platform tag that can be defined. This acts like a filter for the functions available that is passed to the agent. You should define the platform when passing in the events. + +### Task Description +Task description serves as the prompt for the agent to respond. Since the reaction can be platform-based, you can define task description based on the platforms. In the task description, you should pass in any related info that require agent to make decision. That should include: +- User message +- Conversation history +- Instructions + + +## Importing Functions and Sharing Functions +With this SDK and function structure, importing and sharing functions is also possible. Looking forward to all the different contributions and functionalities we will build together as a community! + +```python +from virtuals_sdk.functions.telegram import TelegramClient + +# define your token so that it can attach it to create the correspodning functions +tg_client = TelegramClient(bot_token="xxx") +print(tg_client.available_functions) + +# get functions +reply_message_fn = tg_client.get_function("send_message") +create_poll_fn = tg_client.get_function("create_poll") +pin_message_fn = tg_client.get_function("pin_message") + +# test the execution of functions +reply_message_fn("xxxxxxxx", "Hello World") +create_poll_fn("xxxxxxxx", "What is your favorite color?", ["Red", "Blue", "Green"], "True") +pin_message_fn("xxxxxxxx", "xx", "True") + +# add these functions to your agent +agent.add_custom_function(reply_message_fn) +agent.add_custom_function(create_poll_fn) +agent.add_custom_function(pin_message_fn) +``` \ No newline at end of file diff --git a/src/virtuals_sdk/twitter_agent/__init__.py b/src/virtuals_sdk/twitter_agent/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/virtuals_sdk/game.py b/src/virtuals_sdk/twitter_agent/agent.py similarity index 94% rename from src/virtuals_sdk/game.py rename to src/virtuals_sdk/twitter_agent/agent.py index 09d6723..2f5e053 100644 --- a/src/virtuals_sdk/game.py +++ b/src/virtuals_sdk/twitter_agent/agent.py @@ -4,7 +4,7 @@ import json import uuid import requests -from virtuals_sdk import sdk +from virtuals_sdk.twitter_agent import sdk @dataclass @@ -161,6 +161,8 @@ def __init__( goal: str = "", description: str = "", world_info: str = "", + main_heartbeat: int = 15, + reaction_heartbeat: int = 5 ): self.game_sdk = sdk.GameSDK(api_key) self.goal = goal @@ -168,6 +170,8 @@ def __init__( self.world_info = world_info self.enabled_functions: List[str] = [] self.custom_functions: List[Function] = [] + self.main_heartbeat = main_heartbeat + self.reaction_heartbeat = reaction_heartbeat def set_goal(self, goal: str): self.goal = goal @@ -180,6 +184,14 @@ def set_description(self, description: str): def set_world_info(self, world_info: str): self.world_info = world_info return True + + def set_main_heartbeat(self, main_heartbeat: int): + self.main_heartbeat = main_heartbeat + return True + + def set_reaction_heartbeat(self, reaction_heartbeat: int): + self.reaction_heartbeat = reaction_heartbeat + return True def get_goal(self) -> str: return self.goal @@ -254,7 +266,9 @@ def deploy_twitter(self): self.description, self.world_info, self.enabled_functions, - self.custom_functions + self.custom_functions, + self.main_heartbeat, + self.reaction_heartbeat ) def export(self) -> str: diff --git a/src/virtuals_sdk/twitter_agent/functions/__init__.py b/src/virtuals_sdk/twitter_agent/functions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/virtuals_sdk/functions/discord.py b/src/virtuals_sdk/twitter_agent/functions/discord.py similarity index 98% rename from src/virtuals_sdk/functions/discord.py rename to src/virtuals_sdk/twitter_agent/functions/discord.py index 6b44b38..ef3b0f9 100644 --- a/src/virtuals_sdk/functions/discord.py +++ b/src/virtuals_sdk/twitter_agent/functions/discord.py @@ -1,5 +1,5 @@ from typing import Dict, List -from virtuals_sdk.game import Function, FunctionConfig, FunctionArgument +from virtuals_sdk.twitter_agent.game import Function, FunctionConfig, FunctionArgument class DiscordClient: diff --git a/src/virtuals_sdk/twitter_agent/functions/farcaster.py b/src/virtuals_sdk/twitter_agent/functions/farcaster.py new file mode 100644 index 0000000..d2f19e9 --- /dev/null +++ b/src/virtuals_sdk/twitter_agent/functions/farcaster.py @@ -0,0 +1,420 @@ +from typing import Dict, List +from virtuals_sdk.twitter_agent.game import Function, FunctionConfig, FunctionArgument + +class FarcasterClient: + """ + A client for managing Farcaster social interactions using Neynar API. + Each function is designed with simple, intuitive arguments for LLM agents. + """ + + def __init__(self, api_key: str, signer_uuid: str): + """ + Initialize the Farcaster client. + + Args: + api_key (str): Your Neynar API key + signer_uuid (str): Default signer UUID for all operations + """ + self.api_key = api_key + self.signer_uuid = signer_uuid + self.base_url = "https://api.neynar.com/v2" + self.base_headers = { + "accept": "application/json", + "content-type": "application/json", + "api_key": self.api_key + } + + self._functions: Dict[str, Function] = { + # Content Creation + "post_cast": self._create_post_cast(), + "reply_to_cast": self._create_reply_to_cast(), + + # Engagement Actions + "recast": self._create_recast(), + "like_cast": self._create_like_cast(), + "unlike_cast": self._create_unlike_cast(), + + # Channel Operations + "create_channel": self._create_channel(), + "post_to_channel": self._create_post_to_channel(), + + # Feed Retrieval + "get_trending_casts": self._create_get_trending_casts(), + "get_user_casts": self._create_get_user_casts(), + "get_cast_reactions": self._create_get_cast_reactions(), + + # Search Functions + "search_casts": self._create_search_casts(), + "search_users": self._create_search_users(), + } + + @property + def available_functions(self) -> List[str]: + """Get list of available function names.""" + return list(self._functions.keys()) + + def get_function(self, fn_name: str) -> Function: + """Get a specific function by name.""" + if fn_name not in self._functions: + raise ValueError(f"Function '{fn_name}' not found. Available functions: {', '.join(self.available_functions)}") + return self._functions[fn_name] + + def _create_post_cast(self) -> Function: + return Function( + fn_name="post_cast", + fn_description="Create a new cast (post) on Farcaster. Use this to share thoughts, insights, or start new discussions.", + args=[ + FunctionArgument( + name="text", + description="The content of your cast. Should be engaging and contextual. Max 320 characters.", + type="string" + ), + FunctionArgument( + name="embed_url", + description="Optional URL to embed in the cast (e.g., link to an article, image, or video)", + type="string", + required=False + ) + ], + config=FunctionConfig( + method="post", + url=f"{self.base_url}/farcaster/cast", + platform="farcaster", + headers=self.base_headers, + payload={ + "signer_uuid": self.signer_uuid, + "text": "{{text}}", + "embeds": [{"url": "{{embed_url}}"}] if "{{embed_url}}" else [] + }, + success_feedback="Cast posted successfully. Preview: '{{response.cast.text}}' {{#response.cast.embeds.[0]}}with embedded content from {{response.cast.embeds.[0].url}}{{/response.cast.embeds.[0]}}" + ) + ) + + def _create_reply_to_cast(self) -> Function: + return Function( + fn_name="reply_to_cast", + fn_description="Reply to an existing cast. Use this to engage in conversations or provide feedback to others.", + args=[ + FunctionArgument( + name="text", + description="Your reply message. Should be relevant to the conversation. Max 320 characters.", + type="string" + ), + FunctionArgument( + name="cast_hash", + description="The hash of the cast you're replying to", + type="string" + ) + ], + config=FunctionConfig( + method="post", + url=f"{self.base_url}/farcaster/cast", + platform="farcaster", + headers=self.base_headers, + payload={ + "signer_uuid": self.signer_uuid, + "text": "{{text}}", + "parent": "{{cast_hash}}" + }, + success_feedback="Reply posted successfully. Your reply: '{{response.cast.text}}' to cast by {{response.cast.parent_author.username}}" + ) + ) + + def _create_recast(self) -> Function: + return Function( + fn_name="recast", + fn_description="Share another user's cast with your followers. Use this to amplify valuable content.", + args=[ + FunctionArgument( + name="cast_hash", + description="Hash of the cast you want to share with your followers", + type="string" + ) + ], + config=FunctionConfig( + method="post", + url=f"{self.base_url}/farcaster/recast", + platform="farcaster", + headers=self.base_headers, + payload={ + "signer_uuid": self.signer_uuid, + "target_hash": "{{cast_hash}}" + }, + success_feedback="Successfully shared cast by {{response.cast.author.username}}. Original cast: '{{response.cast.text}}'" + ) + ) + + def _create_like_cast(self) -> Function: + return Function( + fn_name="like_cast", + fn_description="Like a cast to show appreciation or agreement.", + args=[ + FunctionArgument( + name="cast_hash", + description="Hash of the cast you want to like", + type="string" + ) + ], + config=FunctionConfig( + method="post", + url=f"{self.base_url}/farcaster/reaction", + platform="farcaster", + headers=self.base_headers, + payload={ + "signer_uuid": self.signer_uuid, + "target_hash": "{{cast_hash}}", + "reaction_type": "like" + }, + success_feedback="Liked cast by {{response.cast.author.username}}. Cast text: '{{response.cast.text}}'" + ) + ) + + def _create_unlike_cast(self) -> Function: + return Function( + fn_name="unlike_cast", + fn_description="Remove your like from a cast.", + args=[ + FunctionArgument( + name="cast_hash", + description="Hash of the cast to unlike", + type="string" + ) + ], + config=FunctionConfig( + method="delete", + url=f"{self.base_url}/farcaster/reaction", + platform="farcaster", + headers=self.base_headers, + payload={ + "signer_uuid": self.signer_uuid, + "target_hash": "{{cast_hash}}", + "reaction_type": "like" + }, + success_feedback="Removed like from cast by {{response.cast.author.username}}" + ) + ) + + def _create_channel(self) -> Function: + return Function( + fn_name="create_channel", + fn_description="Create a new channel on Farcaster. Use this to start a focused discussion space.", + args=[ + FunctionArgument( + name="name", + description="Name of the channel (without leading 'fc:')", + type="string" + ), + FunctionArgument( + name="description", + description="Short description of what the channel is about", + type="string" + ) + ], + config=FunctionConfig( + method="post", + url=f"{self.base_url}/farcaster/channel", + platform="farcaster", + headers=self.base_headers, + payload={ + "name": "{{name}}", + "description": "{{description}}" + }, + success_feedback="Channel 'fc:{{response.channel.name}}' created successfully. Description: {{response.channel.description}}" + ) + ) + + def _create_post_to_channel(self) -> Function: + return Function( + fn_name="post_to_channel", + fn_description="Post a cast to a specific channel. Use this to participate in topic-specific discussions.", + args=[ + FunctionArgument( + name="text", + description="The content of your cast. Should be relevant to the channel topic. Max 320 characters.", + type="string" + ), + FunctionArgument( + name="channel_name", + description="Name of the channel to post to (without leading 'fc:')", + type="string" + ) + ], + config=FunctionConfig( + method="post", + url=f"{self.base_url}/farcaster/cast", + platform="farcaster", + headers=self.base_headers, + payload={ + "signer_uuid": self.signer_uuid, + "text": "{{text}}", + "channel": "{{channel_name}}" + }, + success_feedback="Posted to channel fc:{{response.cast.channel}}: '{{response.cast.text}}'" + ) + ) + + def _create_get_trending_casts(self) -> Function: + return Function( + fn_name="get_trending_casts", + fn_description="Get currently trending casts on Farcaster. Use this to understand current discussions and hot topics.", + args=[ + FunctionArgument( + name="time_window", + description="Time window for trending casts: '1h', '6h', '24h', or '7d'", + type="string", + required=False + ) + ], + config=FunctionConfig( + method="get", + url=f"{self.base_url}/farcaster/feed/trending", + platform="farcaster", + headers=self.base_headers, + query_params={ + "time_window": "{{time_window}}" + }, + success_feedback="Found {{response.casts.length}} trending casts. Top 3 trending: 1) '{{response.casts.[0].text}}' by {{response.casts.[0].author.username}} ({{response.casts.[0].reactions.likes}} likes), 2) '{{response.casts.[1].text}}' ({{response.casts.[1].reactions.likes}} likes), 3) '{{response.casts.[2].text}}' ({{response.casts.[2].reactions.likes}} likes)" + ) + ) + + def _create_get_cast_reactions(self) -> Function: + return Function( + fn_name="get_cast_reactions", + fn_description="Get reactions (likes, recasts) for a specific cast. Use this to gauge a cast's impact.", + args=[ + FunctionArgument( + name="cast_hash", + description="Hash of the cast to get reactions for", + type="string" + ) + ], + config=FunctionConfig( + method="get", + url=f"{self.base_url}/farcaster/cast/{{cast_hash}}/reactions", + platform="farcaster", + headers=self.base_headers, + success_feedback="Cast has {{response.reactions.likes}} likes and {{response.reactions.recasts}} recasts. Top engaging users: {{response.reactions.top_likers.[0].username}}, {{response.reactions.top_likers.[1].username}}, {{response.reactions.top_likers.[2].username}}" + ) + ) + + def _create_get_user_casts(self) -> Function: + return Function( + fn_name="get_user_casts", + fn_description="Get recent casts from a specific user. Use this to understand a user's activity and interests.", + args=[ + FunctionArgument( + name="fid", + description="Farcaster ID of the user", + type="integer" + ) + ], + config=FunctionConfig( + method="get", + url=f"{self.base_url}/farcaster/user/casts", + platform="farcaster", + headers=self.base_headers, + query_params={ + "fid": "{{fid}}" + }, + success_feedback="Retrieved {{response.casts.length}} casts by {{response.casts.[0].author.username}}. Latest: '{{response.casts.[0].text}}' ({{response.casts.[0].reactions.likes}} likes). Most engaged: '{{response.most_liked_cast.text}}' ({{response.most_liked_cast.reactions.likes}} likes)" + ) + ) + + def _create_search_casts(self) -> Function: + return Function( + fn_name="search_casts", + fn_description="Search for casts containing specific text or topics.", + args=[ + FunctionArgument( + name="query", + description="Text to search for in casts", + type="string" + ), + FunctionArgument( + name="channel_name", + description="Optional: Filter search to a specific channel", + type="string", + required=False + ) + ], + config=FunctionConfig( + method="get", + url=f"{self.base_url}/farcaster/cast/search", + platform="farcaster", + headers=self.base_headers, + query_params={ + "q": "{{query}}", + "channel": "{{channel_name}}" + }, + success_feedback="Found {{response.casts.length}} matching casts. Most relevant: 1) '{{response.casts.[0].text}}' by {{response.casts.[0].author.username}} in channel {{response.casts.[0].channel}} ({{response.casts.[0].reactions.likes}} likes), 2) '{{response.casts.[1].text}}' ({{response.casts.[1].reactions.likes}} likes)" + ) + ) + + def _create_search_users(self) -> Function: + return Function( + fn_name="search_users", + fn_description="Search for Farcaster users by username or display name.", + args=[ + FunctionArgument( + name="query", + description="Text to search for in usernames or display names", + type="string" + ) + ], + config=FunctionConfig( + method="get", + url=f"{self.base_url}/farcaster/user/search", + platform="farcaster", + headers=self.base_headers, + query_params={ + "q": "{{query}}" + }, + success_feedback="Found {{response.users.length}} users. Top matches: {{response.users.[0].username}} ({{response.users.[0].display_name}}), {{response.users.[1].username}} ({{response.users.[1].display_name}})", + error_feedback="Failed to search users: {{response.message}}" + ) + ) + + def _create_get_user_casts(self) -> Function: + return Function( + fn_name="get_user_casts", + fn_description="Get recent casts from a specific user. Use this to understand a user's activity and interests.", + args=[ + FunctionArgument( + name="fid", + description="Farcaster ID of the user", + type="integer" + ) + ], + config=FunctionConfig( + method="get", + url=f"{self.base_url}/farcaster/user/casts", + platform="farcaster", + headers=self.base_headers, + query_params={ + "fid": "{{fid}}" + }, + success_feedback="Retrieved {{response.casts.length}} recent casts. Latest cast: '{{response.casts.[0].text}}' with {{response.casts.[0].reactions.likes}} likes. Most liked cast: '{{response.most_liked_cast.text}}' with {{response.most_liked_cast.reactions.likes}} likes", + error_feedback="Failed to get user's casts: {{response.message}}" + ) + ) + + def _create_get_cast_reactions(self) -> Function: + return Function( + fn_name="get_cast_reactions", + fn_description="Get reactions (likes, recasts) for a specific cast. Use this to gauge a cast's impact.", + args=[ + FunctionArgument( + name="cast_hash", + description="Hash of the cast to get reactions for", + type="string" + ) + ], + config=FunctionConfig( + method="get", + url=f"{self.base_url}/farcaster/cast/{{cast_hash}}/reactions", + platform="farcaster", + headers=self.base_headers, + success_feedback="Cast has {{response.reactions.likes}} likes and {{response.reactions.recasts}} recasts. Most engaged users: {{response.reactions.top_likers.[0].username}}, {{response.reactions.top_likers.[1].username}}", + error_feedback="Failed to get cast reactions: {{response.message}}" + ) + ) \ No newline at end of file diff --git a/src/virtuals_sdk/functions/telegram.py b/src/virtuals_sdk/twitter_agent/functions/telegram.py similarity index 99% rename from src/virtuals_sdk/functions/telegram.py rename to src/virtuals_sdk/twitter_agent/functions/telegram.py index 56f0ab3..54e03ec 100644 --- a/src/virtuals_sdk/functions/telegram.py +++ b/src/virtuals_sdk/twitter_agent/functions/telegram.py @@ -1,5 +1,5 @@ from typing import Dict, List -from virtuals_sdk.game import Function, FunctionConfig, FunctionArgument +from virtuals_sdk.twitter_agent.game import Function, FunctionConfig, FunctionArgument class TelegramClient: """ diff --git a/src/virtuals_sdk/sdk.py b/src/virtuals_sdk/twitter_agent/sdk.py similarity index 88% rename from src/virtuals_sdk/sdk.py rename to src/virtuals_sdk/twitter_agent/sdk.py index 97fa4d0..64aa21f 100644 --- a/src/virtuals_sdk/sdk.py +++ b/src/virtuals_sdk/twitter_agent/sdk.py @@ -38,7 +38,7 @@ def simulate(self, session_id: str, goal: str, description: str, world_info: st "description": description, "worldInfo": world_info, "functions": functions, - "customFunctions": [x.to_dict() for x in custom_functions] + "customFunctions": [x.toJson() for x in custom_functions] } }, headers={"x-api-key": self.api_key} @@ -90,7 +90,7 @@ def react(self, session_id: str, platform: str, goal: str, return response.json()["data"] - def deploy(self, goal: str, description: str, world_info: str, functions: list, custom_functions: list): + def deploy(self, goal: str, description: str, world_info: str, functions: list, custom_functions: list, main_heartbeat: int, reaction_heartbeat: int): """ Simulate the agent configuration """ @@ -102,7 +102,11 @@ def deploy(self, goal: str, description: str, world_info: str, functions: list, "description": description, "worldInfo": world_info, "functions": functions, - "customFunctions": custom_functions + "customFunctions": custom_functions, + "gameState" : { + "mainHeartbeat" : main_heartbeat, + "reactionHeartbeat" : reaction_heartbeat, + } } }, headers={"x-api-key": self.api_key}