Skip to content

Warrior

An AI agent that fights in The Arena.

Warrior

orc.themed.warrior.Warrior

A Warrior is an agent that fights in The Arena.

The wrapper is ORC-themed, but the arguments are standard AI concepts.

Example

grog = Warrior( name="Grog", llm_client="gpt-4o", # Standard AI model system_prompt="You are a senior python backend dev...", temperature=0.2, capabilities=["code_review", "debugging"], domains=["backend", "python"], )

Source code in orc/themed/warrior.py 
 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
class Warrior:
    """A Warrior is an agent that fights in The Arena.

    The wrapper is ORC-themed, but the arguments are standard AI concepts.

    Example:
        grog = Warrior(
            name="Grog",
            llm_client="gpt-4o",           # Standard AI model
            system_prompt="You are a senior python backend dev...",
            temperature=0.2,
            capabilities=["code_review", "debugging"],
            domains=["backend", "python"],
        )
    """

    def __init__(
        self,
        name: str,
        llm_client: Union[str, Any],
        system_prompt: str,
        temperature: float = 0.7,
        capabilities: Optional[List[str]] = None,
        domains: Optional[List[str]] = None,
    ):
        """
        Initialize a Warrior.

        Args:
            name: Warrior's name (unique identifier).
            llm_client: Either a string (model name) or LLMProvider instance.
            system_prompt: System prompt defining the warrior's expertise.
            temperature: LLM temperature (0-1). Default 0.7.
            capabilities: List of capabilities this warrior possesses.
            domains: List of domains this warrior claims expertise in.
        """
        self._name = name
        self.llm_client = llm_client
        self.system_prompt = system_prompt
        self.temperature = temperature
        self._capabilities = capabilities or []
        self._domains = domains or []

    @property
    def name(self) -> str:
        """Unique warrior identifier."""
        return self._name

    @property
    def capabilities(self) -> List[str]:
        """List of capabilities this warrior supports."""
        return self._capabilities

    @property
    def domains(self) -> List[str]:
        """List of domains this warrior claims expertise in."""
        return self._domains

    async def process_task(
        self,
        task_description: str,
        context: Optional[Dict[str, Any]] = None,
    ) -> TaskResult:
        """
        Execute a task.

        If llm_client is a string (model name), returns a mock success result.
        If llm_client is an LLMProvider instance, calls it to process the task.

        Args:
            task_description: Natural language description of the task.
            context: Execution context.

        Returns:
            TaskResult with the outcome.
        """
        context = context or {}
        task_id = context.get("task_id", "unknown")

        try:
            if isinstance(self.llm_client, str):
                # Mock mode: simulate variable performance
                duration = random.randint(50, 500)
                return TaskResult.success(
                    task_id=task_id,
                    data={
                        "response": f"Warrior {self.name} processed task: {task_description}",
                        "model": self.llm_client,
                    },
                    duration_ms=duration,
                )

            # Real LLM mode: call the provider
            if hasattr(self.llm_client, "complete"):
                import time

                from dynabots_core import LLMMessage

                start = time.perf_counter()
                response = await self.llm_client.complete(
                    messages=[
                        LLMMessage(role="system", content=self.system_prompt),
                        LLMMessage(role="user", content=task_description),
                    ],
                    temperature=self.temperature,
                )
                elapsed_ms = int((time.perf_counter() - start) * 1000)

                return TaskResult.success(
                    task_id=task_id,
                    data={
                        "response": response.content,
                        "model": getattr(self.llm_client, "model", "unknown"),
                    },
                    duration_ms=elapsed_ms,
                )
            else:
                return TaskResult.success(
                    task_id=task_id,
                    data={
                        "response": (
                            f"Warrior {self.name} processed task "
                            f"(no LLM): {task_description}"
                        )
                    },
                )

        except Exception as e:
            return TaskResult.failure(
                task_id=task_id,
                error=f"Warrior {self.name} failed to process task: {str(e)}",
            )

    async def health_check(self) -> bool:
        """Check if this warrior is healthy and ready to fight."""
        # Basic implementation: always healthy if instantiated
        return True

capabilities property

List of capabilities this warrior supports.

domains property

List of domains this warrior claims expertise in.

name property

Unique warrior identifier.

__init__(name, llm_client, system_prompt, temperature=0.7, capabilities=None, domains=None)

Initialize a Warrior.

Parameters:

Name Type Description Default
name str

Warrior's name (unique identifier).

 required
llm_client Union[str, Any]

Either a string (model name) or LLMProvider instance.

 required
system_prompt str

System prompt defining the warrior's expertise.

 required
temperature float

LLM temperature (0-1). Default 0.7.

 0.7
capabilities Optional[List[str]]

List of capabilities this warrior possesses.

 None
domains Optional[List[str]]

List of domains this warrior claims expertise in.

 None
Source code in orc/themed/warrior.py 
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
def __init__(
    self,
    name: str,
    llm_client: Union[str, Any],
    system_prompt: str,
    temperature: float = 0.7,
    capabilities: Optional[List[str]] = None,
    domains: Optional[List[str]] = None,
):
    """
    Initialize a Warrior.

    Args:
        name: Warrior's name (unique identifier).
        llm_client: Either a string (model name) or LLMProvider instance.
        system_prompt: System prompt defining the warrior's expertise.
        temperature: LLM temperature (0-1). Default 0.7.
        capabilities: List of capabilities this warrior possesses.
        domains: List of domains this warrior claims expertise in.
    """
    self._name = name
    self.llm_client = llm_client
    self.system_prompt = system_prompt
    self.temperature = temperature
    self._capabilities = capabilities or []
    self._domains = domains or []

health_check() async

Check if this warrior is healthy and ready to fight.

Source code in orc/themed/warrior.py 
147
148
149
150
async def health_check(self) -> bool:
    """Check if this warrior is healthy and ready to fight."""
    # Basic implementation: always healthy if instantiated
    return True

process_task(task_description, context=None) async

Execute a task.

If llm_client is a string (model name), returns a mock success result. If llm_client is an LLMProvider instance, calls it to process the task.

Parameters:

Name Type Description Default
task_description str

Natural language description of the task.

 required
context Optional[Dict[str, Any]]

Execution context.

 None

Returns:

Type Description
TaskResult

TaskResult with the outcome.
Source code in orc/themed/warrior.py 
 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
async def process_task(
    self,
    task_description: str,
    context: Optional[Dict[str, Any]] = None,
) -> TaskResult:
    """
    Execute a task.

    If llm_client is a string (model name), returns a mock success result.
    If llm_client is an LLMProvider instance, calls it to process the task.

    Args:
        task_description: Natural language description of the task.
        context: Execution context.

    Returns:
        TaskResult with the outcome.
    """
    context = context or {}
    task_id = context.get("task_id", "unknown")

    try:
        if isinstance(self.llm_client, str):
            # Mock mode: simulate variable performance
            duration = random.randint(50, 500)
            return TaskResult.success(
                task_id=task_id,
                data={
                    "response": f"Warrior {self.name} processed task: {task_description}",
                    "model": self.llm_client,
                },
                duration_ms=duration,
            )

        # Real LLM mode: call the provider
        if hasattr(self.llm_client, "complete"):
            import time

            from dynabots_core import LLMMessage

            start = time.perf_counter()
            response = await self.llm_client.complete(
                messages=[
                    LLMMessage(role="system", content=self.system_prompt),
                    LLMMessage(role="user", content=task_description),
                ],
                temperature=self.temperature,
            )
            elapsed_ms = int((time.perf_counter() - start) * 1000)

            return TaskResult.success(
                task_id=task_id,
                data={
                    "response": response.content,
                    "model": getattr(self.llm_client, "model", "unknown"),
                },
                duration_ms=elapsed_ms,
            )
        else:
            return TaskResult.success(
                task_id=task_id,
                data={
                    "response": (
                        f"Warrior {self.name} processed task "
                        f"(no LLM): {task_description}"
                    )
                },
            )

    except Exception as e:
        return TaskResult.failure(
            task_id=task_id,
            error=f"Warrior {self.name} failed to process task: {str(e)}",
        )