def __init__(
 self,
 ctx: Optional[Context] = None,
 config: Optional[Dict[str, Any]] = None,
 name: Optional[str] = None,
 model: Optional[str] = None,
 description: Optional[str] = None,
 system_prompt: Optional[str] = None,
 tools: Optional[List[str]] = None,
):
 """Initialize the agent."""
 try:
 if config:
 # Initialize from config
 self.name = config["name"]
 self.model = config["model"]
 self.description = config["description"]
 self.system_prompt = config["system_prompt"]
 tool_names = config.get("tools", [])
 else:
 # Initialize from explicitly passed arguments
 if not all([name, model, description, system_prompt]):
 raise ValueError(
 "If config is not provided, all of 'name', 'model', 'description', and 'system_prompt' must be specified."
 )
 self.name = name
 self.model = model
 self.description = description
 self.system_prompt = system_prompt
 tool_names = tools or []
 if len(tool_names) > 0:
 self.tools = tools_registry.load_tools(tool_names)
 else:
 self.tools = []
 self.context = ctx
 # Replace with actual model manager
 models_manager = get_models_manager()
 self.llm = models_manager.get(self.model)
 self._initialize_tools()
 except Exception as e:
 raise e
def _initialize_tools(self):
 """Inject context into tools."""
 if self.tools:
 _inject_ctx_into_tools(self.context, self.tools)
def _get_tool_descriptions(self) -> Dict[str, str]:
 """Generate descriptions of the tools."""
 return {tool.name: tool.description for tool in self.tools}
def _get_prompt_variables(self) -> Dict[str, Any]:
 """Get prompt variables."""
 _prompt_variables = {
 "agent_description": self.description,
 "tools": self._get_tool_descriptions(),
 "datetime": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
 }
 if self.context.get("prompt_variables"):
 _prompt_variables.update(self.context.get("prompt_variables"))
 return _prompt_variables
@staticmethod
def _format_system_prompt(
 prompt_template: str, prompt_variables: Dict[str, Any]
) -> str:
 """Format the system prompt similar to Langfuse's compile method."""
 placeholders = re.findall(r"\{\{(\w+)\}\}", prompt_template)
 missing_keys = [key for key in placeholders if key not in prompt_variables]
 if missing_keys:
 raise KeyError(f"Missing variables in prompt_variables: {missing_keys}")
 for key in placeholders:
 placeholder = f"{{{{{key}}}}}"
 prompt_template = prompt_template.replace(
 placeholder, str(prompt_variables[key])
 )
 return prompt_template
def run(
 self, messages: List[Dict[str, Any]], stream: bool = False
) -> Dict[str, Any] | Generator[Dict[str, Any], None, None]:
 """Run the agent with the provided message and interaction history."""
 system_prompt = self._format_system_prompt(
 self.system_prompt, self._get_prompt_variables()
 )
 try:
 return self.llm.invoke(
 system_prompt=system_prompt,
 messages=messages,
 ctx=self.context,
 tools=self.tools if self.tools else None,
 stream=stream,
 )
 except Exception as e:
 raise e
def to_dict(self) -> Dict[str, Any]:
 """Convert the agent to a dictionary."""
 return {
 "name": self.name,
 "description": self.description,
 "tools": [tool.to_dict() for tool in self.tools],
 }