-
Notifications
You must be signed in to change notification settings - Fork 850
Expand file tree
/
Copy pathagent_config.py
More file actions
241 lines (196 loc) · 8.58 KB
/
agent_config.py
File metadata and controls
241 lines (196 loc) · 8.58 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
"""Experimental agent configuration utilities.
This module provides utilities for creating agents from configuration files or dictionaries.
Note: Configuration-based agent setup only works for tools that don't require code-based
instantiation. For tools that need constructor arguments or complex setup, use the
programmatic approach after creating the agent:
agent = config_to_agent("config.json")
# Add tools that need code-based instantiation
agent.tool_registry.process_tools([ToolWithConfigArg(HttpsConnection("localhost"))])
The ``model`` field supports two formats:
**String format (backward compatible — defaults to Bedrock):**
{"model": "us.anthropic.claude-sonnet-4-20250514-v1:0"}
**Object format (supports all providers):**
{
"model": {
"provider": "anthropic",
"model_id": "claude-sonnet-4-20250514",
"max_tokens": 10000,
"client_args": {"api_key": "..."}
}
}
Note: The following constructor parameters cannot be specified from JSON because they require
code-based instantiation: ``boto_session`` (Bedrock, SageMaker), ``client`` (OpenAI, Gemini),
``gemini_tools`` (Gemini). Use ``region_name`` / ``client_args`` as JSON-friendly alternatives.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import TYPE_CHECKING, Any
import jsonschema
from jsonschema import ValidationError
if TYPE_CHECKING:
from ..models.model import Model
# JSON Schema for agent configuration
AGENT_CONFIG_SCHEMA = {
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "Agent Configuration",
"description": "Configuration schema for creating agents",
"type": "object",
"properties": {
"name": {"description": "Name of the agent", "type": ["string", "null"], "default": None},
"model": {
"description": (
"The model to use for this agent. Can be a string (Bedrock model_id) "
"or an object with a 'provider' field for any supported provider."
),
"oneOf": [
{"type": "string"},
{"type": "null"},
{
"type": "object",
"properties": {
"provider": {
"description": "The model provider name",
"type": "string",
}
},
"required": ["provider"],
"additionalProperties": True,
},
],
"default": None,
},
"prompt": {
"description": "The system prompt for the agent. Provides high level context to the agent.",
"type": ["string", "null"],
"default": None,
},
"tools": {
"description": "List of tools the agent can use. Can be file paths, "
"Python module names, or @tool annotated functions in files.",
"type": "array",
"items": {"type": "string"},
"default": [],
},
},
"additionalProperties": False,
}
# Pre-compile validator for better performance
_VALIDATOR = jsonschema.Draft7Validator(AGENT_CONFIG_SCHEMA)
# Provider name to model class name — resolved via strands.models lazy __getattr__
PROVIDER_MAP: dict[str, str] = {
"bedrock": "BedrockModel",
"anthropic": "AnthropicModel",
"openai": "OpenAIModel",
"gemini": "GeminiModel",
"ollama": "OllamaModel",
"litellm": "LiteLLMModel",
"mistral": "MistralModel",
"llamaapi": "LlamaAPIModel",
"llamacpp": "LlamaCppModel",
"sagemaker": "SageMakerAIModel",
"writer": "WriterModel",
"openai_responses": "OpenAIResponsesModel",
}
def _create_model_from_dict(model_config: dict[str, Any]) -> Model:
"""Create a Model instance from a provider config dict.
Routes the config to the appropriate model class based on the ``provider`` field,
then delegates to the class's ``from_dict`` method. All imports are lazy to avoid
requiring optional dependencies that are not installed.
Args:
model_config: Dict containing at least a ``provider`` key and provider-specific params.
Returns:
A configured Model instance for the specified provider.
Raises:
ValueError: If the provider name is not recognized.
ImportError: If the provider's optional dependencies are not installed.
"""
config = model_config.copy()
provider = config.pop("provider")
class_name = PROVIDER_MAP.get(provider)
if class_name is None:
supported = ", ".join(sorted(PROVIDER_MAP.keys()))
raise ValueError(f"Unknown model provider: '{provider}'. Supported providers: {supported}")
from .. import models
model_cls: type[Model] = getattr(models, class_name)
return model_cls.from_dict(config)
def config_to_agent(config: str | dict[str, Any], **kwargs: Any) -> Any:
"""Create an Agent from a configuration file or dictionary.
This function supports tools that can be loaded declaratively (file paths, module names,
or @tool annotated functions). For tools requiring code-based instantiation with constructor
arguments, add them programmatically after creating the agent:
agent = config_to_agent("config.json")
agent.process_tools([ToolWithConfigArg(HttpsConnection("localhost"))])
Args:
config: Either a file path (with optional file:// prefix) or a configuration dictionary
**kwargs: Additional keyword arguments to pass to the Agent constructor
Returns:
Agent: A configured Agent instance
Raises:
FileNotFoundError: If the configuration file doesn't exist
json.JSONDecodeError: If the configuration file contains invalid JSON
ValueError: If the configuration is invalid or tools cannot be loaded
Examples:
Create agent from file:
>>> agent = config_to_agent("/path/to/config.json")
Create agent from file with file:// prefix:
>>> agent = config_to_agent("file:///path/to/config.json")
Create agent from dictionary:
>>> config = {"model": "anthropic.claude-3-5-sonnet-20241022-v2:0", "tools": ["calculator"]}
>>> agent = config_to_agent(config)
Create agent with object model config:
>>> config = {
... "model": {"provider": "openai", "model_id": "gpt-4o", "client_args": {"api_key": "..."}}
... }
>>> agent = config_to_agent(config)
"""
# Parse configuration
if isinstance(config, str):
# Handle file path
file_path = config
# Remove file:// prefix if present
if file_path.startswith("file://"):
file_path = file_path[7:]
# Load JSON from file
config_path = Path(file_path)
if not config_path.exists():
raise FileNotFoundError(f"Configuration file not found: {file_path}")
with open(config_path) as f:
config_dict = json.load(f)
elif isinstance(config, dict):
config_dict = config.copy()
else:
raise ValueError("Config must be a file path string or dictionary")
# Validate configuration against schema
try:
_VALIDATOR.validate(config_dict)
except ValidationError as e:
# Provide more detailed error message
error_path = " -> ".join(str(p) for p in e.absolute_path) if e.absolute_path else "root"
raise ValueError(f"Configuration validation error at {error_path}: {e.message}") from e
# Prepare Agent constructor arguments
agent_kwargs: dict[str, Any] = {}
# Handle model field — string vs object format
model_value = config_dict.get("model")
if isinstance(model_value, dict):
# Object format: create Model instance via factory
agent_kwargs["model"] = _create_model_from_dict(model_value)
elif model_value is not None:
# String format (backward compat): pass directly as model_id to Agent
agent_kwargs["model"] = model_value
# Map remaining configuration keys to Agent constructor parameters
config_mapping = {
"prompt": "system_prompt",
"tools": "tools",
"name": "name",
}
# Only include non-None values from config
for config_key, agent_param in config_mapping.items():
if config_key in config_dict and config_dict[config_key] is not None:
agent_kwargs[agent_param] = config_dict[config_key]
# Override with any additional kwargs provided
agent_kwargs.update(kwargs)
# Import Agent at runtime to avoid circular imports
from ..agent import Agent
# Create and return Agent
return Agent(**agent_kwargs)