Skip to content

azad.prompts.dialects.native Package

azad.prompts.dialects.native

Anthropic Native dialect for Azad prompts.

This package contains the Anthropic Native dialect implementation for Azad prompts, which allows using Anthropic's native tool calling format in LLM interactions.

Classes

NativeDialectConfig

Bases: DialectConfig

Configuration for Native dialect parser.

Attributes
pretty_print class-attribute instance-attribute 
pretty_print: bool = Field(default=True, description='Whether to format JSON with indentation')
indentation class-attribute instance-attribute 
indentation: int = Field(default=2, description='Indentation level for pretty printing')
parameter_buffer_threshold class-attribute instance-attribute 
parameter_buffer_threshold: int = Field(default=15, description='Minimum characters for parameter chunks')
flush_interval_ms class-attribute instance-attribute 
flush_interval_ms: int = Field(default=100, description='Milliseconds between forced flushes')

NativeDialectParser 

NativeDialectParser(schema: Dict, prompt_data: PromptData, config: NativeDialectConfig)

Bases: DialectParser

Native dialect parser for processing streaming tool call deltas.

This parser processes tool call deltas directly from the model and efficiently emits events with proper buffering to reduce event frequency while maintaining responsiveness. It implements both size-based and time-based buffering mechanisms to optimize the trade-off between low latency and reducing excessive event emissions.

Initialize the parser with the given schema and configuration.

Source code in azad/prompts/dialects/native/parser.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(self, schema: Dict, prompt_data: PromptData, config: NativeDialectConfig):
    """Initialize the parser with the given schema and configuration."""
    self.schema = schema
    self.prompt_data = prompt_data
    self.config = config
    self.logger = logging.getLogger(__name__)

    # Set default buffering thresholds if not specified in config
    self.PARAMETER_BUFFER_THRESHOLD = config.parameter_buffer_threshold
    self.FLUSH_INTERVAL_MS = config.flush_interval_ms

    # Initialize parser state
    self.reset_state()
Attributes
schema instance-attribute 
schema = schema
prompt_data instance-attribute 
prompt_data = prompt_data
config instance-attribute 
config = config
logger instance-attribute 
logger = getLogger(__name__)
PARAMETER_BUFFER_THRESHOLD instance-attribute 
PARAMETER_BUFFER_THRESHOLD = parameter_buffer_threshold
FLUSH_INTERVAL_MS instance-attribute 
FLUSH_INTERVAL_MS = flush_interval_ms
Functions
reset_state 
reset_state()

Reset the parser state for a new tool call.

Source code in azad/prompts/dialects/native/parser.py 
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
def reset_state(self):
    """Reset the parser state for a new tool call."""
    # Tool call tracking
    self.tool_call_id = None
    self.current_tool_name = None
    self.in_tool_call = False
    self.tool_ready_emitted = False

    # Track multiple tool calls
    self.detected_tool_calls = []

    # Text state
    self.sent_text_start = False

    # Parameter tracking
    self.params_seen = set()               # Parameters we've seen start events for
    self.params_ended = set()              # Parameters that have ended
    self.param_values = {}                 # Current complete value for each parameter
    self.parameter_chunk_buffer = {}       # Buffer for accumulating parameter chunks
    self.last_update_length = {}           # Track last update length for each param
    self.last_flush_time = time.time() * 1000  # Last time we flushed chunks (in ms)

    # JSON parsing state
    self.raw_buffer = ""                   # Raw JSON accumulation
    self.is_complete = False               # Whether the tool call is complete
    self.first_key_value_extracted = False # Whether we've seen the first key-value pair
feed 
feed(data: bytes) -> List[AINetworkEventUnion]

Process plain text content (outside of a tool call).

This method is called when the model outputs regular text instead of a tool call.

Source code in azad/prompts/dialects/native/parser.py 
 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
def feed(self, data: bytes) -> List[AINetworkEventUnion]:
    """
    Process plain text content (outside of a tool call).

    This method is called when the model outputs regular text
    instead of a tool call.
    """
    events: List[AINetworkEventUnion] = []
    text = data.decode('utf-8')

    # Skip empty text
    if not text:
        return events

    # Ignore text when in a tool call (should use feed_tool_call_delta instead)
    if self.in_tool_call:
        return events

    # Start text section if needed
    if not self.sent_text_start:
        events.append(AINetworkEventTextStart())
        self.sent_text_start = True

    # Emit text chunk
    events.append(AINetworkEventTextChunk(content=text))
    return events
feed_tool_call_delta 
feed_tool_call_delta(tool_call: Union[ChatCompletionDeltaToolCall, ChatCompletionMessageToolCall]) -> List[AINetworkEventUnion]

Process tool call deltas from the model with efficient buffering.

This method handles the incremental parts of a tool call that come from the model, buffering small chunks for improved efficiency.

Source code in azad/prompts/dialects/native/parser.py 
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
def feed_tool_call_delta(
    self,
    tool_call: Union[ChatCompletionDeltaToolCall, ChatCompletionMessageToolCall]
) -> List[AINetworkEventUnion]:
    """
    Process tool call deltas from the model with efficient buffering.

    This method handles the incremental parts of a tool call that come
    from the model, buffering small chunks for improved efficiency.
    """
    events = []

    print(f"Received tool call delta: {tool_call}")
    # Skip if tool_call is invalid
    if not tool_call or not hasattr(tool_call, 'function'):
        return events

    # End text mode if we're transitioning to a tool call
    if self.sent_text_start and not self.in_tool_call:
        events.append(AINetworkEventTextEnd())
        self.sent_text_start = False

    function = tool_call.function

    # Start a new tool call if we have a name and ID
    if function.name is not None and tool_call.id is not None and not self.in_tool_call:
        self.tool_call_id = tool_call.id
        self.current_tool_name = function.name
        self.in_tool_call = True

        # Reset timer for new tool call
        self.last_flush_time = time.time() * 1000

        # Emit tool name event
        events.append(AINetworkEventToolName(
            tool_name=function.name,
            tool_call_id=tool_call.id
        ))

    # Process function arguments if provided
    if function.arguments is not None:
        new_content = function.arguments

        # Add to our raw JSON buffer
        self.raw_buffer += new_content

        # Process the updated JSON buffer
        new_events = self._process_json_buffer()
        events.extend(new_events)

        # Do any time-based flushing for all parameters
        current_time = time.time() * 1000
        if current_time - self.last_flush_time >= self.FLUSH_INTERVAL_MS:
            # Force flush any buffers that have content
            for param_name in list(self.parameter_chunk_buffer.keys()):
                if param_name in self.params_seen and param_name not in self.params_ended and self.parameter_chunk_buffer[param_name]:
                    events.append(AINetworkEventParameterChunk(
                        parameter=param_name,
                        content=self.parameter_chunk_buffer[param_name],
                        tool_call_id=self.tool_call_id # type: ignore
                    ))
                    # Update last update length
                    if param_name in self.param_values:
                        self.last_update_length[param_name] = len(self.param_values[param_name])
                    # Clear buffer
                    self.parameter_chunk_buffer[param_name] = ""

            # Update flush time
            self.last_flush_time = current_time

        # Check for completion
        if self._is_json_complete() and not self.is_complete:
            self.is_complete = True

            # End all parameters that haven't ended yet
            for param in self.params_seen:
                if param not in self.params_ended:
                    # Emit any remaining buffer content
                    if param in self.parameter_chunk_buffer and self.parameter_chunk_buffer[param]:
                        events.append(AINetworkEventParameterChunk(
                            parameter=param,
                            content=self.parameter_chunk_buffer[param],
                            tool_call_id=self.tool_call_id # type: ignore

                        ))
                        self.parameter_chunk_buffer[param] = ""

                    # End the parameter
                    events.append(AINetworkEventParameterEnd(parameter=param))
                    self.params_ended.add(param)
            # Emit parameters complete if we have a tool call ID - we should have
            if self.tool_call_id:
                events.append(AINetworkEventParametersComplete(tool_call_id=self.tool_call_id))
            else:
                self.logger.warning("No tool call ID found when emitting parameters complete")

            # Store the completed tool call for later reference
            if self.current_tool_name and self.tool_call_id:
                # Create the final parameters dict
                final_params = {}
                for param_name, param_value in self.param_values.items():
                    final_params[param_name] = param_value

                # Add to detected tool calls list
                self.detected_tool_calls.append({
                    "tool_name": self.current_tool_name,
                    "tool_call_id": self.tool_call_id,
                    "args": final_params
                })

                # Log the tool call detection
                self.logger.debug(f"Detected tool call: {self.current_tool_name} with ID: {self.tool_call_id}")

            # Emit tool ready if not already emitted
            if not self.tool_ready_emitted:
                self.tool_ready_emitted = True

                # Try to parse the final parameters
                final_params = {}
                for param_name, param_value in self.param_values.items():
                    final_params[param_name] = param_value

                # Emit tool ready event
                events.append(AINetworkEventToolReady(
                    tool_name=self.current_tool_name, # type: ignore
                    tool_call_id=self.tool_call_id, # type: ignore
                    args=final_params
                ))

    # Force-flush any remaining buffered content at the end of processing
    if self.in_tool_call:
        for param_name in list(self.parameter_chunk_buffer.keys()):
            if param_name in self.params_seen and param_name not in self.params_ended and self.parameter_chunk_buffer[param_name]:
                events.append(AINetworkEventParameterChunk(
                    parameter=param_name,
                    content=self.parameter_chunk_buffer[param_name],
                    tool_call_id=self.tool_call_id # type: ignore

                ))
                # Update the last update length
                if param_name in self.param_values:
                    self.last_update_length[param_name] = len(self.param_values[param_name])
                # Clear the buffer
                self.parameter_chunk_buffer[param_name] = ""

    return events
get_detected_tool_calls 
get_detected_tool_calls() -> List[Dict[str, Any]]

Return all detected tool calls.

Source code in azad/prompts/dialects/native/parser.py 
432
433
434
def get_detected_tool_calls(self) -> List[Dict[str, Any]]:
    """Return all detected tool calls."""
    return self.detected_tool_calls
is_multi_tool_response 
is_multi_tool_response() -> bool

Check if this was a multiple tool call response.

Source code in azad/prompts/dialects/native/parser.py 
436
437
438
def is_multi_tool_response(self) -> bool:
    """Check if this was a multiple tool call response."""
    return len(self.detected_tool_calls) > 1

NativeDialect 

NativeDialect(config: NativeDialectConfig)

Bases: Dialect

Native dialect for AI agent communication using native tool calling format.

Source code in azad/prompts/dialects/native/dialect.py 
27
28
29
def __init__(self, config: NativeDialectConfig):
    self.config = config
    self.logger = logging.getLogger(__name__)
Attributes
config_cls class-attribute instance-attribute 
config_cls = NativeDialectConfig
config instance-attribute 
config = config
logger instance-attribute 
logger = getLogger(__name__)
Functions
format_tool_schema 
format_tool_schema(tool: ToolMetadata) -> dict

Format a tool's metadata into a JSON schema format.

Parameters:

Returns:

  • dict –

    A dictionary representing the formatted tool schema

Source code in azad/prompts/base_dialect.py 
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
def format_tool_schema(self, tool: ToolMetadata) -> dict:
    """Format a tool's metadata into a JSON schema format.

    Args:
        tool: The tool metadata to format

    Returns:
        A dictionary representing the formatted tool schema
    """
    properties = {}
    for param_name, param_info in tool.parameters.items():
        # Extract parameter information
        if isinstance(param_info, ParameterMetadata):
            description = param_info.description
            # Determine the parameter type - we don't have explicit type info in ParameterMetadata
            # so we'll default to string
            param_type = "string"
        else:
            description = str(param_info)
            param_type = "string"

        properties[param_name] = {
            "type": param_type,
            "description": description
        }

    return {
        "name": tool.name,
        "description": tool.description,
        "input_schema": {
            "type": "object",
            "properties": properties,
            "required": tool.required_parameters,
        }
    }
format_tools_schema 
format_tools_schema(tools: List[ToolMetadata]) -> List[dict]

Format multiple tools' metadata into JSON schema format.

Parameters:

  • tools (List[ToolMetadata]) –

    A list of tool metadata objects to format

Returns:

  • List[dict] –

    A list of dictionaries representing the formatted tool schemas

Source code in azad/prompts/base_dialect.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
def format_tools_schema(self, tools: List[ToolMetadata]) -> List[dict]:
    """Format multiple tools' metadata into JSON schema format.

    Args:
        tools: A list of tool metadata objects to format

    Returns:
        A list of dictionaries representing the formatted tool schemas
    """
    return [self.format_tool_schema(tool) for tool in tools]
inject_prompt_cache 
inject_prompt_cache(messages: list[dict], prompt_data: PromptData)

Modifies the formatted messages list IN-PLACE to add cache control flags based on the model type and specific caching rules.

Source code in azad/prompts/base_dialect.py 
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
def inject_prompt_cache(self, messages: list[dict], prompt_data: PromptData):
    """
    Modifies the formatted messages list IN-PLACE to add cache control flags
    based on the model type and specific caching rules.
    """
    # Check if caching is explicitly enabled via dynamic config first
    cache_dialect_config = getattr(prompt_data.dyanmic_task_config, 'cache_dialect', None)

    # --- Anthropic Caching Logic ---
    # Apply if cache_dialect_config is ANTHROPIC OR if it's None and model is Anthropic
    is_anthropic_model = "anthropic" in prompt_data.task_config.model_name # Or use a more specific check if needed
    should_apply_anthropic = is_anthropic_model and (cache_dialect_config == CacheDialect.ANTHROPIC or cache_dialect_config is None)

    if should_apply_anthropic:
        # This uses the original logic provided in the snippet, which applies
        # cache control to system, second-last user, and last user messages.
        print("Applying Anthropic caching rules (System, Last User, Second-Last User)...")

        last_system_index = -1
        last_user_index = -1
        second_last_user_index = -1

        for i, msg in enumerate(messages):
            if msg.get('role') == 'system': last_system_index = i
            elif msg.get('role') == 'user': second_last_user_index = last_user_index; last_user_index = i

        def apply_anthropic_cache(index, is_last_user=False):
            if index != -1 and index < len(messages):
                msg = messages[index]
                content = msg.get('content')
                if isinstance(content, str): msg['content'] = [{"type": "text", "text": content}]; content = msg['content']
                if not isinstance(content, list): print(f"    Warning: Anthropic - content not list: {type(content)}"); return
                if not content: print(f"    Warning: Anthropic - content list empty."); return

                last_part = content[-1]
                if isinstance(last_part, dict): last_part['cache_control'] = {'type': 'ephemeral'}
                else: print(f"    Warning: Anthropic - last part not dict: {type(last_part)}")

                if is_last_user and len(content) >= 2:
                     second_last_part = content[-2]
                     if isinstance(second_last_part, dict): second_last_part['cache_control'] = {'type': 'ephemeral'}
                     else: print(f"    Warning: Anthropic - second-last part not dict: {type(second_last_part)}")

        apply_anthropic_cache(last_system_index)
        apply_anthropic_cache(second_last_user_index)
        apply_anthropic_cache(last_user_index, is_last_user=True)
        return

    # --- Gemini Caching Logic ---
    is_gemini_model = "gemini" in prompt_data.task_config.model_name
    should_apply_gemini = is_gemini_model and (cache_dialect_config == CacheDialect.GEMINI or cache_dialect_config is None)

    if should_apply_gemini and prompt_data.dyanmic_task_config.enable_explicit_caching:
        print("Checking Gemini caching rules...")

        # Calculate total characters as a proxy for tokens
        total_chars = 0
        for msg in messages:
            content = msg.get('content')
            if isinstance(content, list):
                for part in content:
                    if isinstance(part, dict) and part.get('type') == 'text':
                        total_chars += len(part.get('text', ''))
            elif isinstance(content, str):
                total_chars += len(content)

        # Minimum character count approximation (4096 tokens * ~4 chars/token)
        min_chars_for_cache = 16000 # Adjusted slightly lower
        print(f"  Total characters calculated: {total_chars}")

        if total_chars < min_chars_for_cache:
            print(f"  Skipping Gemini caching: Character count ({total_chars}) is below threshold ({min_chars_for_cache}).")
            return # Not enough content to warrant caching

        # check if includes at least one assistant message
        has_assistant_message = any(msg.get('role') == 'assistant' for msg in messages)

        if not has_assistant_message:
            print("  Skipping Gemini caching: No assistant message found.")
            return

        print("Applying Anthropic caching rules for Gemini model (System, Last User, Second-Last User)...")

        last_system_index = -1
        last_user_index = -1
        second_last_user_index = -1

        for i, msg in enumerate(messages):
            if msg.get('role') == 'system': last_system_index = i
            elif msg.get('role') == 'user': second_last_user_index = last_user_index; last_user_index = i

        def apply_gemini_cache(index, is_last_user=False):
            if index != -1 and index < len(messages):
                msg = messages[index]
                content = msg.get('content')
                if isinstance(content, str): msg['content'] = [{"type": "text", "text": content}]; content = msg['content']
                if not isinstance(content, list): print(f"    Warning: Gemini - content not list: {type(content)}"); return
                if not content: print(f"    Warning: Gemini - content list empty."); return

                last_part = content[-1]
                if isinstance(last_part, dict): last_part['cache_control'] = {'type': 'ephemeral'}
                else: print(f"    Warning: Gemini - last part not dict: {type(last_part)}")

                if is_last_user and len(content) >= 2:
                     second_last_part = content[-2]
                     if isinstance(second_last_part, dict): second_last_part['cache_control'] = {'type': 'ephemeral'}
                     else: print(f"    Warning: Gemini - second-last part not dict: {type(second_last_part)}")

        apply_gemini_cache(last_system_index)
        apply_gemini_cache(second_last_user_index)
        apply_gemini_cache(last_user_index, is_last_user=True)

        return

    # No explicit return needed
    return
format_messages 
format_messages(prompt_data: PromptData, rules_path: Optional[str]) -> list[dict]
Source code in azad/prompts/base_dialect.py 
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
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
def format_messages(self, prompt_data:PromptData, rules_path: Optional[str]) -> list[dict]:
    system_prompt = self.format_system_prompt(prompt_data, rules_path)
    history = self.format_history(prompt_data)

    # delete any messages with empty content or empty tool_calls
    history = [msg for msg in history if msg.get('content') and len(msg['content']) > 0
               or (msg.get('tool_calls') and len(msg['tool_calls']) > 0)]


    formated = [system_prompt, *history]
    formated = self._interleave_consecutive_messages(formated)

    # Check if it's a kodu provider
    is_kodu_provider = "kodu" in prompt_data.task_config.model_name

    if is_kodu_provider:
        from ..env_settings import settings
        from ..ainetwork.errors import AIInsufficientCreditsError, AIUserNotFoundError

        if settings.FLY_PROCESS_GROUP is not None and settings.DATABASE_URL is not None and settings.TURSO_AUTH_TOKEN is not None:
            user_api_key = prompt_data.dyanmic_task_config.model_api_key
            user_credits = 0
            credit_threshold = settings.CREDIT_THRESHOLD
            try:
                from ..db_models import db

                user_credits = db.get_user_credits(user_api_key)
            except Exception as e:
                print(f"Error in Kodu provider check: {str(e)}")

            if user_credits is not None:
                if user_credits < credit_threshold:
                    raise AIInsufficientCreditsError()
            else:
                raise AIUserNotFoundError()


    self.inject_prompt_cache(formated, prompt_data)


    # if we have dynamic environment details, add them to the last user message if it exists
    if prompt_data.dyanmic_task_config.dynamic_environment_details_block:
        last_user_index = -1
        for i, msg in enumerate(formated):
            if msg.get('role') == 'user':
                last_user_index = i
        if last_user_index != -1:
            # validate that the content field is array
            if isinstance(formated[last_user_index]['content'], list):
                formated[last_user_index]['content'].append({
                    "type":"text",
                    "text":prompt_data.dyanmic_task_config.dynamic_environment_details_block
                })
            else:
                # edge case this should not happen
                formated[last_user_index]['content'] = [
                    {
                        "type":"text",
                        "text":formated[last_user_index]['content']
                    },
                    {
                        "type":"text",
                        "text":prompt_data.dyanmic_task_config.dynamic_environment_details_block
                    }
                ]

    return formated
format_system_prompt 
format_system_prompt(prompt_data: PromptData, rules_path: Optional[str]) -> dict

Format the system prompt for the dialect. This method handles the replacement of placeholders in the system prompt with the actual values from the prompt data. If there is no placeholder it will format the system prompt in following way: TOOL_USE_INSTRUCTION AVAILABLE_TOOLS USER AGENT PROMPT SYSTEM GENERAL INSTRUCTIONS USER CUSTOM INSTRUCTIONS

Source code in azad/prompts/base_dialect.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
def format_system_prompt(self, prompt_data:PromptData, rules_path: Optional[str]) -> dict:
    """Format the system prompt for the dialect.
    This method handles the replacement of placeholders in the system prompt
    with the actual values from the prompt data.
    If there is no placeholder it will format the system prompt in following way:
    TOOL_USE_INSTRUCTION
    AVAILABLE_TOOLS
    USER AGENT PROMPT
    SYSTEM GENERAL INSTRUCTIONS
    USER CUSTOM INSTRUCTIONS
    """
    system = f"""{self.format_user_agent_prompt(prompt_data)}"""
    # check if {{AVAILABLE_TOOLS}} is in the system prompt
    def is_key_in_system_prompt(key: str) -> bool:
        return key in system
    def format_key(key: str) -> str:
        return f"{{{key}}}"
    system_key = format_key(PromptTemplateType.SYSTEM_INSTRUCTION)
    available_tools_key = format_key(PromptTemplateType.AVAILABLE_TOOLS)
    tool_use_instruction_key = format_key(PromptTemplateType.TOOL_USE_INSTRUCTION)
    user_prompt_key = format_key(PromptTemplateType.USER_PROMPT)
    if is_key_in_system_prompt(system_key):
        system = system.replace(system_key, self.format_system_rules(prompt_data))
    else:
        system = f"{system}\n{self.format_system_rules(prompt_data)}"
    if is_key_in_system_prompt(available_tools_key):
        system = system.replace(available_tools_key, self.format_tool_docs(prompt_data))
    else:
        system = f"{self.format_tool_docs(prompt_data)}\n{system}"
    if is_key_in_system_prompt(tool_use_instruction_key):
        system = system.replace(tool_use_instruction_key, self.format_dialect_rules(prompt_data, rules_path))
    else:
        system = f"{self.format_dialect_rules(prompt_data, rules_path)}\n{system}"
    if is_key_in_system_prompt(user_prompt_key):
        system = system.replace(user_prompt_key, self.format_user_prompt(prompt_data))
    else:
        system = f"{system}\n{self.format_user_prompt(prompt_data)}"
    return dict(content=[{"type":"text","text": system}],role="system")
format_system_rules 
format_system_rules(prompt_data)
Source code in azad/prompts/base_dialect.py 
353
354
355
356
357
358
359
    def format_system_rules(self, prompt_data):
        return """# You have special messages called Informational Messages that are generated by the local environment and are not part of the user's input. These messages may be visible or invisible to the user, you should observe the informational messages and take them into account when responding to the user in accordance to the task.
# Pay extra attention to your mistakes and try to self improve by learning from them and acting better in the future.
# You should always try to be direct and clear generating the best possible output for the user task.
# When commuinicating with the user about informational messages, you should always be clear and understand that informationals messages are generated by the local environment and the user may not be aware of them.
# This means when talking about any details from the informational messages, you should say the environment instead of the user, so when talking about the informational content always say "the environment" instead of "the user".
"""
format_history 
format_history(prompt_data: PromptData) -> List[dict]

Format the entire message history from the mindmap.

Source code in azad/prompts/base_dialect.py 
391
392
393
394
def format_history(self, prompt_data: PromptData) -> List[dict]:
    """Format the entire message history from the mindmap."""
    messages = prompt_data.messages
    return [msg for msg in (self.format_history_item(item) for item in messages) if msg is not None]
format_user_prompt 
format_user_prompt(prompt_data: PromptData) -> str
Source code in azad/prompts/base_dialect.py 
396
397
398
399
400
401
402
403
404
405
def format_user_prompt(self, prompt_data:PromptData) -> str:
    if not prompt_data.task_config.user_prompt:
        return ""

    return inspect.cleandoc(f"""====
        The user has provided additional instructions or details for you to use, Please understand the the user may or may not have knowledge of the overall system instructions, and this is their attempt to configure your behavior to match their needs.
        Here is the user custom instructions:
        {prompt_data.task_config.user_prompt}
        ====
        """)
format_user_agent_prompt 
format_user_agent_prompt(prompt_data: PromptData) -> str
Source code in azad/prompts/base_dialect.py 
407
408
409
410
411
def format_user_agent_prompt(self, prompt_data:PromptData) -> str:
    if not prompt_data.task_config.user_agent_prompt:
        return ""
    # return inspect.cleandoc(f"""<agent_prompt>{prompt_data.task_config.user_agent_prompt}</agent_prompt>""")
    return prompt_data.task_config.user_agent_prompt
format_tool_docs 
format_tool_docs(prompt_data: PromptData) -> str

Format documentation for available tools.

This base implementation handles the structure, while specific dialects format the individual examples.

Parameters:

  • tools –

    List of Tool classes or ToolSignature objects to document

Returns:

  • str –

    Formatted tool documentation string

Source code in azad/prompts/base_dialect.py 
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
    def format_tool_docs(self, prompt_data:PromptData) -> str:
        """Format documentation for available tools.

        This base implementation handles the structure, while specific dialects
        format the individual examples.

        Args:
            tools: List of Tool classes or ToolSignature objects to document

        Returns:
            Formatted tool documentation string
        """

        tools: list[ToolMetadata] = prompt_data.tool_metadata
        docs = []
        for tool in tools:
            # Handle both Tool and ToolSignature objects
            params = []
            for name, param_info in tool.parameters.items():
                required = "(required)" if name in tool.required_parameters else "(optional)"
                params.append(f"- {name}: {required} {self._get_parameter_description(param_info)}") # type: ignore

            examples = []
            for i, example in enumerate(tool.examples, 1):
                formatted_example = self.format_example(tool.name, example.parameters) # type: ignore
                examples.append(f"""<tool_call_example tool_name="{tool.name}">### {example.explanation}\n> Azad Output :\n{formatted_example}\n</<tool_call_example>""")

            doc = f"""<tool_doc tool_name="{tool.name}">
Tool name: {tool.name}
Tool Description: {tool.description}
Tool Input Parameters:
{"".join(params)}
Tool Usage examples:
<tool_call_examples tool_name="{tool.name}">
Here is a list of examples of how to use the tool, please read them carefully to understand how to use the tool effectively, if no instructions are provided, please use the tool as you see fit.
{"".join(examples)}
</tool_call_examples>
"""
            docs.append(doc)

        return f"""## Tool Documentation
Here are all the tools available for you to use in the task, please read the documentation carefully to understand how to use them effectively.
{"\n\n".join(docs).strip()}
"""
format_dialect_rules 
format_dialect_rules(prompt_data: PromptData) -> str

Return the native dialect rules.

Source code in azad/prompts/dialects/native/dialect.py 
31
32
33
34
35
def format_dialect_rules(self, prompt_data: PromptData) -> str:
    """Return the native dialect rules."""
    with open(os.path.join(os.path.dirname(__file__), 'native.rules.prompt')) as f:
        rules = f.read()
    return rules
dict_to_human_readable_simple 
dict_to_human_readable_simple(data_dict: dict) -> str

Converts a dictionary into a simple human-readable string.

Parameters:

  • data_dict (dict) –

    The dictionary to convert.

Returns:

  • str –

    A string representation with each key-value pair on a new line,

  • str –

    or a message if the input is not a dictionary or is empty.

Source code in azad/prompts/dialects/native/dialect.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def dict_to_human_readable_simple(self,data_dict: dict) -> str:
    """
    Converts a dictionary into a simple human-readable string.

    Args:
        data_dict: The dictionary to convert.

    Returns:
        A string representation with each key-value pair on a new line,
        or a message if the input is not a dictionary or is empty.
    """
    if not isinstance(data_dict, dict):
        return "Error: Input must be a dictionary."

    if not data_dict:
        return "(Empty Dictionary)"

    lines = []
    for key, value in data_dict.items():
        # You can adjust the formatting here (e.g., using '->' instead of ':')
        lines.append(f"{key}: {value}")

    return "\n".join(lines)
format_example 
format_example(tool_name: str, parameters: dict) -> str

Format an example tool call in native format.

Source code in azad/prompts/dialects/native/dialect.py 
61
62
63
64
65
66
def format_example(self, tool_name: str, parameters: dict) -> str:
    """Format an example tool call in native format."""
    # Convert parameters to a human-readable string
    params_used = self.dict_to_human_readable_simple(parameters)

    return f"""Tool name: {tool_name}\nParameters used:\n{params_used}"""
format_tool_call 
format_tool_call(tool_call: ToolCallPart) -> dict

Format a tool call into native format.

Source code in azad/prompts/dialects/native/dialect.py 
68
69
70
71
72
73
74
75
76
77
def format_tool_call(self, tool_call: ToolCallPart) -> dict:
    """Format a tool call into native format."""
    # Use toolCallId instead of id (which doesn't exist in ToolCallPart)
    tool_id = tool_call.tool_call_id or f"toolu_{uuid.uuid4().hex[:24]}"
    return {
        "type": "tool_use",
        "id": tool_id,
        "name": tool_call.tool_name,
        "input": tool_call.args
    }
format_tool_result 
format_tool_result(tool_result: ToolResultPart) -> dict

Format a tool result into native format.

Source code in azad/prompts/dialects/native/dialect.py 
79
80
81
82
83
84
85
86
87
88
def format_tool_result(self, tool_result: ToolResultPart) -> dict:
    """Format a tool result into native format."""
    return {
        "type": "tool_result",
        "id": tool_result.tool_call_id,
        "name": tool_result.tool_name,
        "result": tool_result.result,
        "observerResults": tool_result.observer_results,
        "toolResultContents": tool_result.tool_result_contents
    }
format_history_item 
format_history_item(item: Message) -> Optional[dict]

Convert a Message to a LiteLLM message with native tool calls and image support.

Source code in azad/prompts/dialects/native/dialect.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def format_history_item(self, item: Message) -> Optional[dict]:
    """Convert a Message to a LiteLLM message with native tool calls and image support."""
    if item.role == MessageRole.user:
        return self._format_user_message(item)
    elif item.role == MessageRole.assistant:
        return self._format_assistant_message(item)
    elif item.role == MessageRole.tool:
        return self._format_tool_message(item)
    elif item.role == MessageRole.informational and any(info.type == "informational" and info.is_visible_ai for info in item.content):
        return self._format_informational_message(item)
    elif item.role == MessageRole.system:
        return self._format_system_message(item)
    return None  # Skip unsupported or invisible messages
create_parser 
create_parser(prompt_data: PromptData) -> DialectParser

Create an native dialect parser configured with the specified tools.

Source code in azad/prompts/dialects/native/dialect.py 
210
211
212
213
def create_parser(self, prompt_data: PromptData) -> DialectParser:
    """Create an native dialect parser configured with the specified tools."""
    schema = {"tools": [tool.name for tool in prompt_data.tool_metadata]}
    return NativeDialectParser(schema=schema, prompt_data=prompt_data, config=self.config)
is_native_toolcalling 
is_native_toolcalling() -> bool

Check if the prompt data uses native tool calling format.

Source code in azad/prompts/dialects/native/dialect.py 
215
216
217
def is_native_toolcalling(self) -> bool:
    """Check if the prompt data uses native tool calling format."""
    return True

Modules

dialect

Attributes
Classes
NativeDialect 
NativeDialect(config: NativeDialectConfig)

Bases: Dialect

Native dialect for AI agent communication using native tool calling format.

Source code in azad/prompts/dialects/native/dialect.py 
27
28
29
def __init__(self, config: NativeDialectConfig):
    self.config = config
    self.logger = logging.getLogger(__name__)
Attributes
config_cls class-attribute instance-attribute 
config_cls = NativeDialectConfig
config instance-attribute 
config = config
logger instance-attribute 
logger = getLogger(__name__)
Functions
format_dialect_rules 
format_dialect_rules(prompt_data: PromptData) -> str

Return the native dialect rules.

Source code in azad/prompts/dialects/native/dialect.py 
31
32
33
34
35
def format_dialect_rules(self, prompt_data: PromptData) -> str:
    """Return the native dialect rules."""
    with open(os.path.join(os.path.dirname(__file__), 'native.rules.prompt')) as f:
        rules = f.read()
    return rules
dict_to_human_readable_simple 
dict_to_human_readable_simple(data_dict: dict) -> str

Converts a dictionary into a simple human-readable string.

Parameters:

  • data_dict (dict) –

    The dictionary to convert.

Returns:

  • str –

    A string representation with each key-value pair on a new line,

  • str –

    or a message if the input is not a dictionary or is empty.

Source code in azad/prompts/dialects/native/dialect.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def dict_to_human_readable_simple(self,data_dict: dict) -> str:
    """
    Converts a dictionary into a simple human-readable string.

    Args:
        data_dict: The dictionary to convert.

    Returns:
        A string representation with each key-value pair on a new line,
        or a message if the input is not a dictionary or is empty.
    """
    if not isinstance(data_dict, dict):
        return "Error: Input must be a dictionary."

    if not data_dict:
        return "(Empty Dictionary)"

    lines = []
    for key, value in data_dict.items():
        # You can adjust the formatting here (e.g., using '->' instead of ':')
        lines.append(f"{key}: {value}")

    return "\n".join(lines)
format_example 
format_example(tool_name: str, parameters: dict) -> str

Format an example tool call in native format.

Source code in azad/prompts/dialects/native/dialect.py 
61
62
63
64
65
66
def format_example(self, tool_name: str, parameters: dict) -> str:
    """Format an example tool call in native format."""
    # Convert parameters to a human-readable string
    params_used = self.dict_to_human_readable_simple(parameters)

    return f"""Tool name: {tool_name}\nParameters used:\n{params_used}"""
format_tool_call 
format_tool_call(tool_call: ToolCallPart) -> dict

Format a tool call into native format.

Source code in azad/prompts/dialects/native/dialect.py 
68
69
70
71
72
73
74
75
76
77
def format_tool_call(self, tool_call: ToolCallPart) -> dict:
    """Format a tool call into native format."""
    # Use toolCallId instead of id (which doesn't exist in ToolCallPart)
    tool_id = tool_call.tool_call_id or f"toolu_{uuid.uuid4().hex[:24]}"
    return {
        "type": "tool_use",
        "id": tool_id,
        "name": tool_call.tool_name,
        "input": tool_call.args
    }
format_tool_result 
format_tool_result(tool_result: ToolResultPart) -> dict

Format a tool result into native format.

Source code in azad/prompts/dialects/native/dialect.py 
79
80
81
82
83
84
85
86
87
88
def format_tool_result(self, tool_result: ToolResultPart) -> dict:
    """Format a tool result into native format."""
    return {
        "type": "tool_result",
        "id": tool_result.tool_call_id,
        "name": tool_result.tool_name,
        "result": tool_result.result,
        "observerResults": tool_result.observer_results,
        "toolResultContents": tool_result.tool_result_contents
    }
format_history_item 
format_history_item(item: Message) -> Optional[dict]

Convert a Message to a LiteLLM message with native tool calls and image support.

Source code in azad/prompts/dialects/native/dialect.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def format_history_item(self, item: Message) -> Optional[dict]:
    """Convert a Message to a LiteLLM message with native tool calls and image support."""
    if item.role == MessageRole.user:
        return self._format_user_message(item)
    elif item.role == MessageRole.assistant:
        return self._format_assistant_message(item)
    elif item.role == MessageRole.tool:
        return self._format_tool_message(item)
    elif item.role == MessageRole.informational and any(info.type == "informational" and info.is_visible_ai for info in item.content):
        return self._format_informational_message(item)
    elif item.role == MessageRole.system:
        return self._format_system_message(item)
    return None  # Skip unsupported or invisible messages
create_parser 
create_parser(prompt_data: PromptData) -> DialectParser

Create an native dialect parser configured with the specified tools.

Source code in azad/prompts/dialects/native/dialect.py 
210
211
212
213
def create_parser(self, prompt_data: PromptData) -> DialectParser:
    """Create an native dialect parser configured with the specified tools."""
    schema = {"tools": [tool.name for tool in prompt_data.tool_metadata]}
    return NativeDialectParser(schema=schema, prompt_data=prompt_data, config=self.config)
is_native_toolcalling 
is_native_toolcalling() -> bool

Check if the prompt data uses native tool calling format.

Source code in azad/prompts/dialects/native/dialect.py 
215
216
217
def is_native_toolcalling(self) -> bool:
    """Check if the prompt data uses native tool calling format."""
    return True
format_tool_schema 
format_tool_schema(tool: ToolMetadata) -> dict

Format a tool's metadata into a JSON schema format.

Parameters:

Returns:

  • dict –

    A dictionary representing the formatted tool schema

Source code in azad/prompts/base_dialect.py 
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
def format_tool_schema(self, tool: ToolMetadata) -> dict:
    """Format a tool's metadata into a JSON schema format.

    Args:
        tool: The tool metadata to format

    Returns:
        A dictionary representing the formatted tool schema
    """
    properties = {}
    for param_name, param_info in tool.parameters.items():
        # Extract parameter information
        if isinstance(param_info, ParameterMetadata):
            description = param_info.description
            # Determine the parameter type - we don't have explicit type info in ParameterMetadata
            # so we'll default to string
            param_type = "string"
        else:
            description = str(param_info)
            param_type = "string"

        properties[param_name] = {
            "type": param_type,
            "description": description
        }

    return {
        "name": tool.name,
        "description": tool.description,
        "input_schema": {
            "type": "object",
            "properties": properties,
            "required": tool.required_parameters,
        }
    }
format_tools_schema 
format_tools_schema(tools: List[ToolMetadata]) -> List[dict]

Format multiple tools' metadata into JSON schema format.

Parameters:

  • tools (List[ToolMetadata]) –

    A list of tool metadata objects to format

Returns:

  • List[dict] –

    A list of dictionaries representing the formatted tool schemas

Source code in azad/prompts/base_dialect.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
def format_tools_schema(self, tools: List[ToolMetadata]) -> List[dict]:
    """Format multiple tools' metadata into JSON schema format.

    Args:
        tools: A list of tool metadata objects to format

    Returns:
        A list of dictionaries representing the formatted tool schemas
    """
    return [self.format_tool_schema(tool) for tool in tools]
inject_prompt_cache 
inject_prompt_cache(messages: list[dict], prompt_data: PromptData)

Modifies the formatted messages list IN-PLACE to add cache control flags based on the model type and specific caching rules.

Source code in azad/prompts/base_dialect.py 
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
def inject_prompt_cache(self, messages: list[dict], prompt_data: PromptData):
    """
    Modifies the formatted messages list IN-PLACE to add cache control flags
    based on the model type and specific caching rules.
    """
    # Check if caching is explicitly enabled via dynamic config first
    cache_dialect_config = getattr(prompt_data.dyanmic_task_config, 'cache_dialect', None)

    # --- Anthropic Caching Logic ---
    # Apply if cache_dialect_config is ANTHROPIC OR if it's None and model is Anthropic
    is_anthropic_model = "anthropic" in prompt_data.task_config.model_name # Or use a more specific check if needed
    should_apply_anthropic = is_anthropic_model and (cache_dialect_config == CacheDialect.ANTHROPIC or cache_dialect_config is None)

    if should_apply_anthropic:
        # This uses the original logic provided in the snippet, which applies
        # cache control to system, second-last user, and last user messages.
        print("Applying Anthropic caching rules (System, Last User, Second-Last User)...")

        last_system_index = -1
        last_user_index = -1
        second_last_user_index = -1

        for i, msg in enumerate(messages):
            if msg.get('role') == 'system': last_system_index = i
            elif msg.get('role') == 'user': second_last_user_index = last_user_index; last_user_index = i

        def apply_anthropic_cache(index, is_last_user=False):
            if index != -1 and index < len(messages):
                msg = messages[index]
                content = msg.get('content')
                if isinstance(content, str): msg['content'] = [{"type": "text", "text": content}]; content = msg['content']
                if not isinstance(content, list): print(f"    Warning: Anthropic - content not list: {type(content)}"); return
                if not content: print(f"    Warning: Anthropic - content list empty."); return

                last_part = content[-1]
                if isinstance(last_part, dict): last_part['cache_control'] = {'type': 'ephemeral'}
                else: print(f"    Warning: Anthropic - last part not dict: {type(last_part)}")

                if is_last_user and len(content) >= 2:
                     second_last_part = content[-2]
                     if isinstance(second_last_part, dict): second_last_part['cache_control'] = {'type': 'ephemeral'}
                     else: print(f"    Warning: Anthropic - second-last part not dict: {type(second_last_part)}")

        apply_anthropic_cache(last_system_index)
        apply_anthropic_cache(second_last_user_index)
        apply_anthropic_cache(last_user_index, is_last_user=True)
        return

    # --- Gemini Caching Logic ---
    is_gemini_model = "gemini" in prompt_data.task_config.model_name
    should_apply_gemini = is_gemini_model and (cache_dialect_config == CacheDialect.GEMINI or cache_dialect_config is None)

    if should_apply_gemini and prompt_data.dyanmic_task_config.enable_explicit_caching:
        print("Checking Gemini caching rules...")

        # Calculate total characters as a proxy for tokens
        total_chars = 0
        for msg in messages:
            content = msg.get('content')
            if isinstance(content, list):
                for part in content:
                    if isinstance(part, dict) and part.get('type') == 'text':
                        total_chars += len(part.get('text', ''))
            elif isinstance(content, str):
                total_chars += len(content)

        # Minimum character count approximation (4096 tokens * ~4 chars/token)
        min_chars_for_cache = 16000 # Adjusted slightly lower
        print(f"  Total characters calculated: {total_chars}")

        if total_chars < min_chars_for_cache:
            print(f"  Skipping Gemini caching: Character count ({total_chars}) is below threshold ({min_chars_for_cache}).")
            return # Not enough content to warrant caching

        # check if includes at least one assistant message
        has_assistant_message = any(msg.get('role') == 'assistant' for msg in messages)

        if not has_assistant_message:
            print("  Skipping Gemini caching: No assistant message found.")
            return

        print("Applying Anthropic caching rules for Gemini model (System, Last User, Second-Last User)...")

        last_system_index = -1
        last_user_index = -1
        second_last_user_index = -1

        for i, msg in enumerate(messages):
            if msg.get('role') == 'system': last_system_index = i
            elif msg.get('role') == 'user': second_last_user_index = last_user_index; last_user_index = i

        def apply_gemini_cache(index, is_last_user=False):
            if index != -1 and index < len(messages):
                msg = messages[index]
                content = msg.get('content')
                if isinstance(content, str): msg['content'] = [{"type": "text", "text": content}]; content = msg['content']
                if not isinstance(content, list): print(f"    Warning: Gemini - content not list: {type(content)}"); return
                if not content: print(f"    Warning: Gemini - content list empty."); return

                last_part = content[-1]
                if isinstance(last_part, dict): last_part['cache_control'] = {'type': 'ephemeral'}
                else: print(f"    Warning: Gemini - last part not dict: {type(last_part)}")

                if is_last_user and len(content) >= 2:
                     second_last_part = content[-2]
                     if isinstance(second_last_part, dict): second_last_part['cache_control'] = {'type': 'ephemeral'}
                     else: print(f"    Warning: Gemini - second-last part not dict: {type(second_last_part)}")

        apply_gemini_cache(last_system_index)
        apply_gemini_cache(second_last_user_index)
        apply_gemini_cache(last_user_index, is_last_user=True)

        return

    # No explicit return needed
    return
format_messages 
format_messages(prompt_data: PromptData, rules_path: Optional[str]) -> list[dict]
Source code in azad/prompts/base_dialect.py 
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
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
def format_messages(self, prompt_data:PromptData, rules_path: Optional[str]) -> list[dict]:
    system_prompt = self.format_system_prompt(prompt_data, rules_path)
    history = self.format_history(prompt_data)

    # delete any messages with empty content or empty tool_calls
    history = [msg for msg in history if msg.get('content') and len(msg['content']) > 0
               or (msg.get('tool_calls') and len(msg['tool_calls']) > 0)]


    formated = [system_prompt, *history]
    formated = self._interleave_consecutive_messages(formated)

    # Check if it's a kodu provider
    is_kodu_provider = "kodu" in prompt_data.task_config.model_name

    if is_kodu_provider:
        from ..env_settings import settings
        from ..ainetwork.errors import AIInsufficientCreditsError, AIUserNotFoundError

        if settings.FLY_PROCESS_GROUP is not None and settings.DATABASE_URL is not None and settings.TURSO_AUTH_TOKEN is not None:
            user_api_key = prompt_data.dyanmic_task_config.model_api_key
            user_credits = 0
            credit_threshold = settings.CREDIT_THRESHOLD
            try:
                from ..db_models import db

                user_credits = db.get_user_credits(user_api_key)
            except Exception as e:
                print(f"Error in Kodu provider check: {str(e)}")

            if user_credits is not None:
                if user_credits < credit_threshold:
                    raise AIInsufficientCreditsError()
            else:
                raise AIUserNotFoundError()


    self.inject_prompt_cache(formated, prompt_data)


    # if we have dynamic environment details, add them to the last user message if it exists
    if prompt_data.dyanmic_task_config.dynamic_environment_details_block:
        last_user_index = -1
        for i, msg in enumerate(formated):
            if msg.get('role') == 'user':
                last_user_index = i
        if last_user_index != -1:
            # validate that the content field is array
            if isinstance(formated[last_user_index]['content'], list):
                formated[last_user_index]['content'].append({
                    "type":"text",
                    "text":prompt_data.dyanmic_task_config.dynamic_environment_details_block
                })
            else:
                # edge case this should not happen
                formated[last_user_index]['content'] = [
                    {
                        "type":"text",
                        "text":formated[last_user_index]['content']
                    },
                    {
                        "type":"text",
                        "text":prompt_data.dyanmic_task_config.dynamic_environment_details_block
                    }
                ]

    return formated
format_system_prompt 
format_system_prompt(prompt_data: PromptData, rules_path: Optional[str]) -> dict

Format the system prompt for the dialect. This method handles the replacement of placeholders in the system prompt with the actual values from the prompt data. If there is no placeholder it will format the system prompt in following way: TOOL_USE_INSTRUCTION AVAILABLE_TOOLS USER AGENT PROMPT SYSTEM GENERAL INSTRUCTIONS USER CUSTOM INSTRUCTIONS

Source code in azad/prompts/base_dialect.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
def format_system_prompt(self, prompt_data:PromptData, rules_path: Optional[str]) -> dict:
    """Format the system prompt for the dialect.
    This method handles the replacement of placeholders in the system prompt
    with the actual values from the prompt data.
    If there is no placeholder it will format the system prompt in following way:
    TOOL_USE_INSTRUCTION
    AVAILABLE_TOOLS
    USER AGENT PROMPT
    SYSTEM GENERAL INSTRUCTIONS
    USER CUSTOM INSTRUCTIONS
    """
    system = f"""{self.format_user_agent_prompt(prompt_data)}"""
    # check if {{AVAILABLE_TOOLS}} is in the system prompt
    def is_key_in_system_prompt(key: str) -> bool:
        return key in system
    def format_key(key: str) -> str:
        return f"{{{key}}}"
    system_key = format_key(PromptTemplateType.SYSTEM_INSTRUCTION)
    available_tools_key = format_key(PromptTemplateType.AVAILABLE_TOOLS)
    tool_use_instruction_key = format_key(PromptTemplateType.TOOL_USE_INSTRUCTION)
    user_prompt_key = format_key(PromptTemplateType.USER_PROMPT)
    if is_key_in_system_prompt(system_key):
        system = system.replace(system_key, self.format_system_rules(prompt_data))
    else:
        system = f"{system}\n{self.format_system_rules(prompt_data)}"
    if is_key_in_system_prompt(available_tools_key):
        system = system.replace(available_tools_key, self.format_tool_docs(prompt_data))
    else:
        system = f"{self.format_tool_docs(prompt_data)}\n{system}"
    if is_key_in_system_prompt(tool_use_instruction_key):
        system = system.replace(tool_use_instruction_key, self.format_dialect_rules(prompt_data, rules_path))
    else:
        system = f"{self.format_dialect_rules(prompt_data, rules_path)}\n{system}"
    if is_key_in_system_prompt(user_prompt_key):
        system = system.replace(user_prompt_key, self.format_user_prompt(prompt_data))
    else:
        system = f"{system}\n{self.format_user_prompt(prompt_data)}"
    return dict(content=[{"type":"text","text": system}],role="system")
format_system_rules 
format_system_rules(prompt_data)
Source code in azad/prompts/base_dialect.py 
353
354
355
356
357
358
359
    def format_system_rules(self, prompt_data):
        return """# You have special messages called Informational Messages that are generated by the local environment and are not part of the user's input. These messages may be visible or invisible to the user, you should observe the informational messages and take them into account when responding to the user in accordance to the task.
# Pay extra attention to your mistakes and try to self improve by learning from them and acting better in the future.
# You should always try to be direct and clear generating the best possible output for the user task.
# When commuinicating with the user about informational messages, you should always be clear and understand that informationals messages are generated by the local environment and the user may not be aware of them.
# This means when talking about any details from the informational messages, you should say the environment instead of the user, so when talking about the informational content always say "the environment" instead of "the user".
"""
format_history 
format_history(prompt_data: PromptData) -> List[dict]

Format the entire message history from the mindmap.

Source code in azad/prompts/base_dialect.py 
391
392
393
394
def format_history(self, prompt_data: PromptData) -> List[dict]:
    """Format the entire message history from the mindmap."""
    messages = prompt_data.messages
    return [msg for msg in (self.format_history_item(item) for item in messages) if msg is not None]
format_user_prompt 
format_user_prompt(prompt_data: PromptData) -> str
Source code in azad/prompts/base_dialect.py 
396
397
398
399
400
401
402
403
404
405
def format_user_prompt(self, prompt_data:PromptData) -> str:
    if not prompt_data.task_config.user_prompt:
        return ""

    return inspect.cleandoc(f"""====
        The user has provided additional instructions or details for you to use, Please understand the the user may or may not have knowledge of the overall system instructions, and this is their attempt to configure your behavior to match their needs.
        Here is the user custom instructions:
        {prompt_data.task_config.user_prompt}
        ====
        """)
format_user_agent_prompt 
format_user_agent_prompt(prompt_data: PromptData) -> str
Source code in azad/prompts/base_dialect.py 
407
408
409
410
411
def format_user_agent_prompt(self, prompt_data:PromptData) -> str:
    if not prompt_data.task_config.user_agent_prompt:
        return ""
    # return inspect.cleandoc(f"""<agent_prompt>{prompt_data.task_config.user_agent_prompt}</agent_prompt>""")
    return prompt_data.task_config.user_agent_prompt
format_tool_docs 
format_tool_docs(prompt_data: PromptData) -> str

Format documentation for available tools.

This base implementation handles the structure, while specific dialects format the individual examples.

Parameters:

  • tools –

    List of Tool classes or ToolSignature objects to document

Returns:

  • str –

    Formatted tool documentation string

Source code in azad/prompts/base_dialect.py 
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
    def format_tool_docs(self, prompt_data:PromptData) -> str:
        """Format documentation for available tools.

        This base implementation handles the structure, while specific dialects
        format the individual examples.

        Args:
            tools: List of Tool classes or ToolSignature objects to document

        Returns:
            Formatted tool documentation string
        """

        tools: list[ToolMetadata] = prompt_data.tool_metadata
        docs = []
        for tool in tools:
            # Handle both Tool and ToolSignature objects
            params = []
            for name, param_info in tool.parameters.items():
                required = "(required)" if name in tool.required_parameters else "(optional)"
                params.append(f"- {name}: {required} {self._get_parameter_description(param_info)}") # type: ignore

            examples = []
            for i, example in enumerate(tool.examples, 1):
                formatted_example = self.format_example(tool.name, example.parameters) # type: ignore
                examples.append(f"""<tool_call_example tool_name="{tool.name}">### {example.explanation}\n> Azad Output :\n{formatted_example}\n</<tool_call_example>""")

            doc = f"""<tool_doc tool_name="{tool.name}">
Tool name: {tool.name}
Tool Description: {tool.description}
Tool Input Parameters:
{"".join(params)}
Tool Usage examples:
<tool_call_examples tool_name="{tool.name}">
Here is a list of examples of how to use the tool, please read them carefully to understand how to use the tool effectively, if no instructions are provided, please use the tool as you see fit.
{"".join(examples)}
</tool_call_examples>
"""
            docs.append(doc)

        return f"""## Tool Documentation
Here are all the tools available for you to use in the task, please read the documentation carefully to understand how to use them effectively.
{"\n\n".join(docs).strip()}
"""
Functions

parser

Native dialect parser for streaming tool calls with real-time event emission.

Attributes
Classes
NativeDialectConfig

Bases: DialectConfig

Configuration for Native dialect parser.

Attributes
pretty_print class-attribute instance-attribute 
pretty_print: bool = Field(default=True, description='Whether to format JSON with indentation')
indentation class-attribute instance-attribute 
indentation: int = Field(default=2, description='Indentation level for pretty printing')
parameter_buffer_threshold class-attribute instance-attribute 
parameter_buffer_threshold: int = Field(default=15, description='Minimum characters for parameter chunks')
flush_interval_ms class-attribute instance-attribute 
flush_interval_ms: int = Field(default=100, description='Milliseconds between forced flushes')
NativeDialectParser 
NativeDialectParser(schema: Dict, prompt_data: PromptData, config: NativeDialectConfig)

Bases: DialectParser

Native dialect parser for processing streaming tool call deltas.

This parser processes tool call deltas directly from the model and efficiently emits events with proper buffering to reduce event frequency while maintaining responsiveness. It implements both size-based and time-based buffering mechanisms to optimize the trade-off between low latency and reducing excessive event emissions.

Initialize the parser with the given schema and configuration.

Source code in azad/prompts/dialects/native/parser.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(self, schema: Dict, prompt_data: PromptData, config: NativeDialectConfig):
    """Initialize the parser with the given schema and configuration."""
    self.schema = schema
    self.prompt_data = prompt_data
    self.config = config
    self.logger = logging.getLogger(__name__)

    # Set default buffering thresholds if not specified in config
    self.PARAMETER_BUFFER_THRESHOLD = config.parameter_buffer_threshold
    self.FLUSH_INTERVAL_MS = config.flush_interval_ms

    # Initialize parser state
    self.reset_state()
Attributes
schema instance-attribute 
schema = schema
prompt_data instance-attribute 
prompt_data = prompt_data
config instance-attribute 
config = config
logger instance-attribute 
logger = getLogger(__name__)
PARAMETER_BUFFER_THRESHOLD instance-attribute 
PARAMETER_BUFFER_THRESHOLD = parameter_buffer_threshold
FLUSH_INTERVAL_MS instance-attribute 
FLUSH_INTERVAL_MS = flush_interval_ms
Functions
reset_state 
reset_state()

Reset the parser state for a new tool call.

Source code in azad/prompts/dialects/native/parser.py 
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
def reset_state(self):
    """Reset the parser state for a new tool call."""
    # Tool call tracking
    self.tool_call_id = None
    self.current_tool_name = None
    self.in_tool_call = False
    self.tool_ready_emitted = False

    # Track multiple tool calls
    self.detected_tool_calls = []

    # Text state
    self.sent_text_start = False

    # Parameter tracking
    self.params_seen = set()               # Parameters we've seen start events for
    self.params_ended = set()              # Parameters that have ended
    self.param_values = {}                 # Current complete value for each parameter
    self.parameter_chunk_buffer = {}       # Buffer for accumulating parameter chunks
    self.last_update_length = {}           # Track last update length for each param
    self.last_flush_time = time.time() * 1000  # Last time we flushed chunks (in ms)

    # JSON parsing state
    self.raw_buffer = ""                   # Raw JSON accumulation
    self.is_complete = False               # Whether the tool call is complete
    self.first_key_value_extracted = False # Whether we've seen the first key-value pair
feed 
feed(data: bytes) -> List[AINetworkEventUnion]

Process plain text content (outside of a tool call).

This method is called when the model outputs regular text instead of a tool call.

Source code in azad/prompts/dialects/native/parser.py 
 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
def feed(self, data: bytes) -> List[AINetworkEventUnion]:
    """
    Process plain text content (outside of a tool call).

    This method is called when the model outputs regular text
    instead of a tool call.
    """
    events: List[AINetworkEventUnion] = []
    text = data.decode('utf-8')

    # Skip empty text
    if not text:
        return events

    # Ignore text when in a tool call (should use feed_tool_call_delta instead)
    if self.in_tool_call:
        return events

    # Start text section if needed
    if not self.sent_text_start:
        events.append(AINetworkEventTextStart())
        self.sent_text_start = True

    # Emit text chunk
    events.append(AINetworkEventTextChunk(content=text))
    return events
feed_tool_call_delta 
feed_tool_call_delta(tool_call: Union[ChatCompletionDeltaToolCall, ChatCompletionMessageToolCall]) -> List[AINetworkEventUnion]

Process tool call deltas from the model with efficient buffering.

This method handles the incremental parts of a tool call that come from the model, buffering small chunks for improved efficiency.

Source code in azad/prompts/dialects/native/parser.py 
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
def feed_tool_call_delta(
    self,
    tool_call: Union[ChatCompletionDeltaToolCall, ChatCompletionMessageToolCall]
) -> List[AINetworkEventUnion]:
    """
    Process tool call deltas from the model with efficient buffering.

    This method handles the incremental parts of a tool call that come
    from the model, buffering small chunks for improved efficiency.
    """
    events = []

    print(f"Received tool call delta: {tool_call}")
    # Skip if tool_call is invalid
    if not tool_call or not hasattr(tool_call, 'function'):
        return events

    # End text mode if we're transitioning to a tool call
    if self.sent_text_start and not self.in_tool_call:
        events.append(AINetworkEventTextEnd())
        self.sent_text_start = False

    function = tool_call.function

    # Start a new tool call if we have a name and ID
    if function.name is not None and tool_call.id is not None and not self.in_tool_call:
        self.tool_call_id = tool_call.id
        self.current_tool_name = function.name
        self.in_tool_call = True

        # Reset timer for new tool call
        self.last_flush_time = time.time() * 1000

        # Emit tool name event
        events.append(AINetworkEventToolName(
            tool_name=function.name,
            tool_call_id=tool_call.id
        ))

    # Process function arguments if provided
    if function.arguments is not None:
        new_content = function.arguments

        # Add to our raw JSON buffer
        self.raw_buffer += new_content

        # Process the updated JSON buffer
        new_events = self._process_json_buffer()
        events.extend(new_events)

        # Do any time-based flushing for all parameters
        current_time = time.time() * 1000
        if current_time - self.last_flush_time >= self.FLUSH_INTERVAL_MS:
            # Force flush any buffers that have content
            for param_name in list(self.parameter_chunk_buffer.keys()):
                if param_name in self.params_seen and param_name not in self.params_ended and self.parameter_chunk_buffer[param_name]:
                    events.append(AINetworkEventParameterChunk(
                        parameter=param_name,
                        content=self.parameter_chunk_buffer[param_name],
                        tool_call_id=self.tool_call_id # type: ignore
                    ))
                    # Update last update length
                    if param_name in self.param_values:
                        self.last_update_length[param_name] = len(self.param_values[param_name])
                    # Clear buffer
                    self.parameter_chunk_buffer[param_name] = ""

            # Update flush time
            self.last_flush_time = current_time

        # Check for completion
        if self._is_json_complete() and not self.is_complete:
            self.is_complete = True

            # End all parameters that haven't ended yet
            for param in self.params_seen:
                if param not in self.params_ended:
                    # Emit any remaining buffer content
                    if param in self.parameter_chunk_buffer and self.parameter_chunk_buffer[param]:
                        events.append(AINetworkEventParameterChunk(
                            parameter=param,
                            content=self.parameter_chunk_buffer[param],
                            tool_call_id=self.tool_call_id # type: ignore

                        ))
                        self.parameter_chunk_buffer[param] = ""

                    # End the parameter
                    events.append(AINetworkEventParameterEnd(parameter=param))
                    self.params_ended.add(param)
            # Emit parameters complete if we have a tool call ID - we should have
            if self.tool_call_id:
                events.append(AINetworkEventParametersComplete(tool_call_id=self.tool_call_id))
            else:
                self.logger.warning("No tool call ID found when emitting parameters complete")

            # Store the completed tool call for later reference
            if self.current_tool_name and self.tool_call_id:
                # Create the final parameters dict
                final_params = {}
                for param_name, param_value in self.param_values.items():
                    final_params[param_name] = param_value

                # Add to detected tool calls list
                self.detected_tool_calls.append({
                    "tool_name": self.current_tool_name,
                    "tool_call_id": self.tool_call_id,
                    "args": final_params
                })

                # Log the tool call detection
                self.logger.debug(f"Detected tool call: {self.current_tool_name} with ID: {self.tool_call_id}")

            # Emit tool ready if not already emitted
            if not self.tool_ready_emitted:
                self.tool_ready_emitted = True

                # Try to parse the final parameters
                final_params = {}
                for param_name, param_value in self.param_values.items():
                    final_params[param_name] = param_value

                # Emit tool ready event
                events.append(AINetworkEventToolReady(
                    tool_name=self.current_tool_name, # type: ignore
                    tool_call_id=self.tool_call_id, # type: ignore
                    args=final_params
                ))

    # Force-flush any remaining buffered content at the end of processing
    if self.in_tool_call:
        for param_name in list(self.parameter_chunk_buffer.keys()):
            if param_name in self.params_seen and param_name not in self.params_ended and self.parameter_chunk_buffer[param_name]:
                events.append(AINetworkEventParameterChunk(
                    parameter=param_name,
                    content=self.parameter_chunk_buffer[param_name],
                    tool_call_id=self.tool_call_id # type: ignore

                ))
                # Update the last update length
                if param_name in self.param_values:
                    self.last_update_length[param_name] = len(self.param_values[param_name])
                # Clear the buffer
                self.parameter_chunk_buffer[param_name] = ""

    return events
get_detected_tool_calls 
get_detected_tool_calls() -> List[Dict[str, Any]]

Return all detected tool calls.

Source code in azad/prompts/dialects/native/parser.py 
432
433
434
def get_detected_tool_calls(self) -> List[Dict[str, Any]]:
    """Return all detected tool calls."""
    return self.detected_tool_calls
is_multi_tool_response 
is_multi_tool_response() -> bool

Check if this was a multiple tool call response.

Source code in azad/prompts/dialects/native/parser.py 
436
437
438
def is_multi_tool_response(self) -> bool:
    """Check if this was a multiple tool call response."""
    return len(self.detected_tool_calls) > 1

test_native_parser

Test suite for the Native dialect parser.

Attributes
Classes
MockPromptData 
MockPromptData()

Mock prompt data for testing.

Source code in azad/prompts/dialects/native/test_native_parser.py 
26
27
28
29
30
31
def __init__(self):
    self.task = None
    self.task_config = None
    self.tool_metadata = []
    self.current_tool_call_id = None
    self.current_assistant_id = None
Attributes
task instance-attribute 
task = None
task_config instance-attribute 
task_config = None
tool_metadata instance-attribute 
tool_metadata = []
current_tool_call_id instance-attribute 
current_tool_call_id = None
current_assistant_id instance-attribute 
current_assistant_id = None
TestNativeParser

Bases: TestCase

Test suite for the Native dialect parser.

Functions
setUp 
setUp()

Set up test case.

Source code in azad/prompts/dialects/native/test_native_parser.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def setUp(self):
    """Set up test case."""
    schema = {
        "tools": {
            "attempt_completion": {
                "parameters": {
                    "result": {"type": "string"}
                }
            }
        }
    }
    self.prompt_data = MockPromptData()
    self.config = NativeDialectConfig()
    self.parser = NativeDialectParser(schema, self.prompt_data, self.config) # type: ignore
create_tool_call_delta 
create_tool_call_delta(tool_id: Optional[str] = None, tool_name: Optional[str] = None, arguments: str = '', tool_type: str = 'function', index: int = 0) -> ChatCompletionDeltaToolCall

Helper to create a tool call delta with the specified properties.

Source code in azad/prompts/dialects/native/test_native_parser.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def create_tool_call_delta(
    self,
    tool_id: Optional[str] = None,
    tool_name: Optional[str] = None,
    arguments: str = "",
    tool_type: str = "function",
    index: int = 0
) -> ChatCompletionDeltaToolCall:
    """Helper to create a tool call delta with the specified properties."""
    return ChatCompletionDeltaToolCall(
        id=tool_id,
        function=Function(
            arguments=arguments,
            name=tool_name
        ),
        type=tool_type,
        index=index
    )
test_native_parser_with_realistic_stream 
test_native_parser_with_realistic_stream()

Test parsing a realistic stream of tool call deltas.

Source code in azad/prompts/dialects/native/test_native_parser.py 
 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
def test_native_parser_with_realistic_stream(self):
    """Test parsing a realistic stream of tool call deltas."""
    events = []

    # Simulate chunks as seen in the LiteLLM stream example
    # First chunk with tool ID and name
    tool_call = self.create_tool_call_delta(
        tool_id='toolu_01T8rmwSEpbXrr5fL8qZY5jt',
        tool_name='attempt_completion',
        arguments=''
    )
    events.extend(self.parser.feed_tool_call_delta(tool_call))

    # Subsequent chunks with argument pieces
    argument_pieces = [
        '{"',
        'result": "T',
        'he ',
        'sum o',
        'f 25 +',
        ' 25 eq',
        'uals 50."}'
    ]

    for piece in argument_pieces:
        tool_call = self.create_tool_call_delta(arguments=piece)
        events.extend(self.parser.feed_tool_call_delta(tool_call))

    # Validate events
    self._validate_events(events)
test_stream_with_long_parameter 
test_stream_with_long_parameter()

Test parsing a stream with a longer parameter value that comes in many pieces.

Source code in azad/prompts/dialects/native/test_native_parser.py 
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
def test_stream_with_long_parameter(self):
    """Test parsing a stream with a longer parameter value that comes in many pieces."""
    events = []

    # First chunk with tool ID and name
    tool_call = self.create_tool_call_delta(
        tool_id='toolu_01T8rmwSEpbXrr5fL8qZY5jt',
        tool_name='attempt_completion',
        arguments=''
    )
    events.extend(self.parser.feed_tool_call_delta(tool_call))

    # Start JSON
    tool_call = self.create_tool_call_delta(arguments='{"result": "')
    events.extend(self.parser.feed_tool_call_delta(tool_call))

    # Send a longer text in small chunks (simulating letter-by-letter streaming)
    long_text = "I have completed the task of analyzing the data. The results show that 42% of users preferred the new UI design, while 58% preferred the original. Key feedback points included: better color contrast in the new design, more intuitive navigation in the original, and concerns about button placement in both designs."
    chunk_size = 3  # Really small chunks to test buffering

    for i in range(0, len(long_text), chunk_size):
        chunk = long_text[i:i+chunk_size]
        tool_call = self.create_tool_call_delta(arguments=chunk)
        events.extend(self.parser.feed_tool_call_delta(tool_call))

    # Close JSON
    tool_call = self.create_tool_call_delta(arguments='"}\n')
    events.extend(self.parser.feed_tool_call_delta(tool_call))

    # Validate that we didn't get an event for every single chunk
    param_chunks = [e for e in events if isinstance(e, AINetworkEventParameterChunk)]

    # We should have fewer parameter chunk events than individual chunks we sent
    # This validates that the parser is bundling small chunks together
    expected_max_chunks = (len(long_text) // chunk_size) + 2  # +2 for start/end pieces
    self.assertLess(len(param_chunks), expected_max_chunks, 
                    f"Expected fewer than {expected_max_chunks} parameter chunks, got {len(param_chunks)}")

    # But we should still have the complete parameter value when combined
    combined = "".join([chunk.content for chunk in param_chunks])
    self.assertIn(long_text, combined, "Combined parameter chunks should contain the full text")