heaven_base.configs.hermes_config
1# hermes_config.py 2import json 3from dataclasses import dataclass, field 4from pydantic import Field, BaseModel 5from typing import Optional, List, Dict, Any, Union, TypedDict, Type 6from collections.abc import Callable 7from .base_config import BaseFunctionConfig 8 9 10class HermesConfig(BaseFunctionConfig): 11 """ 12 A configuration class for defining a Hermes execution. 13 It must define func_name and args_template as required by BaseFunctionConfig 14 """ 15 func_name: str = "use_hermes" # The function this config is FOR, for metadata purposes, not Callable 16 args_template: Dict[str, Any] = Field( 17 default_factory=lambda: { 18 "goal": "", 19 "iterations": 1, 20 "agent": None, 21 "history_id": None, 22 "return_summary": False, 23 "ai_messages_only": True, 24 "continuation": None, 25 "additional_tools": None, 26 "remove_agents_config_tools": False, 27 "orchestration_preprocess": False, 28 "variable_inputs": {}, 29 "system_prompt_suffix": None 30 } 31 ) 32 33 def to_command_data(self, variable_inputs: Optional[Dict[str, Union[Dict[str, Any], List[Any]]]] = None) -> Dict[str, Any]: 34 # print("\n=== START TO_COMMAND_DATA ===") 35 # print(f"Initial args_template: {json.dumps(self.args_template, indent=2)}") 36 # print(f"Received variable_inputs: {json.dumps(variable_inputs, indent=2)}") 37 38 command_data = self.args_template.copy() 39 #### TESTING THIS 40 # Apply any custom args_template values that differ from defaults 41 # Apply any custom args_template values that differ from defaults 42 for key, value in self.args_template.items(): 43 if value != command_data.get(key): 44 command_data[key] = value 45 46 # print(f"After applying custom args: {json.dumps(command_data, indent=2)}") 47 #### 48 if not variable_inputs: 49 print("No variable_inputs provided, returning default command_data") 50 return command_data 51 52 # Get the template configuration 53 template_config = command_data["variable_inputs"] 54 # print(f"\nTemplate config from args_template: {json.dumps(template_config, indent=2)}") 55 56 # Apply the provided values to their respective parameters 57 for param_name, param_config in template_config.items(): 58 # print(f"\nProcessing parameter: {param_name}") 59 # print(f"Parameter config: {json.dumps(param_config, indent=2)}") 60 61 if not param_config.get("template"): 62 print(f"Skipping {param_name} - not templated") 63 continue 64 65 if param_name not in variable_inputs: 66 print(f"Skipping {param_name} - not in variable_inputs") 67 continue 68 69 # print(f"Value from variable_inputs: {json.dumps(variable_inputs[param_name], indent=2)}") 70 71 # Special handling for goal (or any parameter with variables) 72 if "variables" in param_config: 73 # print(f"Processing as variable template") 74 # print(f"Before: {command_data[param_name]}") 75 command_data[param_name] = command_data[param_name].format(**variable_inputs[param_name]) 76 # print(f"After: {command_data[param_name]}") 77 else: 78 # print(f"Processing as direct value") 79 # print(f"Before: {command_data[param_name]}") 80 command_data[param_name] = variable_inputs[param_name] 81 # print(f"After: {command_data[param_name]}") 82 83 # print(f"\nFinal command_data: {json.dumps(command_data, indent=2)}") 84 # print("=== END TO_COMMAND_DATA ===\n") 85 return command_data 86 87 88 89 90 91 92class HermesConfigInput(BaseModel): 93 """Represents input mapping for a Hermes configuration""" 94 source_key: str # Key to extract from previous result 95 transform: Optional[Callable] = None # Optional transformation function 96 required: bool = True # Whether this input is mandatory 97 98class DovetailModel(BaseModel): 99 """Defines how to chain Hermes configurations""" 100 expected_outputs: List[str] 101 input_map: Dict[str, HermesConfigInput] 102 103 def prepare_next_config(self, previous_result: Dict[str, Any]) -> Dict[str, Any]: 104 """Prepare inputs for the next Hermes configuration""" 105 next_config_inputs = {} 106 107 for config_key, input_spec in self.input_map.items(): 108 # Extract source value 109 source_value = previous_result.get(input_spec.source_key) 110 111 # Apply optional transformation 112 if input_spec.transform: 113 source_value = input_spec.transform(source_value) 114 115 # Check required inputs 116 if input_spec.required and source_value is None: 117 raise ValueError(f"Required input {config_key} not found") 118 119 next_config_inputs[config_key] = source_value 120 121 return next_config_inputs 122 123 124 125 126 127# Here’s a sketch of two approaches: 128 129# A plain Python “runner” class 130 131 132# class HermesChainRunner: 133 134# def __init__(self, base_config: HermesConfig, dovetail: DovetailModel, max_steps: int = 5): 135 136# self.config = base_config 137 138# self.dovetail = dovetail 139 140# self.max_steps = max_steps 141 142# self.last_result = None 143 144# self.last_extracts = {} 145 146 147# def run(self): 148 149# from heaven_base.tool_utils.hermes_utils import hermes_step 150 151 152# for i in range(self.max_steps): 153 154# result = hermes_step(hermes_config=self.config) 155 156# status = result.get("agent_status") 157 158# extracts = getattr(status, "extracted_content", {}) or {} 159 160# self.last_result, self.last_extracts = result, extracts 161 162 163# # stop when all expected keys are present 164 165# if all(k in extracts for k in self.dovetail.expected_outputs): 166 167# break 168 169 170# next_args = self.dovetail.prepare_next_config(extracts) 171 172# self.config.args_template.update(next_args) 173 174 175# return self.last_result, self.last_extracts 176# A Pydantic-driven orchestrator model 177 178 179# from pydantic import BaseModel, Field 180 181 182# class HermesChainConfig(BaseModel): 183 184# base_config: HermesConfig 185 186# dovetail: DovetailModel 187 188# max_steps: int = Field(5, gt=0) 189 190 191# class HermesChain(BaseModel): 192 193# cfg: HermesChainConfig 194 195 196# def run(self): 197 198# from heaven_base.tool_utils.hermes_utils import hermes_step 199 200 201# config = self.cfg.base_config 202 203# for _ in range(self.cfg.max_steps): 204 205# result = hermes_step(hermes_config=config) 206 207# status = result.get("agent_status") 208 209# extracts = getattr(status, "extracted_content", {}) or {} 210 211 212# if all(k in extracts for k in self.cfg.dovetail.expected_outputs): 213 214# return result, extracts 215 216 217# next_args = self.cfg.dovetail.prepare_next_config(extracts) 218 219# config.args_template.update(next_args) 220 221 222# return result, extracts 223# Why this helps: 224 225# • Encapsulation—you can instantiate once with your JSON or Python HermesConfig + DovetailModel and just call .run(). 226 227# • Validation—Pydantic will catch missing or mis-typed fields before you ever start the loop. 228 229# • Reusability—you can register this as a tool or import it wherever you need a multi-step Hermes chain. 230 231# Which pattern makes more sense depends on how strictly you want schema-validation (go Pydantic) versus simplicity (plain class). 232 233 234 235 236 237 238 239# OLD 240 241# BASIC IDEA 242 243# from langchain_core.tools import tool 244 245# @tool 246# def hermes_sequential_chain_tool(initial_data: Dict[str, Any]) -> Dict[str, Any]: 247# """Execute a sequential Hermes chain as a tool""" 248# hermes_sequential_chain = SequentialChain( 249# input_variables=['initial_data'], 250# chains=[ 251# RunnableLambda( 252# lambda x: use_hermes_dict( 253# hermes_config=config1, 254# variable_inputs=x 255# ) 256# ), 257# RunnableLambda( 258# lambda prev_result: use_hermes_dict( 259# hermes_config=config2, 260# variable_inputs=prev_result 261# ) 262# ) 263# ] 264# ) 265 266# # Synchronous invocation for tools 267# result = hermes_sequential_chain.invoke(initial_data) 268# return result 269 270# # If you need async support 271# @tool 272# async def async_hermes_sequential_chain_tool(initial_data: Dict[str, Any]) -> Dict[str, Any]: 273# hermes_sequential_chain = SequentialChain(...) 274# result = await hermes_sequential_chain.ainvoke(initial_data) 275# return result 276 277# every chain function will look like that (basically). it will have a chain type that gets set up then executed and return result, and this function will be a tool, so then agents can call it 278 279 280 # def to_command_data(self, variable_inputs: Optional[Dict[str, Union[Dict[str, Any], List[Any]]]] = None) -> Dict[str, Any]: 281 # """ 282 # Prepare the final command data, handling any templated parameters according to their configuration 283 # Args: 284 # variable_inputs: Dictionary of values to use for templated parameters 285 # """ 286 # try: 287 # with open('/tmp/hermes_debug.log', 'a') as f: 288 # f.write("\nDEBUG to_command_data:") 289 # f.write(f"\nargs_template: {json.dumps(self.args_template, indent=2)}") 290 # f.write(f"\nvariable_inputs: {json.dumps(variable_inputs, indent=2)}") 291 # except Exception as e: 292 # print(f"Debug logging failed: {e}") 293 294 # command_data = self.args_template.copy() 295 296 # if not variable_inputs: 297 # return command_data 298 299 # # Get the template configuration 300 # template_config = command_data["variable_inputs"] 301 302 # # Apply the provided values to their respective parameters 303 # for param_name, param_config in template_config.items(): 304 # if not param_config.get("template"): 305 # continue 306 307 # if param_name not in variable_inputs: 308 # continue 309 310 # # Special handling for goal (or any parameter with variables) 311 # if "variables" in param_config: 312 # command_data[param_name] = command_data[param_name].format(**variable_inputs[param_name]) 313 # else: 314 # # Direct copy for all other templated parameters 315 # command_data[param_name] = variable_inputs[param_name] 316 317 # try: 318 # with open('/tmp/hermes_debug.log', 'a') as f: 319 # f.write(f"\nFinal command_data: {json.dumps(command_data, indent=2)}\n") 320 # except Exception as e: 321 # print(f"Debug logging failed: {e}") 322 323 # return command_data 324 325 # def to_command_data(self, variable_inputs: Optional[Dict[str, Union[Dict[str, Any], List[Any]]]] = None) -> Dict[str, Any]: 326 # """ 327 # Prepare the final command data, handling any templated parameters according to their configuration 328 # Args: 329 # variable_inputs: Dictionary of values to use for templated parameters 330 # """ 331 # try: 332 # with open('/tmp/hermes_debug.log', 'a') as f: 333 # f.write("\nDEBUG to_command_data:") 334 # f.write(f"\nargs_template: {json.dumps(self.args_template, indent=2)}") 335 # f.write(f"\nvariable_inputs: {json.dumps(variable_inputs, indent=2)}") 336 # except Exception as e: 337 # print(f"Debug logging failed: {e}") 338 339 # command_data = self.args_template.copy() 340 341 # if not variable_inputs: 342 # return command_data 343 344 # # Get the template configuration - this is our schema for what can be templated 345 # template_config = command_data["variable_inputs"] 346 347 # # Now apply the provided values to their respective parameters 348 # for param_name, param_config in template_config.items(): 349 # if not param_config.get("template"): 350 # continue 351 352 # # Handle string templates (like goal) 353 # if "variables" in param_config: 354 # # Format the parameter using the provided values 355 # command_data[param_name] = command_data[param_name].format(**variable_inputs[param_name]) 356 357 # # Handle list templates 358 # elif param_config.get("type") == "list": 359 # command_data[param_name] = variable_inputs[param_name] 360 361 # # Handle direct value templates 362 # elif "value" in param_config: 363 # command_data[param_name] = variable_inputs[param_name] 364 365 # try: 366 # with open('/tmp/hermes_debug.log', 'a') as f: 367 # f.write(f"\nFinal command_data: {json.dumps(command_data, indent=2)}\n") 368 # except Exception as e: 369 # print(f"Debug logging failed: {e}") 370 371 # return command_data
11class HermesConfig(BaseFunctionConfig): 12 """ 13 A configuration class for defining a Hermes execution. 14 It must define func_name and args_template as required by BaseFunctionConfig 15 """ 16 func_name: str = "use_hermes" # The function this config is FOR, for metadata purposes, not Callable 17 args_template: Dict[str, Any] = Field( 18 default_factory=lambda: { 19 "goal": "", 20 "iterations": 1, 21 "agent": None, 22 "history_id": None, 23 "return_summary": False, 24 "ai_messages_only": True, 25 "continuation": None, 26 "additional_tools": None, 27 "remove_agents_config_tools": False, 28 "orchestration_preprocess": False, 29 "variable_inputs": {}, 30 "system_prompt_suffix": None 31 } 32 ) 33 34 def to_command_data(self, variable_inputs: Optional[Dict[str, Union[Dict[str, Any], List[Any]]]] = None) -> Dict[str, Any]: 35 # print("\n=== START TO_COMMAND_DATA ===") 36 # print(f"Initial args_template: {json.dumps(self.args_template, indent=2)}") 37 # print(f"Received variable_inputs: {json.dumps(variable_inputs, indent=2)}") 38 39 command_data = self.args_template.copy() 40 #### TESTING THIS 41 # Apply any custom args_template values that differ from defaults 42 # Apply any custom args_template values that differ from defaults 43 for key, value in self.args_template.items(): 44 if value != command_data.get(key): 45 command_data[key] = value 46 47 # print(f"After applying custom args: {json.dumps(command_data, indent=2)}") 48 #### 49 if not variable_inputs: 50 print("No variable_inputs provided, returning default command_data") 51 return command_data 52 53 # Get the template configuration 54 template_config = command_data["variable_inputs"] 55 # print(f"\nTemplate config from args_template: {json.dumps(template_config, indent=2)}") 56 57 # Apply the provided values to their respective parameters 58 for param_name, param_config in template_config.items(): 59 # print(f"\nProcessing parameter: {param_name}") 60 # print(f"Parameter config: {json.dumps(param_config, indent=2)}") 61 62 if not param_config.get("template"): 63 print(f"Skipping {param_name} - not templated") 64 continue 65 66 if param_name not in variable_inputs: 67 print(f"Skipping {param_name} - not in variable_inputs") 68 continue 69 70 # print(f"Value from variable_inputs: {json.dumps(variable_inputs[param_name], indent=2)}") 71 72 # Special handling for goal (or any parameter with variables) 73 if "variables" in param_config: 74 # print(f"Processing as variable template") 75 # print(f"Before: {command_data[param_name]}") 76 command_data[param_name] = command_data[param_name].format(**variable_inputs[param_name]) 77 # print(f"After: {command_data[param_name]}") 78 else: 79 # print(f"Processing as direct value") 80 # print(f"Before: {command_data[param_name]}") 81 command_data[param_name] = variable_inputs[param_name] 82 # print(f"After: {command_data[param_name]}") 83 84 # print(f"\nFinal command_data: {json.dumps(command_data, indent=2)}") 85 # print("=== END TO_COMMAND_DATA ===\n") 86 return command_data
A configuration class for defining a Hermes execution. It must define func_name and args_template as required by BaseFunctionConfig
def
to_command_data( self, variable_inputs: Optional[Dict[str, Union[Dict[str, Any], List[Any]]]] = None) -> Dict[str, Any]:
34 def to_command_data(self, variable_inputs: Optional[Dict[str, Union[Dict[str, Any], List[Any]]]] = None) -> Dict[str, Any]: 35 # print("\n=== START TO_COMMAND_DATA ===") 36 # print(f"Initial args_template: {json.dumps(self.args_template, indent=2)}") 37 # print(f"Received variable_inputs: {json.dumps(variable_inputs, indent=2)}") 38 39 command_data = self.args_template.copy() 40 #### TESTING THIS 41 # Apply any custom args_template values that differ from defaults 42 # Apply any custom args_template values that differ from defaults 43 for key, value in self.args_template.items(): 44 if value != command_data.get(key): 45 command_data[key] = value 46 47 # print(f"After applying custom args: {json.dumps(command_data, indent=2)}") 48 #### 49 if not variable_inputs: 50 print("No variable_inputs provided, returning default command_data") 51 return command_data 52 53 # Get the template configuration 54 template_config = command_data["variable_inputs"] 55 # print(f"\nTemplate config from args_template: {json.dumps(template_config, indent=2)}") 56 57 # Apply the provided values to their respective parameters 58 for param_name, param_config in template_config.items(): 59 # print(f"\nProcessing parameter: {param_name}") 60 # print(f"Parameter config: {json.dumps(param_config, indent=2)}") 61 62 if not param_config.get("template"): 63 print(f"Skipping {param_name} - not templated") 64 continue 65 66 if param_name not in variable_inputs: 67 print(f"Skipping {param_name} - not in variable_inputs") 68 continue 69 70 # print(f"Value from variable_inputs: {json.dumps(variable_inputs[param_name], indent=2)}") 71 72 # Special handling for goal (or any parameter with variables) 73 if "variables" in param_config: 74 # print(f"Processing as variable template") 75 # print(f"Before: {command_data[param_name]}") 76 command_data[param_name] = command_data[param_name].format(**variable_inputs[param_name]) 77 # print(f"After: {command_data[param_name]}") 78 else: 79 # print(f"Processing as direct value") 80 # print(f"Before: {command_data[param_name]}") 81 command_data[param_name] = variable_inputs[param_name] 82 # print(f"After: {command_data[param_name]}") 83 84 # print(f"\nFinal command_data: {json.dumps(command_data, indent=2)}") 85 # print("=== END TO_COMMAND_DATA ===\n") 86 return command_data
Inherited Members
class
HermesConfigInput(pydantic.main.BaseModel):
93class HermesConfigInput(BaseModel): 94 """Represents input mapping for a Hermes configuration""" 95 source_key: str # Key to extract from previous result 96 transform: Optional[Callable] = None # Optional transformation function 97 required: bool = True # Whether this input is mandatory
Represents input mapping for a Hermes configuration
class
DovetailModel(pydantic.main.BaseModel):
99class DovetailModel(BaseModel): 100 """Defines how to chain Hermes configurations""" 101 expected_outputs: List[str] 102 input_map: Dict[str, HermesConfigInput] 103 104 def prepare_next_config(self, previous_result: Dict[str, Any]) -> Dict[str, Any]: 105 """Prepare inputs for the next Hermes configuration""" 106 next_config_inputs = {} 107 108 for config_key, input_spec in self.input_map.items(): 109 # Extract source value 110 source_value = previous_result.get(input_spec.source_key) 111 112 # Apply optional transformation 113 if input_spec.transform: 114 source_value = input_spec.transform(source_value) 115 116 # Check required inputs 117 if input_spec.required and source_value is None: 118 raise ValueError(f"Required input {config_key} not found") 119 120 next_config_inputs[config_key] = source_value 121 122 return next_config_inputs 123 124 125 126 127 128# Here’s a sketch of two approaches: 129 130# A plain Python “runner” class 131 132 133# class HermesChainRunner: 134 135# def __init__(self, base_config: HermesConfig, dovetail: DovetailModel, max_steps: int = 5): 136 137# self.config = base_config 138 139# self.dovetail = dovetail 140 141# self.max_steps = max_steps 142 143# self.last_result = None 144 145# self.last_extracts = {} 146 147 148# def run(self): 149 150# from heaven_base.tool_utils.hermes_utils import hermes_step 151 152 153# for i in range(self.max_steps): 154 155# result = hermes_step(hermes_config=self.config) 156 157# status = result.get("agent_status") 158 159# extracts = getattr(status, "extracted_content", {}) or {} 160 161# self.last_result, self.last_extracts = result, extracts 162 163 164# # stop when all expected keys are present 165 166# if all(k in extracts for k in self.dovetail.expected_outputs): 167 168# break 169 170 171# next_args = self.dovetail.prepare_next_config(extracts) 172 173# self.config.args_template.update(next_args) 174 175 176# return self.last_result, self.last_extracts 177# A Pydantic-driven orchestrator model 178 179 180# from pydantic import BaseModel, Field 181 182 183# class HermesChainConfig(BaseModel): 184 185# base_config: HermesConfig 186 187# dovetail: DovetailModel 188 189# max_steps: int = Field(5, gt=0) 190 191 192# class HermesChain(BaseModel): 193 194# cfg: HermesChainConfig 195 196 197# def run(self): 198 199# from heaven_base.tool_utils.hermes_utils import hermes_step 200 201 202# config = self.cfg.base_config 203 204# for _ in range(self.cfg.max_steps): 205 206# result = hermes_step(hermes_config=config) 207 208# status = result.get("agent_status") 209 210# extracts = getattr(status, "extracted_content", {}) or {} 211 212 213# if all(k in extracts for k in self.cfg.dovetail.expected_outputs): 214 215# return result, extracts 216 217 218# next_args = self.cfg.dovetail.prepare_next_config(extracts) 219 220# config.args_template.update(next_args) 221 222 223# return result, extracts 224# Why this helps: 225 226# • Encapsulation—you can instantiate once with your JSON or Python HermesConfig + DovetailModel and just call .run(). 227 228# • Validation—Pydantic will catch missing or mis-typed fields before you ever start the loop. 229 230# • Reusability—you can register this as a tool or import it wherever you need a multi-step Hermes chain. 231 232# Which pattern makes more sense depends on how strictly you want schema-validation (go Pydantic) versus simplicity (plain class). 233 234 235 236 237 238 239 240# OLD 241 242# BASIC IDEA 243 244# from langchain_core.tools import tool 245 246# @tool 247# def hermes_sequential_chain_tool(initial_data: Dict[str, Any]) -> Dict[str, Any]: 248# """Execute a sequential Hermes chain as a tool""" 249# hermes_sequential_chain = SequentialChain( 250# input_variables=['initial_data'], 251# chains=[ 252# RunnableLambda( 253# lambda x: use_hermes_dict( 254# hermes_config=config1, 255# variable_inputs=x 256# ) 257# ), 258# RunnableLambda( 259# lambda prev_result: use_hermes_dict( 260# hermes_config=config2, 261# variable_inputs=prev_result 262# ) 263# ) 264# ] 265# ) 266 267# # Synchronous invocation for tools 268# result = hermes_sequential_chain.invoke(initial_data) 269# return result 270 271# # If you need async support 272# @tool 273# async def async_hermes_sequential_chain_tool(initial_data: Dict[str, Any]) -> Dict[str, Any]: 274# hermes_sequential_chain = SequentialChain(...) 275# result = await hermes_sequential_chain.ainvoke(initial_data) 276# return result 277 278# every chain function will look like that (basically). it will have a chain type that gets set up then executed and return result, and this function will be a tool, so then agents can call it 279 280 281 # def to_command_data(self, variable_inputs: Optional[Dict[str, Union[Dict[str, Any], List[Any]]]] = None) -> Dict[str, Any]: 282 # """ 283 # Prepare the final command data, handling any templated parameters according to their configuration 284 # Args: 285 # variable_inputs: Dictionary of values to use for templated parameters 286 # """ 287 # try: 288 # with open('/tmp/hermes_debug.log', 'a') as f: 289 # f.write("\nDEBUG to_command_data:") 290 # f.write(f"\nargs_template: {json.dumps(self.args_template, indent=2)}") 291 # f.write(f"\nvariable_inputs: {json.dumps(variable_inputs, indent=2)}") 292 # except Exception as e: 293 # print(f"Debug logging failed: {e}") 294 295 # command_data = self.args_template.copy() 296 297 # if not variable_inputs: 298 # return command_data 299 300 # # Get the template configuration 301 # template_config = command_data["variable_inputs"] 302 303 # # Apply the provided values to their respective parameters 304 # for param_name, param_config in template_config.items(): 305 # if not param_config.get("template"): 306 # continue 307 308 # if param_name not in variable_inputs: 309 # continue 310 311 # # Special handling for goal (or any parameter with variables) 312 # if "variables" in param_config: 313 # command_data[param_name] = command_data[param_name].format(**variable_inputs[param_name]) 314 # else: 315 # # Direct copy for all other templated parameters 316 # command_data[param_name] = variable_inputs[param_name] 317 318 # try: 319 # with open('/tmp/hermes_debug.log', 'a') as f: 320 # f.write(f"\nFinal command_data: {json.dumps(command_data, indent=2)}\n") 321 # except Exception as e: 322 # print(f"Debug logging failed: {e}") 323 324 # return command_data 325 326 # def to_command_data(self, variable_inputs: Optional[Dict[str, Union[Dict[str, Any], List[Any]]]] = None) -> Dict[str, Any]: 327 # """ 328 # Prepare the final command data, handling any templated parameters according to their configuration 329 # Args: 330 # variable_inputs: Dictionary of values to use for templated parameters 331 # """ 332 # try: 333 # with open('/tmp/hermes_debug.log', 'a') as f: 334 # f.write("\nDEBUG to_command_data:") 335 # f.write(f"\nargs_template: {json.dumps(self.args_template, indent=2)}") 336 # f.write(f"\nvariable_inputs: {json.dumps(variable_inputs, indent=2)}") 337 # except Exception as e: 338 # print(f"Debug logging failed: {e}") 339 340 # command_data = self.args_template.copy() 341 342 # if not variable_inputs: 343 # return command_data 344 345 # # Get the template configuration - this is our schema for what can be templated 346 # template_config = command_data["variable_inputs"] 347 348 # # Now apply the provided values to their respective parameters 349 # for param_name, param_config in template_config.items(): 350 # if not param_config.get("template"): 351 # continue 352 353 # # Handle string templates (like goal) 354 # if "variables" in param_config: 355 # # Format the parameter using the provided values 356 # command_data[param_name] = command_data[param_name].format(**variable_inputs[param_name]) 357 358 # # Handle list templates 359 # elif param_config.get("type") == "list": 360 # command_data[param_name] = variable_inputs[param_name] 361 362 # # Handle direct value templates 363 # elif "value" in param_config: 364 # command_data[param_name] = variable_inputs[param_name] 365 366 # try: 367 # with open('/tmp/hermes_debug.log', 'a') as f: 368 # f.write(f"\nFinal command_data: {json.dumps(command_data, indent=2)}\n") 369 # except Exception as e: 370 # print(f"Debug logging failed: {e}") 371 372 # return command_data
Defines how to chain Hermes configurations
def
prepare_next_config(self, previous_result: Dict[str, Any]) -> Dict[str, Any]:
104 def prepare_next_config(self, previous_result: Dict[str, Any]) -> Dict[str, Any]: 105 """Prepare inputs for the next Hermes configuration""" 106 next_config_inputs = {} 107 108 for config_key, input_spec in self.input_map.items(): 109 # Extract source value 110 source_value = previous_result.get(input_spec.source_key) 111 112 # Apply optional transformation 113 if input_spec.transform: 114 source_value = input_spec.transform(source_value) 115 116 # Check required inputs 117 if input_spec.required and source_value is None: 118 raise ValueError(f"Required input {config_key} not found") 119 120 next_config_inputs[config_key] = source_value 121 122 return next_config_inputs
Prepare inputs for the next Hermes configuration