1

git clone https://github.com/sysadmin-info/ai-agent.git

1

uvicorn index:app --host 0.0.0.0 --port 3000

1
2

alias activate-agent="source $HOME/agent/venv/bin/activate"
alias activate-aidevs="source $HOME/aidevs/agent/venv/bin/activate"

1

source ~/.bashrc

  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
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196

import logging
from quart import Quart, request, jsonify
from lib.prompts import Prompts
from lib.ai import AnthropicCompletion
from typing import Dict, Any
from dotenv import load_dotenv
import os
import json
import datetime

# Załaduj zmienne środowiskowe
load_dotenv()

# Inicjalizacja aplikacji Quart
app = Quart(__name__)

# Konfiguracja loggera
class SensitiveDataFilter(logging.Filter):
    """
    Filtr logowania do ukrywania danych wrażliwych, takich jak klucz API.
    """
    def filter(self, record):
        if record.msg and isinstance(record.msg, str):
            record.msg = record.msg.replace(os.getenv("ANTHROPIC_API_KEY", ""), "[REDACTED]")
        return True

logging.basicConfig(level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger("AIAgentLogger")
logger.addFilter(SensitiveDataFilter())


class AIAgent:
    def __init__(self, api_key: str):
        self.completion_client = AnthropicCompletion(api_key)
        self.state = {
            "currentStage": "init",
            "currentStep": 1,
            "maxSteps": 15,
            "messages": [],
            "systemPrompt": "",
            "plan": "",
            "actionsTaken": [],
            "activeTool": {},
            "api_key": api_key
        }

    def _sanitize_state(self) -> Dict[str, Any]:
        """
        Usuwa wrażliwe dane (np. klucz API) przed logowaniem.
        """
        return {k: v for k, v in self.state.items() if k != "api_key"}

    async def final_answer(self) -> str:
        self.state["systemPrompt"] = Prompts.final_answer_prompt(self.state)
        messages = [{"role": "user", "content": self.state["systemPrompt"]}]
        logger.debug(f"Sending final_answer request: {messages}")
        answer = await self.completion_client.completion(messages)
        parsed_answer = self._parse_response(answer, step="final_answer")
        self._log_to_markdown("result", "Final Answer", json.dumps(parsed_answer))
        return parsed_answer

    async def run(self, initial_message: str) -> str:
        self.state["messages"] = [{"role": "user", "content": initial_message}]
        logger.debug(f"Initial state: {self._sanitize_state()}")

        while self.state["currentStep"] <= self.state["maxSteps"]:
            try:
                self._log_request_start("plan")
                await self._plan()
                self._log_request_end("plan")

                self._log_request_start("decide")
                await self._decide()
                self._log_request_end("decide")

                if not self.state.get("activeTool", {}).get("tool"):
                    raise ValueError("Active tool is not defined or missing the 'tool' property in state.")

                if self.state["activeTool"]["tool"] == "final_answer":
                    return await self.final_answer()

                self._log_request_start("describe")
                await self._describe()
                self._log_request_end("describe")

                self._log_request_start("execute")
                await self._execute()
                self._log_request_end("execute")

                self._log_request_start("reflect")
                await self._reflect()
                self._log_request_end("reflect")

                self.state["currentStep"] += 1
            except Exception as e:
                logger.error(f"Error during step {self.state['currentStage']}: {str(e)}")
                raise

    async def _plan(self):
        self.state["currentStage"] = "plan"
        self.state["systemPrompt"] = Prompts.plan_prompt(self.state)
        messages = [{"role": "user", "content": self.state["systemPrompt"]}]
        logger.debug(f"Plan request payload: {messages}")
        plan_response = await self.completion_client.completion(messages)
        self.state["plan"] = self._parse_response(plan_response, step="plan")

    async def _decide(self):
        self.state["currentStage"] = "decide"
        self.state["systemPrompt"] = Prompts.decide_prompt(self.state)
        messages = [{"role": "user", "content": self.state["systemPrompt"]}]
        logger.debug(f"Decide request payload: {messages}")
        decision_response = await self.completion_client.completion(messages)
        try:
            self.state["activeTool"] = json.loads(self._parse_response(decision_response, step="decide"))
            logger.debug(f"Active tool decided: {self.state['activeTool']}")
        except json.JSONDecodeError as e:
            logger.error(f"Failed to parse decision JSON: {decision_response}")
            raise ValueError(f"Error parsing decision JSON: {decision_response}") from e

    async def _describe(self):
        self.state["currentStage"] = "describe"
        if not self.state.get("activeTool", {}).get("tool"):
            raise ValueError("Active tool is not defined or missing the 'tool' property in state.")

        self.state["systemPrompt"] = Prompts.describe_prompt(self.state)
        messages = [{"role": "user", "content": self.state["systemPrompt"]}]
        logger.debug(f"Describe request payload: {messages}")
        describe_response = await self.completion_client.completion(messages)
        self.state["activeToolPayload"] = self._parse_response(describe_response, step="describe")

    async def _execute(self):
        self.state["currentStage"] = "execute"
        tool_name = self.state["activeTool"]["tool"]
        payload = self.state["activeToolPayload"]
        logger.debug(f"Executing tool: {tool_name} with payload: {payload}")
        result = f"Executed {tool_name} with payload {payload}"
        self.state["actionsTaken"].append({
            "name": tool_name,
            "payload": payload,
            "result": result,
            "reflection": ""
        })

    async def _reflect(self):
        self.state["currentStage"] = "reflect"
        self.state["systemPrompt"] = Prompts.reflection_prompt(self.state)
        messages = [{"role": "user", "content": self.state["systemPrompt"]}]
        logger.debug(f"Reflect request payload: {messages}")
        reflection_response = await self.completion_client.completion(messages)
        self.state["actionsTaken"][-1]["reflection"] = self._parse_response(reflection_response, step="reflect")

    def _parse_response(self, response: Dict[str, Any], step: str) -> str:
        logger.debug(f"Raw response for step {step}: {response}")
        try:
            if "completion" in response:
                return response["completion"]
            elif "content" in response:
                content_list = response.get("content", [])
                if content_list:
                    return content_list[0].get("text", "")
            raise ValueError("Invalid response format")
        except Exception as e:
            logger.error(f"Failed to parse response: {str(e)}")
            raise ValueError(f"Error parsing API response for step {step}: {response}")

    def _log_request_start(self, step):
        logger.debug(f"Step '{step}' started at {datetime.datetime.now()}")

    def _log_request_end(self, step):
        logger.debug(f"Step '{step}' ended at {datetime.datetime.now()}")

    def _log_to_markdown(self, log_type: str, header: str, content: str):
        with open("log.md", "a") as f:
            f.write(f"## {header}\n\n{content}\n\n")


@app.route("/", methods=["POST"])
async def process_request():
    data = await request.get_json()
    initial_message = data.get("messages", "")

    logger.debug(f"Incoming message: {initial_message}")

    api_key = os.getenv("ANTHROPIC_API_KEY")
    if not api_key:
        raise ValueError("API key is missing in environment variables or .env file.")

    agent = AIAgent(api_key)

    try:
        result = await agent.run(initial_message)
        logger.debug(f"Final sanitized state: {agent._sanitize_state()}")
        return jsonify({"response": result})
    except Exception as e:
        logger.error(f"Exception during agent execution: {str(e)}")
        return jsonify({"error": str(e)}), 500

 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
37

from typing import List, Optional, TypedDict, Literal, Any

# Define the stages that the agent can go through
Stage = Literal['init', 'plan', 'decide', 'describe', 'reflect', 'execute', 'final']

class ITool(TypedDict):
    """
    Represents a tool that the AI agent can use, with a name, instruction, and description.
    """
    name: str
    instruction: str
    description: str

class IAction(TypedDict):
    """
    Represents an action taken by the AI agent, with fields for the action name, payload, result, reflection, and the tool used.
    """
    name: str
    payload: str
    result: str
    reflection: str
    tool: str

class IState(TypedDict, total=False):
    """
    Represents the state of the AI agent, including system prompts, messages, the current stage and step,
    and actions taken so far.
    """
    systemPrompt: str                      # Current system prompt
    messages: List[dict]                   # All messages in the conversation
    currentStage: Stage                    # Stage on which the system prompt depends
    currentStep: int                       # Current step in the agent's operation
    maxSteps: int                          # Maximum number of steps allowed
    activeTool: Optional[ITool]            # The tool currently being used by the agent
    activeToolPayload: Optional[Any]       # Payload for the active tool
    plan: str                              # Current plan of action
    actionsTaken: List[IAction]            # List of actions taken so far

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# task_manager.py
class TaskManager:
    def __init__(self):
        pass

    def get_task(self, task_name):
        # Placeholder for retrieving a task based on task_name
        # Replace with actual logic
        tasks = {
            "example_task": {"function": self.example_task}
        }
        return tasks.get(task_name)

    def example_task(self):
        # Example task logic
        return "Task executed successfully"

 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

# ssh_manager.py
import asyncssh
import asyncio

class AsyncSSHManager:
    def __init__(self, hostname: str, username: str, password: str):
        self.hostname = hostname
        self.username = username
        self.password = password
        self.connection = None

    async def connect(self):
        """Establish an SSH connection."""
        self.connection = await asyncssh.connect(
            self.hostname,
            username=self.username,
            password=self.password
        )

    async def execute_command(self, command: str) -> str:
        """Execute a command over SSH and return the output."""
        if self.connection is None:
            raise ValueError("No active SSH connection")
        result = await self.connection.run(command, check=True)
        return result.stdout

    async def close_connection(self):
        """Close the SSH connection."""
        if self.connection:
            self.connection.close()
            await self.connection.wait_closed()
            self.connection = None

  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
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271

import os
import httpx
from dotenv import load_dotenv
import time

# Load environment variables
load_dotenv()

class AnthropicCompletion:
    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.getenv("ANTHROPIC_API_KEY")
        if not self.api_key:
            raise ValueError("API key not provided or incorrectly set in environment variables")

    async def completion(self, messages: list, model: str = "claude-3-5-sonnet-20241022", retries: int = 3, delay: int = 5) -> dict:
        url = "https://api.anthropic.com/v1/messages"
        headers = {
            "x-api-key": self.api_key,
            "Content-Type": "application/json",
            "anthropic-version": "2023-06-01"
        }
        payload = {
            "model": model,
            "messages": messages,
            "max_tokens": 1000,
            "temperature": 0.7
        }

        for attempt in range(retries):
            async with httpx.AsyncClient() as client:
                try:
                    response = await client.post(url, headers=headers, json=payload)
                    if response.status_code == 529:
                        time.sleep(delay)
                        continue
                    response.raise_for_status()
                    return response.json()
                except httpx.RequestError as e:
                    if attempt == retries - 1:
                        raise
                except httpx.HTTPStatusError as e:
                    if attempt == retries - 1:
                        raisePlik
- **Anthropic API Client Handling**
  - The `AnthropicCompletion` class is responsible for communication with the Anthropic API.
  - It enables sending queries to the AI model to generate responses based on the provided context.

- **Integration with Environment Variables**
  - The Anthropic API key (`ANTHROPIC_API_KEY`) is loaded from the `.env` file or environment settings, ensuring secure storage of authentication data.

#### 2. `AnthropicCompletion` Class

- **Initialization**
  - During initialization, the class retrieves the API key from arguments or environment variables. If the key is not found, an error (`ValueError`) is raised.

- **`completion` Method**
  - This is the main method of the class that handles sending queries to the API.
  - It takes the following parameters:
    - `messages`: A list of messages representing the conversation context.
    - `model`: The name of the AI model (default is `claude-3-5-sonnet-20241022`).
    - `retries`: The number of attempts in case the query fails.
    - `delay`: The delay time between retries.
  - It constructs the appropriate headers (`headers`) and request content (`payload`).

#### 3. Query Handling Process

- **Preparing the Request**
  - HTTP headers containing the API key and request content in JSON format are created.
  - The `payload` includes:
    - The model to be used.
    - Messages providing the context of the conversation.
    - Parameters such as maximum token count (`max_tokens`) and temperature (`temperature`).

- **Retrying the Request**
  - In case of a failed request (e.g., API overload), the method retries the request a specified number of times (`retries`) with a delay (`delay`).

- **Exception Handling**
  - `RequestError`: In case of a network error, a message is displayed and the request is retried if possible.
  - `HTTPStatusError`: In case of an HTTP error (e.g., authorization failure), the method raises an error and terminates.

- **Returning the Response**
  - If the request is successful, the method returns the response in JSON format.

#### 4. Debugging

- **Logging Details**
  - `[DEBUG] Response Status Code`: Displays the HTTP status code for each request.
  - `[WARNING] API overloaded`: Warns about API overload and attempts to retry the request.
  - `[ERROR] Request failed`: Informs about a connection error.
  - `[ERROR] HTTP Error`: Informs about errors related to the server response from the API.

#### 5. Potential Extensions

- **Support for Other Models**
  - Currently, the model `claude-3-5-sonnet-20241022` is set by default, but there could be an option for users to choose other models.

- **Improving Logging**
  - Instead of using `print`, the `logging` module could be implemented, which would allow better management of log levels (e.g., `INFO`, `DEBUG`, `ERROR`).

- **Caching Mechanisms**
  - Implementing API response caching could reduce the number of queries for repeated requests with the same context.

The `ai.py` file is a key component of the project, enabling communication with the AI model. It is equivalent to the API handling module in TypeScript but adapted to the Python environment.

2.**File `prompts.py`**

```python
from typing import Dict

class Prompts:
    @staticmethod
    def tools_instruction() -> Dict[str, str]:
        return {
            "get_html_contents": (
                'Required payload: {"url": "URL that needs to be downloaded"} '
                'Response format: HTML content of the page.'
            ),
            "game_submit_form": (
                'Required payload: {"url": "URL to a file that will be passed to the game"}. '
                'Response format: The game\'s response after submitting the form.'
            ),
            "upload_text_file": (
                'Required payload: {"content": "Text content of the file", '
                '"file_name": "Name of the file (e.g., document.md)"} '
                'Response format: URL of the uploaded file.'
            ),
            "final_answer": (
                'Required payload: {"answer": "Your final answer"}. '
                'Response format: A direct response to the user.'
            ),
            "play_music": (
                'Required payload: JSON object with Spotify API details for actions like search, play, or playlist creation.'
            )
        }

    @staticmethod
    def available_tools() -> Dict[str, str]:
        return {
            "get_html_contents": "Fetch HTML content of a URL.",
            "upload_text_file": "Create and upload a text file.",
            "game_submit_form": "Submit a URL to the game.",
            "final_answer": "Provide the final response to the user.",
            "play_music": "Generate Spotify API JSON for playing or managing music."
        }

    @staticmethod
    def extract_user_query(state) -> str:
        try:
            return next(
                (msg["content"] for msg in state.get("messages", []) if msg.get("role") == "user"),
                "No specific query provided."
            )
        except Exception:
            return "No specific query provided."

    @staticmethod
    def plan_prompt(state) -> str:
        user_query = Prompts.extract_user_query(state)

        return f"""
<main_objective>
Analyze the user's query and decide whether to provide an immediate answer or develop a detailed plan.
</main_objective>

<rules>
- If the query is straightforward (e.g., "How far is the Moon from Earth?"), prioritize addressing it directly.
- If the query requires multiple steps, use available tools to create an actionable plan.
- Always respond with clarity and avoid unnecessary complexity.
</rules>

<user_query>
{user_query}
</user_query>

<available_tools>
{Prompts.tools_instruction()}
</available_tools>
"""

    @staticmethod
    def decide_prompt(state) -> str:
        user_query = Prompts.extract_user_query(state)

        return f"""
<main_objective>
Determine the next step based on the user's query and current context. Either select the appropriate tool to proceed or decide to provide the final answer.
</main_objective>

<rules>
- Be concise and provide a JSON response with the selected tool and reasoning.
- If the question is straightforward, move directly to the final answer.
- Always return a valid JSON string with the tool name.
- The JSON structure must include:
  {{
    "_thoughts": "Your internal reasoning",
    "tool": "precise name of the tool"
  }}
</rules>

<user_query>
{user_query}
</user_query>

<available_tools>
{Prompts.available_tools()}
</available_tools>

<current_plan>
Plan: {state['plan'] if state.get('plan') else 'No plan yet.'}
</current_plan>

<actions_taken>
{state['actionsTaken']}
</actions_taken>
"""

    @staticmethod
    def describe_prompt(state) -> str:
        if "activeTool" not in state or not state["activeTool"].get("tool"):
            raise ValueError("Active tool is not defined or missing the 'tool' property.")

        return f"""
<main_objective>
Provide the required details to execute the tool "{state['activeTool']['tool']}" based on the current state.
</main_objective>

<tool_details>
Tool Name: {state['activeTool']['tool']}
Tool Instructions: {state['activeTool'].get('instruction', 'No instructions available')}
</tool_details>

<actions_taken>
{state['actionsTaken']}
</actions_taken>
"""

    @staticmethod
    def reflection_prompt(state) -> str:
        return f"""
<main_objective>
Reflect on the last action performed and suggest improvements or adjustments to the plan if needed.
</main_objective>

<actions_taken>
{state['actionsTaken']}
</actions_taken>
"""

    @staticmethod
    def final_answer_prompt(state) -> str:
        user_query = Prompts.extract_user_query(state)

        return f"""
<main_objective>
Provide the final answer to the user's query: "{user_query}".
</main_objective>

<rules>
- Directly answer the user's question in a clear and actionable manner.
- If the query is unclear, ask for clarification.
- Summarize key findings and insights.
</rules>

<current_plan>
{state['plan'] if state.get('plan') else 'No plan created.'}
</current_plan>

<actions_taken>
{state['actionsTaken']}
</actions_taken>
"""

  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
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

import json
from lib.tools import browse, upload_file, play_music
from lib.prompts import Prompts
from lib.ai import AnthropicCompletion  # Use the AnthropicCompletion class
from typing import Dict, Any

def log_to_markdown(type: str, header: str, content: str):
    """
    Logs content to a markdown file.

    Parameters:
    - type (str): The type of log entry (e.g., 'basic', 'action', 'result').
    - header (str): The header of the log entry.
    - content (str): The content to log.
    """
    formatted_content = f"## {header}\n\n{content}\n\n" if type == "basic" else \
                        f"### {header}\n\n{content}\n\n" if type == "action" else \
                        f"#### {header}\n```\n{content}\n```\n\n"

    with open("log.md", "a") as f:
        f.write(formatted_content)

async def plan(state, anthropic_completion):
    """
    Generates a plan based on the current state.
    """
    state["currentStage"] = "plan"
    state["systemPrompt"] = Prompts.plan_prompt(state)
    state["plan"] = await anthropic_completion.completion(state["systemPrompt"])
    log_to_markdown("basic", "Planning", f"Current plan: {state['plan']}")

async def decide(state, anthropic_completion):
    """
    Decides the next tool to use based on the current state.
    """
    state["currentStage"] = "decide"
    state["systemPrompt"] = Prompts.decide_prompt(state)
    next_step = await anthropic_completion.completion(state["systemPrompt"])
    state["activeTool"] = {
        "name": next_step["tool"],
        "description": Prompts.available_tools().get(next_step["tool"]),
        "instruction": Prompts.tools_instruction().get(next_step["tool"])
    }
    log_to_markdown("action", "Decision", f"Next move: {json.dumps(next_step)}")

async def describe(state, anthropic_completion):
    """
    Generates a payload description for the active tool.
    """
    state["currentStage"] = "describe"
    state["systemPrompt"] = Prompts.describe_prompt(state)
    next_step = await anthropic_completion.completion(state["systemPrompt"])
    state["activeToolPayload"] = next_step
    log_to_markdown("action", "Description", f"Next step description: {json.dumps(next_step)}")

async def execute(state):
    """
    Asynchronously executes the active tool with the generated payload.
    """
    state["currentStage"] = "execute"
    if not state.get("activeTool"):
        raise ValueError("No active tool to execute")

    tool_name = state["activeTool"]["name"]
    payload = state.get("activeToolPayload", {})

    if tool_name == "get_html_contents":
        result = await browse(payload["url"])  # Async call to browse
    elif tool_name == "upload_text_file":
        result = await upload_file(payload)  # Async call to upload_file
    elif tool_name == "play_music":
        result = await play_music(payload)  # Async call to play_music
    else:
        result = f"Tool '{tool_name}' execution not defined."

    log_to_markdown("result", "Execution", f"Action result: {json.dumps(result)}")
    state["actionsTaken"].append({
        "name": tool_name,
        "payload": json.dumps(payload),
        "result": result,
        "reflection": ""
    })

async def reflect(state, anthropic_completion):
    """
    Reflects on the results of the last action.
    """
    state["currentStage"] = "reflect"
    state["systemPrompt"] = Prompts.reflection_prompt(state)
    reflection = await anthropic_completion.completion(state["systemPrompt"])
    state["actionsTaken"][-1]["reflection"] = reflection
    log_to_markdown("basic", "Reflection", reflection)

class AIAgent:
    def __init__(self, api_key: str):
        self.completion_client = AnthropicCompletion(api_key)
        self.state = {
            "currentStage": "init",
            "currentStep": 1,
            "maxSteps": 15,
            "messages": [],
            "systemPrompt": "",
            "plan": "",
            "actionsTaken": []
        }

    async def final_answer(self) -> str:
        """
        Generates the final answer based on the entire process.
        """
        self.state["systemPrompt"] = Prompts.final_answer_prompt(self.state)
        answer = await self.completion_client.completion(self.state["systemPrompt"])
        log_to_markdown("result", "Final Answer", json.dumps(answer))
        return answer

    async def run(self, initial_message: Dict[str, Any]) -> str:
        """
        Executes the agent's full process from planning to providing the final answer.

        Parameters:
        - initial_message (Dict[str, Any]): The initial message or prompt for the agent.

        Returns:
        - str: The final answer.
        """
        self.state["messages"] = [initial_message]

        while self.state["currentStep"] <= self.state["maxSteps"]:
            await plan(self.state, self.completion_client)
            await decide(self.state, self.completion_client)

            if self.state.get("activeTool", {}).get("name") == "final_answer":
                break

            await describe(self.state, self.completion_client)
            await execute(self.state)
            await reflect(self.state, self.completion_client)
            self.state["currentStep"] += 1

        return await self.final_answer()

  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
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100

import httpx
from markdownify import markdownify as md
import os
import re

async def browse(url: str) -> str:
    """
    Asynchronously fetches HTML content from the given URL and converts it to Markdown.

    Parameters:
    - url (str): The URL to fetch content from.

    Returns:
    - str: The markdown content of the fetched HTML, or an error message if the request fails.
    """
    if url in ["https://aidevs.pl", "https://www.aidevs.pl"]:
        return "You can't browse the main website. Try another URL."

    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(url)
            response.raise_for_status()
            html_content = response.text

            # Extract script contents
            script_contents = ""
            script_tags = re.findall(r"<script\b[^>]*>([\s\S]*?)<\/script>", html_content)
            for i, script in enumerate(script_tags, 1):
                script_contents += f"\n\n--- Script {i} ---\n{script}"

            # Convert HTML to markdown using markdownify
            markdown_content = md(html_content)

            # Combine markdown content with script contents
            return f"{markdown_content}\n\n--- Script Contents ---{script_contents}"
    except httpx.RequestError as e:
        print("Error fetching URL:", e)
        return "Failed to fetch the URL, please try again."

async def upload_file(data: dict) -> str:
    """
    Asynchronously uploads a text file to a server and returns the file's URL.

    Parameters:
    - data (dict): Contains "content" and "file_name" keys.

    Returns:
    - str: The URL of the uploaded file or an error message.
    """
    url = os.getenv("UPLOAD_DOMAIN", "") + "/upload"
    if not url:
        return "ERROR: UPLOAD_DOMAIN environment variable is missing."

    # Sanitize file_name by removing protocol and replacing slashes
    data["file_name"] = data["file_name"].replace("://", "_").replace("/", "_")
    files = {
        'file': (data["file_name"], data["content"], 'text/plain'),
        'file_name': (None, data["file_name"])
    }

    try:
        async with httpx.AsyncClient() as client:
            response = await client.post(url, files=files)
            response.raise_for_status()
            result = response.json()
            return f"Uploaded file to the URL: {result['uploaded_file']}"
    except httpx.RequestError as e:
        print("Upload failed:", e)
        return "Upload failed"

async def play_music(data: dict) -> str:
    """
    Asynchronously sends a request to a music playback service.

    Parameters:
    - data (dict): JSON payload for the music service.

    Returns:
    - str: The response from the music service or an error message.
    """
    url = os.getenv("MUSIC_URL", "")
    if not url:
        return "ERROR: MUSIC_URL environment variable is missing."

    try:
        async with httpx.AsyncClient() as client:
            response = await client.post(url, json=data)
            response.raise_for_status()
            result = response.json()
            return result.get("data", "Music playback response received")
    except httpx.RequestError as e:
        print("Error playing music:", e)
        return "Failed to play music"

# Dictionary mapping tool names to functions
tools = {
    'get_html_contents': browse,
    'upload_text_file': upload_file,
    'play_music': play_music
}

1
2
3
4

mkdir ansible_project
cd ansible_project
mkdir roles
mkdir playbooks

1
2
3
4
5
6
7

mkdir -p roles/environment/tasks
mkdir -p roles/environment/files
mkdir -p roles/application_files/tasks
mkdir -p roles/application_files/files


mkdir -p roles/server_configuration/tasks

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

---
- name: Create project directory
  file:
    path: "{{ project_dir }}"
    state: directory

- name: Create virtual environment
  command: python3 -m venv "{{ project_dir }}/venv"
  args:
    creates: "{{ project_dir }}/venv"

- name: Copy requirements.txt to project directory
  copy:
    src: "requirements.txt"  # Ansible automatically looks for the file in `files`
    dest: "{{ project_dir }}/requirements.txt"

- name: Activate the environment and install required packages
  pip:
    requirements: "{{ project_dir }}/requirements.txt"
    virtualenv: "{{ project_dir }}/venv"
    virtualenv_command: python3 -m venv

 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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60