refactor: improve agent configuration and routing logic in LLM and tool agents

This commit is contained in:
2025-06-30 18:12:05 +00:00
parent 315c583ed9
commit 9049478bdc
20 changed files with 1059 additions and 269 deletions
+47 -59
View File
@@ -63,35 +63,27 @@ class LLMAgent(AgentWithMemory):
"""
super().__init__(name, config, template_renderer)
# The actual agent settings may be nested inside a "config" dictionary.
config_data = config.get("config", config)
# Extract configuration
self.provider = config.get("provider", "openai").lower()
self.model = config.get("model", "gpt-3.5-turbo")
self.api_key = config.get(
self.provider = config_data.get("provider", "openai").lower()
self.model = config_data.get("model", "gpt-3.5-turbo")
self.api_key = config_data.get(
"api_key", os.environ.get(f"{self.provider.upper()}_API_KEY")
)
# Always start with the default system message.
self.system_message = DEFAULT_SYSTEM_MESSAGE
# If NO prompt template is configured we allow an override
# via ``system_message`` in the agent config. When a prompt
# template is supplied we keep the default to guarantee the
# first line of every formatted prompt is identical (tests rely
# on this behaviour).
if not config.get("prompt_template"):
self.system_message = config.get("system_message", DEFAULT_SYSTEM_MESSAGE)
self.temperature = float(config.get("temperature", 0.7))
self.max_tokens = int(config.get("max_tokens", 1000))
# Always prioritise the configured system message, falling back to the default.
self.system_message = config_data.get("system_message", DEFAULT_SYSTEM_MESSAGE)
self.temperature = float(config_data.get("temperature", 0.7))
self.max_tokens = int(config_data.get("max_tokens", 1000))
# Check that we have an API key
if not self.api_key:
raise AgentCreationError(f"No API key provided for {self.provider} LLM")
# Initialize memory with conversation history
# Initialize memory for conversation history (system message is injected
# fresh on every API call to avoid stale configuration issues).
self.memory["messages"] = []
if self.system_message:
self.memory["messages"].append(
{"role": "system", "content": self.system_message}
)
async def process(
self, message: str, context: Optional[Dict[str, Any]] = None
@@ -124,16 +116,34 @@ class LLMAgent(AgentWithMemory):
{"role": "user", "content": formatted_message}
)
# Build the messages list for this specific API call, ensuring the
# current system message is the very first element.
api_messages: List[Dict[str, str]] = []
if self.system_message:
api_messages.append({"role": "system", "content": self.system_message})
api_messages.extend(self.memory["messages"])
prompt_for_logging = "\n".join(
f"[{m.get('role')}] {m.get('content')}" for m in api_messages
)
logger.debug(
f"Agent '{self.name}' sending prompt:\n"
f"---PROMPT---\n{prompt_for_logging}\n---END PROMPT---"
)
# Generate a response
if self.provider == "openai":
response = await self._generate_openai_response()
response = await self._generate_openai_response(api_messages)
elif self.provider == "anthropic":
response = await self._generate_anthropic_response()
response = await self._generate_anthropic_response(api_messages)
else:
raise ExecutionError(f"Unsupported LLM provider: {self.provider}")
# Process the response
processed_response = self.process_response(response)
logger.debug(
f"Agent '{self.name}' received response:\n---RESPONSE---\n{processed_response}\n---END RESPONSE---"
)
# Add the response to the conversation history
self.memory["messages"].append(
@@ -144,7 +154,7 @@ class LLMAgent(AgentWithMemory):
except Exception as e:
raise ExecutionError(f"Failed to process message: {str(e)}")
async def _generate_openai_response(self) -> str:
async def _generate_openai_response(self, messages: List[Dict[str, str]]) -> str:
"""
Generate a response using the OpenAI API.
@@ -163,7 +173,7 @@ class LLMAgent(AgentWithMemory):
}
data = {
"model": self.model,
"messages": self.memory["messages"],
"messages": messages,
"temperature": self.temperature,
"max_tokens": self.max_tokens,
}
@@ -187,7 +197,7 @@ class LLMAgent(AgentWithMemory):
except Exception as e:
raise ExecutionError(f"Failed to generate OpenAI response: {str(e)}")
async def _generate_anthropic_response(self) -> str:
async def _generate_anthropic_response(self, messages: List[Dict[str, str]]) -> str:
"""
Generate a response using the Anthropic API.
@@ -206,29 +216,16 @@ class LLMAgent(AgentWithMemory):
"anthropic-version": "2023-06-01",
}
# Convert messages to Anthropic format
messages = []
for message in self.memory["messages"]:
role = message["role"]
content = message["content"]
if role == "system":
# Anthropic doesn't have system messages, so we'll add it to the first user message
continue
elif role == "user":
messages.append({"role": "user", "content": content})
elif role == "assistant":
messages.append({"role": "assistant", "content": content})
# Add system message to the first user message if it exists
if messages and messages[0]["role"] == "user" and self.system_message:
messages[0][
"content"
] = f"{self.system_message}\n\n{messages[0]['content']}"
# Separate system message from conversation history for Anthropic API
system_prompt = next(
(m["content"] for m in messages if m.get("role") == "system"), ""
)
conversation_messages = [m for m in messages if m.get("role") != "system"]
data = {
"model": self.model,
"messages": messages,
"system": system_prompt,
"messages": conversation_messages,
"max_tokens": self.max_tokens,
"temperature": self.temperature,
}
@@ -303,12 +300,8 @@ class LLMAgent(AgentWithMemory):
system_msg = self.system_message or ""
if not template_spec:
# -------------------- legacy / default formatting --------------------
parts: List[str] = []
if system_msg:
parts.append(f"System: {system_msg}")
parts.append(f"User: {message}")
return "\n\n".join(parts)
# No template configured; return the raw user message unchanged.
return message
# context supplied to the template
render_context: Dict[str, Any] = {
@@ -330,16 +323,11 @@ class LLMAgent(AgentWithMemory):
template_spec, render_context
).strip()
# guarantee “System: …” is the first line, as tests rely on this
return f"System: {system_msg}\n\n{rendered}".rstrip()
return rendered.rstrip()
except Exception:
# any error → fall back to default formatting
logger.exception("Failed to render prompt template, using default.")
parts = []
if system_msg:
parts.append(f"System: {system_msg}")
parts.append(f"User: {message}")
return "\n\n".join(parts)
logger.exception("Failed to render prompt template, using raw message.")
return message
def process_response(self, response: str) -> str:
"""
+86 -11
View File
@@ -9,12 +9,15 @@ import importlib
import inspect
import json
import logging
import os
import re
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
import aiohttp
from cleveragents.agents.base import Agent
from cleveragents.core.exceptions import AgentCreationError
from cleveragents.core.exceptions import ExecutionError
@@ -114,13 +117,15 @@ class CalculatorTool(Tool):
class WebSearchTool(Tool):
"""
A tool for searching the web.
A tool for searching the web using Google Custom Search.
This tool performs web searches and returns the results.
This tool performs web searches and returns the results. It requires a Google
API Key and a Programmable Search Engine ID (cx).
Attributes:
api_key (str): The API key for the search engine.
search_engine (str): The search engine to use.
search_engine (str): The search engine to use (currently only 'google').
cx (str): The Programmable Search Engine ID.
"""
def __init__(self, config: Dict[str, Any]):
@@ -129,37 +134,105 @@ class WebSearchTool(Tool):
Args:
config: Configuration for the tool.
Raises:
AgentCreationError: If required configuration is missing.
"""
super().__init__(
name="web_search",
description="Searches the web for information. Input should be a search query.",
description="Searches the web for information using Google Custom Search. Input should be a search query.",
config=config,
)
self.api_key = config.get("api_key")
self.search_engine = config.get("search_engine", "google")
if self.search_engine != "google":
raise AgentCreationError(
f"Search engine '{self.search_engine}' is not supported. Only 'google' is available."
)
# API key can be in config or from environment variable GOOGLE_API_KEY
self.api_key = config.get("api_key") or os.environ.get("GOOGLE_API_KEY")
# CX (Custom Search Engine ID) can be in config or from environment variable GOOGLE_CX
self.cx = config.get("cx") or os.environ.get("GOOGLE_CX")
if not self.api_key:
raise AgentCreationError("No API key provided for web search tool")
raise AgentCreationError(
"Google API Key not found. Please provide it in the tool config "
"or set the GOOGLE_API_KEY environment variable."
)
if not self.cx:
raise AgentCreationError(
"Google Custom Search Engine ID (cx) not found. Please provide it "
"in the tool config or set the GOOGLE_CX environment variable."
)
async def execute(
self, input_data: str, context: Optional[Dict[str, Any]] = None
) -> str:
"""
Execute the web search tool.
Execute the web search tool by calling the Google Custom Search API.
Args:
input_data: A search query.
context: Additional context (not used).
Returns:
The search results.
The title and snippet of the first search result, or an error message.
Raises:
ExecutionError: If the search fails.
"""
# Placeholder implementation; in real use, integrate with an actual web search API.
return f"Web search results for '{input_data}'"
if self.search_engine != "google":
# This is where logic for other search engines would go.
raise ExecutionError(
f"Search engine '{self.search_engine}' is not supported."
)
# mypy: self.api_key and self.cx are Optional[str] at the attribute level.
# Runtime checks in __init__ guarantee they are populated, but we assert
# here to make that explicit for static analysis.
assert self.api_key is not None, "Google API Key is not configured."
assert (
self.cx is not None
), "Google Custom Search Engine ID (cx) is not configured."
url = "https://www.googleapis.com/customsearch/v1"
params = {"key": self.api_key, "cx": self.cx, "q": input_data}
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
if response.status != 200:
error_text = await response.text()
logger.error(
f"Google Search API returned status {response.status}: {error_text}"
)
raise ExecutionError(
f"Google Search API error: Received status {response.status}. "
"Check your API key and CX."
)
data = await response.json()
items = data.get("items", [])
if items:
first_result = items[0]
title = first_result.get("title", "No title")
snippet = first_result.get("snippet", "No snippet available.")
return f"{title}\n{snippet}"
return "No web search results found."
except aiohttp.ClientError as e:
logger.error(f"Failed to connect to Google Search API: {e}")
raise ExecutionError(f"Failed to connect to Google Search API: {str(e)}")
except Exception as e:
logger.error(
f"An unexpected error occurred during web search: {e}", exc_info=True
)
raise ExecutionError(
f"An unexpected error occurred during web search: {str(e)}"
)
class ToolAgent(Agent):
@@ -228,7 +301,9 @@ class ToolAgent(Agent):
self.tools[tool_name] = CalculatorTool()
logger.debug(f"Initialized calculator tool for agent '{self.name}'")
elif tool_name == "web_search":
self.tools[tool_name] = WebSearchTool(config_data)
# Extract the nested configuration block if present
tool_specific_config = config_data.get("config", config_data)
self.tools[tool_name] = WebSearchTool(tool_specific_config)
logger.debug(f"Initialized web search tool for agent '{self.name}'")
else:
# Try to load a custom tool
+40 -7
View File
@@ -7,6 +7,7 @@ in both single-shot and interactive modes.
"""
import logging
import sys
from pathlib import Path
from typing import Any
from typing import Dict
@@ -53,9 +54,13 @@ class CleverAgentsApp:
CleverAgentsException: If configuration loading fails.
"""
# Set up logging
self.logger = logging.getLogger("cleveragents")
log_level = logging.DEBUG if verbose else logging.INFO
self.logger.setLevel(log_level)
logging.basicConfig(
level=log_level,
format="[%(name)s] %(message)s",
stream=sys.stderr,
)
self.logger = logging.getLogger(__name__)
# Initialize components
self.config_manager = ConfigurationManager()
@@ -89,7 +94,12 @@ class CleverAgentsApp:
self.config_manager.validate()
# Initialize template renderer
template_engine_name = self.config_manager.get("template_engine", "simple")
# Prefer the nested cleveragents.template_engine path, but
# gracefully fall back to the legacy top-level key if needed.
template_engine_name = self.config_manager.get(
"cleveragents.template_engine",
self.config_manager.get("template_engine", "simple"),
)
try:
template_engine = TemplateEngine[template_engine_name.upper()]
except KeyError:
@@ -100,6 +110,29 @@ class CleverAgentsApp:
template_renderer = TemplateRenderer(template_engine)
# Detect silent fallback to SIMPLE engine (e.g., when `jinja2`
# is requested but not installed). Abort early with a clear
# message so users can fix their environment instead of getting
# hard-to-debug template errors later in execution.
requested_engine = template_engine
loaded_engine = getattr(
template_renderer,
"engine_type", # Preferred attribute
getattr( # Fallback to legacy/internal attr
template_renderer, "engine", TemplateEngine.SIMPLE
),
)
if (
requested_engine is not TemplateEngine.SIMPLE
and loaded_engine is TemplateEngine.SIMPLE
):
raise CleverAgentsException(
f"Requested template engine '{requested_engine.name}' but it "
"failed to initialise, so the system fell back to SIMPLE. "
"Please install the required dependency (e.g. `pip install jinja2`) "
"or change `template_engine` to 'simple' in your configuration."
)
# Load templates from configuration
templates = self.config_manager.get("prompts", {})
for name, template_def in templates.items():
@@ -164,11 +197,11 @@ class CleverAgentsApp:
self.logger.debug(f"Created agent: {agent_name}")
# Create router
router_name = self.config_manager.get("routing.name", "main_router")
router_name = self.config_manager.get("router.name", "main_router")
router = Router(router_name, agents, self.agent_factory.template_renderer)
# Add routes
routes = self.config_manager.get("routing.flows", [])
routes = self.config_manager.get("router.routes", [])
router.add_routes(routes)
# Get global context
@@ -224,11 +257,11 @@ class CleverAgentsApp:
self.logger.debug(f"Created agent: {agent_name}")
# Create router
router_name = self.config_manager.get("routing.name", "main_router")
router_name = self.config_manager.get("router.name", "main_router")
router = Router(router_name, agents, self.agent_factory.template_renderer)
# Add routes
routes = self.config_manager.get("routing.flows", [])
routes = self.config_manager.get("router.routes", [])
router.add_routes(routes)
# Create interactive session
+9 -10
View File
@@ -34,16 +34,11 @@ class AgentNetwork:
if config_files:
self.config_manager.load_files(config_files)
self.config_manager.validate()
# Initialize template renderer with SIMPLE engine
template_renderer = TemplateRenderer(TemplateEngine.SIMPLE)
self.agent_factory = AgentFactory(
self.config_manager.to_dict(), template_renderer
)
agents_conf = self.config_manager.get("agents", {})
agents = {}
for name in agents_conf:
agents[name] = self.agent_factory.create_agent(name)
self.router = Router("main_router", agents, template_renderer)
# This network class is not fully utilized by the application yet.
# The core logic is currently in CleverAgentsApp.
self.agent_factory: Optional[AgentFactory] = None
self.router: Optional[Router] = None
async def process(
self, message: str, context: Optional[Dict[Any, Any]] = None
@@ -58,6 +53,10 @@ class AgentNetwork:
Returns:
str: The final output message.
"""
# Ensure router is initialized (helps mypy understand it's not None)
assert (
self.router is not None
), "Router not initialized. Load configuration first."
return await self.router.process_message(message, context)
def get_router(self):
+46 -17
View File
@@ -21,6 +21,27 @@ class ConditionEvaluator:
Evaluates routing conditions based on context.
"""
@staticmethod
def _evaluate_keywords(condition: Dict[str, Any], message: str) -> bool:
"""Evaluate a 'keywords' type condition."""
keywords_val = condition.get("keywords")
keywords: List[str] = []
if isinstance(keywords_val, str):
keywords = [keywords_val]
elif isinstance(keywords_val, list):
keywords = [str(k) for k in keywords_val]
# If no keywords specified, the condition cannot be satisfied
if not keywords:
return False
match_all = condition.get("match_all", False)
message_lower = message.lower()
if match_all:
return all(kw.lower() in message_lower for kw in keywords)
return any(kw.lower() in message_lower for kw in keywords)
@staticmethod
def evaluate(
condition: Union[str, Dict[str, Any]],
@@ -43,14 +64,36 @@ class ConditionEvaluator:
"""
if context is None:
context = {}
else:
# Work on a shallow copy to avoid mutating the caller's context.
context = context.copy()
# Add message to context for use in conditions
context["message"] = message
try:
if isinstance(condition, str):
# Simple string condition - check if it's in the message
return condition.lower() in message.lower()
# First, attempt to treat the string as a valid Python boolean
# expression (e.g. "'SEARCH' in message"). If that fails,
# gracefully fall back to a *case-insensitive substring* search
# so that simple strings like "hello" are interpreted as
# "contains 'hello'".
local_scope: Dict[str, Any] = {"message": message, "context": context}
# Strategy 1: `eval` single-line expressions that return bool
try:
return bool(eval(condition, {"__builtins__": {}}, local_scope))
except Exception:
pass # Ignore and try the next strategy
# Strategy 2: `exec` multi-line code that sets `result`
try:
local_scope["result"] = False
exec(condition, {"__builtins__": {}}, local_scope)
return bool(local_scope.get("result", False))
except Exception:
# Strategy 3: simple substring search (default behaviour)
return condition.lower() in message.lower()
elif isinstance(condition, dict):
condition_type = condition.get("type", "simple")
@@ -60,21 +103,7 @@ class ConditionEvaluator:
return bool(re.search(pattern, message, re.IGNORECASE))
elif condition_type == "keywords":
# Check for presence of keywords
keywords = condition.get("keywords", [])
if isinstance(keywords, str):
keywords = [keywords]
match_all = condition.get("match_all", False)
if match_all:
return all(
keyword.lower() in message.lower() for keyword in keywords
)
else:
return any(
keyword.lower() in message.lower() for keyword in keywords
)
return ConditionEvaluator._evaluate_keywords(condition, message)
elif condition_type == "python":
# Execute Python code
+141 -66
View File
@@ -7,6 +7,7 @@ based on conditions and transformations.
import asyncio
import logging
import pprint
from typing import Any
from typing import Dict
from typing import List
@@ -62,6 +63,17 @@ class Router:
if not isinstance(route, dict):
raise RoutingError("Route must be a dictionary")
# Standardize keys: 'source'/'destination' are aliases for 'from'/'to'.
# Canonical keys take precedence.
if "from" not in route and "source" in route:
route["from"] = route["source"]
if "to" not in route and "destination" in route:
route["to"] = route["destination"]
# Clean up alias keys to avoid confusion later.
route.pop("source", None)
route.pop("destination", None)
required_keys = ["from", "to"]
for key in required_keys:
if key not in route:
@@ -98,7 +110,8 @@ class Router:
context: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""
Route a message according to the routing rules.
Route a message according to the routing rules, with detailed
debug logging and corrected template-rendering context.
Args:
message: The message to route.
@@ -106,74 +119,100 @@ class Router:
context: Additional context for routing.
Returns:
Dictionary with the routing result, including the destination and transformed message.
A dict containing the destination agent name and the (possibly
transformed) message.
Raises:
RoutingError: If routing fails.
RoutingError: If routing or transformation fails.
"""
if context is None:
context = {}
# Find matching routes
matching_routes = []
for route in self.routes:
if route["from"] == source:
# Check condition if present
if "condition" in route:
try:
condition_met = self.condition_evaluator.evaluate(
route["condition"], message, context
)
if not condition_met:
continue
except Exception as e:
raise RoutingError(f"Error evaluating condition: {str(e)}")
logger.debug(f"Routing message from source='{source}': {repr(message)}")
matching_routes.append(route)
# Find all matching routes
matching_routes: List[Dict[str, Any]] = []
for idx, route in enumerate(self.routes):
logger.debug(
f"[Route {idx + 1}/{len(self.routes)}] Checking route: "
f"from='{route.get('from')}', to='{route.get('to')}'"
)
if route.get("from") != source:
logger.debug(" - Source does not match.")
continue
logger.debug(" - Source matches.")
# Evaluate condition, if any
if "condition" in route:
condition = route["condition"]
logger.debug(f" - Evaluating condition: {repr(condition)}")
try:
if not self.condition_evaluator.evaluate(
condition, message, context
):
logger.debug(" -> Condition FAILED.")
continue
logger.debug(" -> Condition PASSED.")
except Exception as e:
raise RoutingError(f"Error evaluating condition: {str(e)}")
else:
logger.debug(" - No condition specified.")
matching_routes.append(route)
# No routes? Decide what to do.
if not matching_routes:
logger.debug(f"No matching routes for source '{source}'.")
if source == "input":
if len(self.agents) == 1:
agent_name = list(self.agents.keys())[0]
logger.info(
"No explicit 'input' route found; "
f"defaulting to single agent: {agent_name}"
default_agent = list(self.agents.keys())[0]
logger.debug(
f"Single agent configured; defaulting destination to '{default_agent}'."
)
return {"destination": agent_name, "message": message}
return {"destination": default_agent, "message": message}
raise RoutingError("No route found for input message")
else:
# If no routes found and not from input, route to output
return {"destination": "output", "message": message}
logger.debug("Defaulting destination to 'output'.")
return {"destination": "output", "message": message}
# Process the first matching route (for now)
# Select the first matching route (FIFO priority)
route = matching_routes[0]
destination = route["to"]
logger.debug(f"Selected destination: '{destination}'")
# Apply transformation if present
# Apply transformation, if provided
transformed_message = message
if "transform" in route:
transform = route["transform"]
if isinstance(transform, str):
# This is a string template, not a template name
try:
logger.debug(f"Applying transform: {repr(transform)}")
# Provide both the original message and the full context under
# a dedicated key to avoid key collisions inside templates.
render_context = {"message": message, "context": context}
logger.debug(f" - Context before transform: {render_context}")
logger.debug(" - Full context for transform:\n%s", pprint.pformat(context))
try:
if isinstance(transform, str):
transformed_message = self.template_renderer.render_string(
transform, {"message": "{message}", **(context or {})}
transform, render_context
)
except Exception as e:
raise RoutingError(
f"Error applying string template transformation: {str(e)}"
)
elif isinstance(transform, dict) and "template" in transform:
# Inline template
try:
elif isinstance(transform, dict) and "template" in transform:
transformed_message = self.template_renderer.render_string(
transform["template"],
{"message": "{message}", **(context or {})},
transform["template"], render_context
)
except Exception as e:
else:
raise RoutingError(
f"Error applying inline template transformation: {str(e)}"
f"Unsupported transform format: {type(transform)}"
)
except Exception as e:
raise RoutingError(f"Error applying template transformation: {str(e)}")
logger.debug(f" - Context after transform: {render_context}")
logger.debug(f" - Transform result: {repr(transformed_message)}")
else:
logger.debug("No transform defined for this route.")
return {"destination": destination, "message": transformed_message}
@@ -181,52 +220,88 @@ class Router:
self, message: str, context: Optional[Dict[str, Any]] = None
) -> str:
"""
Process a message through the routing network.
Process a message end-to-end through the router, logging each step.
This method routes a message through the network of agents according to
the routing rules, until it reaches the 'output' destination.
The method repeatedly routes the message and invokes the appropriate
agent until the destination is 'output' or the maximum number of
hops is reached.
Args:
message: The initial message.
context: Additional context for processing.
message: The initial user message.
context: Optional shared context dictionary.
Returns:
The final output message.
The final message that should be returned to the user.
Raises:
RoutingError: If processing fails.
RoutingError: If processing fails or exceeds max steps.
"""
# Work on a *copy* of the provided context to avoid mutating the
# caller's dictionary through side-effects inside the router.
if context is None:
context = {}
else:
context = context.copy()
# Preserve the user's original input for template transforms.
if "initial_message" not in context:
context["initial_message"] = message
if "history" not in context:
context["history"] = []
current_source = "input"
current_message = message
max_steps = 100 # Prevent infinite loops
max_steps = 100 # Safety guard against infinite loops
for _ in range(max_steps):
# Route the message
result = await self.route_message(current_message, current_source, context)
# Ensure result is a dictionary
if isinstance(result, dict):
destination = result.get("destination", "output")
transformed_message = result.get("message", current_message)
else:
# If result is not a dictionary, use defaults
destination = "output"
transformed_message = current_message if result is None else str(result)
logger.debug("=== Begin router processing ===")
logger.debug(f"Initial message: {repr(current_message)}")
# If destination is 'output', we're done
for step in range(max_steps):
logger.debug(f"--- Step {step + 1} ---")
route_result = await self.route_message(
current_message, current_source, context
)
destination = route_result.get("destination", "output")
transformed_message = route_result.get("message", current_message)
logger.debug(
f"Routed to '{destination}' with message: {repr(transformed_message)}"
)
# Reached the terminal node
if destination == "output":
logger.debug("Reached 'output'. Processing complete.")
return transformed_message
# Process the message with the destination agent
# Validate destination agent exists
if destination not in self.agents:
raise RoutingError(
f"Destination agent '{destination}' not found in router."
)
agent = self.agents[destination]
logger.debug(f"Invoking agent '{destination}'")
try:
current_message = await agent.process(transformed_message, context)
current_source = destination
result_message = await agent.process(transformed_message, context)
logger.debug(f"Agent '{destination}' returned: {repr(result_message)}")
except Exception as e:
raise RoutingError(
f"Error processing message with agent '{destination}': {str(e)}"
)
# Append the completed step to the history
context["history"].append(
{
"source": current_source,
"destination": destination,
"message": result_message,
}
)
# Prepare for next hop
current_source = destination
current_message = result_message
# If we exit the loop, something went wrong
raise RoutingError("Maximum routing steps exceeded")
+42 -3
View File
@@ -5,6 +5,7 @@ This module defines the TemplateRenderer class, which is responsible for renderi
with provided context data. It supports multiple template engines including Jinja2 and simple string formatting.
"""
import logging
import re
from enum import Enum
from typing import Any
@@ -18,6 +19,8 @@ from typing import Union
from cleveragents.core.exceptions import TemplateError
logger = logging.getLogger(__name__)
class Jinja2Template(Protocol):
def render(self, **kwargs: Any) -> str: ...
@@ -113,10 +116,46 @@ class TemplateRenderer:
@staticmethod
def _render_simple_with_jinja_like(template: str, context: Dict[str, Any]) -> str:
# replace {{ ... }} placeholders (supports dotted paths)
def _sub(match: re.Match[str]) -> str: # pragma: no cover
"""
Render the template string replacing {{ ... }} placeholders using a
very small Jinja-like subset.
Behavioural tweak:
1. Try to resolve the given dotted path exactly as written.
2. If that returns an empty value *and* the path starts with
``context.``, retry the resolution without that prefix.
This allows templates written for a nested ``context`` object
(e.g. ``{{ context.initial_message }}``) to also work when the
context dict is already flat.
"""
def _sub(match: re.Match[str]) -> str:
expr = match.group(1).strip()
return str(_resolve_path(expr, context))
try:
# Attempt to evaluate the placeholder as a Python expression first.
# This enables support for constructs like `context.get('key', default)`
# which the legacy implementation could not handle.
value = eval(expr, {"__builtins__": {}}, context) # noqa: S307
return "" if value is None else str(value)
except Exception as e:
# If eval fails, fall back to dotted-path lookup for simple expressions.
logger.debug(
"Eval failed for expr %r in simple template. Falling back to path resolution. Error: %s",
expr,
e,
)
# Legacy behaviour dotted path resolution.
value = _resolve_path(expr, context)
# Fallback strip leading "context." if still unresolved
if (value == "" or value is None) and expr.startswith("context."):
fallback_expr = expr[len("context.") :]
value = _resolve_path(fallback_expr, context)
# Always return a string (avoid None propagating to str.format)
return "" if value is None else str(value)
return re.sub(r"\{\{\s*(.*?)\s*\}\}", _sub, template)
+184
View File
@@ -0,0 +1,184 @@
# CleverAgents Multi-Agent Workflow Example
#
# This configuration file demonstrates a more complex agent network.
# It showcases a multi-agent workflow where an initial "classifier" agent
# routes tasks to specialized agents (a calculator and a web searcher),
# and a final "responder" agent synthesizes the results into a cohesive answer.
#
# To run this example, you will need to set the OPENAI_API_KEY environment variable.
# Example usage from the command line:
#
# export OPENAI_API_KEY="your_api_key_here"
# cleveragents run --config src/examples/multi_agent_workflow.yaml --prompt "What is the capital of France?"
# cleveragents run --config src/examples/multi_agent_workflow.yaml --prompt "Calculate 25 * 4"
cleveragents:
version: "1.0"
logging:
level: "INFO" # Set to "DEBUG" for more verbose output
template_engine: "JINJA2"
# Agent Definitions
# -----------------
# This section defines all the agents that will be part of the network.
# Each agent has a unique name, a type, and a specific configuration.
agents:
user_input_classifier:
type: llm
config:
provider: openai
model: gpt-3.5-turbo
api_key: "${OPENAI_API_KEY}" # Uses environment variable for security
system_message: |
You are a classification expert. Your task is to analyze the user's prompt and determine its primary intent.
Respond with ONLY ONE of the following keywords based on the prompt:
- CALCULATION: If the prompt requires a calculation.
- GENERAL: For all other prompts.
temperature: 0.0 # Low temperature for deterministic classification
calculator_extractor:
type: llm
config:
provider: openai
model: gpt-3.5-turbo
api_key: "${OPENAI_API_KEY}" # Uses environment variable for security
system_message: |
The user is providing a prompt that requires a calculation to be performed. Respond with ONLY the
calculation that needs to be performed without any words or texts present. Make sure the response uses
numbers and operators in algebraic format, no natural language. Here are some examples of prompts and
the responses I would expect:
- Prompt: "If I have one apple and someone gives me five more, how many apples do I have?"
Response: "1+5"
- Prompt: "What is six time 7"
Response: "6*7"
- Prompt: "5/9"
Response: "5/9"
temperature: 0.0 # Low temperature for deterministic classification
calculator_processor:
type: tool
config:
tools:
- name: calculator
calculator_responder:
type: llm
config:
provider: openai
model: gpt-4o-mini
api_key: "${OPENAI_API_KEY}"
system_message: |
You are a response synthesizer. Your job is to take the raw output from a tool which performs the
the mathematical calculation present in a users response, along with the users original prompt, and
construct a natural language response that provides the calculated answer as part of a friendly natural
language response to answer the question. Make sure your response restates the equation and the
calculation result as presented here. The equation will be provided by appending it to the end of
the original user prompt with a line beginning with "EQUATION: ". Similarly the calculation result will
be provided by appending it and the line will begin with "CALCULATION: ".
Here is an example of a prompt and the responses I might expect:
- Prompt:
If I have one apple and someone gives me five more, how many apples do I have?
EQUATION: 1+5
CALCULATION: 6
- Response:
This can be solved by adding 1+5 which results in 6, therefore you would have 6 apples.
temperature: 0.7
knowledge_checker:
type: llm
config:
provider: openai
model: gpt-4o-mini
api_key: "${OPENAI_API_KEY}"
system_message: |
You are friendly and helpful assistant. Your task is to analyze the user's prompt and determine if you
have enough knowledge to answer the question like an expert on the topic.
Respond with ONLY ONE of the following keywords based on the prompt:
- INFORMED: If you can answer the prompt with enough knowledge to give an expert quality response.
- IGNORANT: For all other prompts.
temperature: 0.7
web_search_agent:
type: tool
config:
tools:
- name: web_search
config:
# Uncomment and set the values below, or use environment variables.
# api_key: "${GOOGLE_API_KEY}"
# cx: "${GOOGLE_CX}"
search_engine: "google"
web_search_responder:
type: llm
config:
provider: openai
model: gpt-4o-mini
api_key: "${OPENAI_API_KEY}"
system_message: |
You are a response synthesizer. Your job is to take the raw output from a web search, along with the
original prompt a user provided, which the web search is relevant to, and provide an appropriate
response while incorporating the search data into that response as additional knowledge to provide a
more well rounded answer than you could on your own. You will be provided the original users response
followed by "SEARCH_RESULTS: " on its own line, which is then followed by all the information from
the search results.
The following is an example of a prompt you might get:
Tell me about the company CleverThis?
SEARCH_RESULTS: CleverThis is an AI company founded in 2024 by Jeffrey Phillips Freeman, Drew Morris,
and Justin Morris
temperature: 0.7
general_responder:
type: llm
config:
provider: openai
model: gpt-4o-mini
api_key: "${OPENAI_API_KEY}"
system_message: "You are a friendly and helpful assistant. Provide clear and concise answers to the user's questions."
temperature: 0.5
# Router Definition
# -----------------
# The router controls the flow of messages between the agents defined above.
# It uses a set of rules to direct messages based on their source, content, and context.
router:
name: main_router
# The 'routes' list contains the rules for message passing.
# Each rule specifies a source, a destination, and a condition for the route to be taken.
routes:
- source: input
destination: user_input_classifier
transform: "{{ context.get('initial_message', message) }}"
- source: user_input_classifier
destination: calculator_extractor
condition: "'CALCULATION' in message"
transform: "{{ context.get('initial_message', message) }}"
- source: calculator_extractor
destination: calculator_processor
- source: calculator_processor
destination: calculator_responder
transform: |-
{{ context.initial_message }}
EQUATION: {{ context.history[-2].message }}
CALCULATION: {{ message }}
- source: user_input_classifier
destination: knowledge_checker
condition: "'GENERAL' in message"
transform: "{{ context.get('initial_message', message) }}"
- source: knowledge_checker
destination: general_responder
condition: "'INFORMED' in message"
transform: "{{ context.get('initial_message', message) }}"
- source: knowledge_checker
destination: web_search_agent
condition: "'IGNORANT' in message"
transform: "{{ context.get('initial_message', message) }}"
- source: web_search_agent
destination: web_search_responder
transform: |-
{{ context.initial_message }}
SEARCH_RESULTS: {{ message }}
- source: web_search_responder
destination: output
- source: calculator_responder
destination: output
+8 -1
View File
@@ -4,5 +4,12 @@ agents:
api_key: mock-api-key
model: test-model
provider: openai
name: dummy
type: llm
router:
name: default_router
routes:
- destination: dummy
source: input
- destination: output
source: dummy
version: '1.0'
@@ -0,0 +1,9 @@
Feature: Message Context Persistence in Multi-Agent Workflow
To ensure reliable communication in complex agent networks,
the initial message from the user should be preserved in the context
and be accessible to agents throughout the entire workflow.
Scenario: Original message is accessible in a subsequent routing step
Given a router configured for a multi-step workflow
When I process the prompt "What is the capital of France?"
Then the final output should be "The capital of France is Paris."
@@ -0,0 +1,6 @@
Feature: Multi-Agent LLM Agent Configuration
Scenario: An LLM agent in a network should use its specific system message
Given a configuration with an LLM agent having a specific system message
When I run the network with a prompt
Then the LLM provider should have been called with the specific system message
@@ -0,0 +1,9 @@
Feature: Reproduce Multi-Agent Routing Bugs
This feature test aims to reproduce bugs observed in a multi-agent workflow,
specifically related to template transforms and condition evaluation.
Scenario: A message routes incorrectly in a multi-step workflow
Given a router with a buggy multi-step configuration
When I process the initial prompt "What is the capital of France?"
Then the buggy workflow output should be "search result for: What is the capital of France?"
@@ -31,13 +31,13 @@ def step_impl(context, provider):
raise AgentCreationError(f"Unsupported LLM provider: {provider}")
if provider == "openai":
async def mock_openai_response():
async def mock_openai_response(messages):
return "This is a mock response from OpenAI."
context.agent._generate_openai_response = mock_openai_response
elif provider == "anthropic":
async def mock_anthropic_response():
async def mock_anthropic_response(messages):
return "This is a mock response from Anthropic."
context.agent._generate_anthropic_response = mock_anthropic_response
@@ -80,13 +80,13 @@ def step_impl(context, provider):
)
if provider == "openai":
async def mock_openai_response():
async def mock_openai_response(messages):
return "This is a mock response from OpenAI."
context.agent._generate_openai_response = mock_openai_response
elif provider == "anthropic":
async def mock_anthropic_response():
async def mock_anthropic_response(messages):
return "This is a mock response from Anthropic."
context.agent._generate_anthropic_response = mock_anthropic_response
@@ -146,7 +146,7 @@ def step_impl(context):
context.agent.format_prompt = mock_format_prompt
async def mock_openai_response():
async def mock_openai_response(messages):
return "This is a mock response."
context.agent._generate_openai_response = mock_openai_response
@@ -185,7 +185,7 @@ def step_impl(context):
context.agent.process_response = mock_process_response
async def mock_openai_response():
async def mock_openai_response(messages):
return "This is a mock response."
context.agent._generate_openai_response = mock_openai_response
@@ -213,7 +213,7 @@ def step_impl(context, model):
context.template_renderer,
)
async def mock_openai_response():
async def mock_openai_response(messages):
return "This is a mock response."
context.agent._generate_openai_response = mock_openai_response
+3 -3
View File
@@ -26,13 +26,13 @@ def step_impl(context, provider):
)
if provider == "openai":
async def mock_openai_response():
async def mock_openai_response(messages):
return "This is a mock response from OpenAI."
context.agent._generate_openai_response = mock_openai_response
elif provider == "anthropic":
async def mock_anthropic_response():
async def mock_anthropic_response(messages):
return "This is a mock response from Anthropic."
context.agent._generate_anthropic_response = mock_anthropic_response
@@ -48,7 +48,7 @@ def step_impl(context):
def step_impl(context):
assert context.response is not None
assert len(context.response) > 0
assert len(context.agent.memory["messages"]) >= 3
assert len(context.agent.memory["messages"]) == 2
assert context.agent.memory["messages"][-1]["role"] == "assistant"
assert context.agent.memory["messages"][-1]["content"] == context.response
@@ -0,0 +1,141 @@
import asyncio
import unittest.mock
from typing import Any
from typing import Dict
from typing import Optional
from behave import given
from behave import then
from behave import when
from cleveragents.agents.base import Agent
from cleveragents.routing.router import Router
from cleveragents.templates.renderer import TemplateEngine
from cleveragents.templates.renderer import TemplateRenderer
class MockAgent(Agent):
"""A mock agent to control its output and track its inputs."""
def __init__(self, name, config, template_renderer, response="default response"):
super().__init__(name, config, template_renderer)
# Use create_autospec so the mock has the correct async signature,
# ensuring it handles keyword arguments like `context=...`.
self.process = unittest.mock.create_autospec(
self.process, return_value=response
)
async def process(
self,
message: str,
context: Optional[Dict[str, Any]] = None,
) -> str: # type: ignore[override]
"""
Concrete implementation to satisfy the abstract method in `Agent`.
It is immediately shadowed by the AsyncMock assigned in `__init__`,
so it should never be invoked during tests.
"""
raise NotImplementedError(
"This method is expected to be mocked."
) # pragma: no cover
def get_capabilities(self):
return ["mock"]
@given("a router configured for a multi-step workflow")
def step_impl(context):
"""
Sets up a router with two agents: a classifier and a responder.
The key part is the route from the classifier to the responder,
which uses a transform to pass the original message from the context.
"""
# Use the simple template engine which exhibits the bug
context.template_renderer = TemplateRenderer(engine_type=TemplateEngine.SIMPLE)
# Mock agents
context.classifier_agent = MockAgent(
name="classifier",
config={},
template_renderer=context.template_renderer,
response="NEEDS_INFO",
)
context.responder_agent = MockAgent(
name="responder",
config={},
template_renderer=context.template_renderer,
response="The capital of France is Paris.",
)
agents = {
"classifier": context.classifier_agent,
"responder": context.responder_agent,
}
# Setup router
context.router = Router(
name="test_router", agents=agents, template_renderer=context.template_renderer
)
# Define routes that mimic the scenario leading to the bug
routes = [
{"from": "input", "to": "classifier"},
{
"from": "classifier",
"to": "responder",
# This transform is crucial. It attempts to retrieve the original message.
# The bug is that this evaluates to an empty string.
"transform": "{{ context.initial_message }}",
},
{"from": "responder", "to": "output"},
]
context.router.add_routes(routes)
@when('I process the prompt "{prompt}"')
def step_impl(context, prompt):
"""
Processes the given prompt through the configured router.
"""
context.initial_prompt = prompt
context.final_result = asyncio.run(
context.router.process_message(context.initial_prompt)
)
@then('the final output should be "{expected_output}"')
def step_impl(context, expected_output):
"""
Verifies that the responder agent received the correct (original) message
and that the final output from the router is as expected.
"""
# 1. Verify the first agent was called with the initial prompt.
context.classifier_agent.process.assert_called_once_with(
context.initial_prompt, context=unittest.mock.ANY
)
# 2. This is the key assertion that should FAIL due to the bug.
# We expect the responder agent to be called with the initial prompt,
# passed via the transform. The bug causes it to be called with "".
try:
context.responder_agent.process.assert_called_once_with(
context.initial_prompt, context=unittest.mock.ANY
)
except AssertionError as e:
# We expect this to fail, but let's add a more informative message
# to the test output if it does.
call_args, _ = context.responder_agent.process.call_args
actual_message = call_args[0]
error_message = (
f"BUG REPRODUCED: Responder agent was called with incorrect message.\n"
f"Expected: '{context.initial_prompt}'\n"
f"Actual: '{actual_message}'\n"
f"Original AssertionError: {e}"
)
raise AssertionError(error_message) from e
# 3. Check the final output. Note: this assertion may pass if the mock
# is not dependent on its input, but the assertion above is the real test.
assert (
context.final_result == expected_output
), f"Expected final output '{expected_output}', but got '{context.final_result}'"
@@ -0,0 +1,122 @@
import asyncio
import os
import tempfile
import textwrap
from pathlib import Path
from unittest.mock import AsyncMock
from unittest.mock import MagicMock
from unittest.mock import patch
import yaml
from behave import given
from behave import then
from behave import when
from cleveragents.core.application import CleverAgentsApp
from tests.mocks.llm_providers import MockOpenAIResponse
@given("a configuration with an LLM agent having a specific system message")
def step_impl(context):
"""
Creates a temporary YAML configuration file with a specific system message.
"""
context.expected_system_message = (
"You are a test classifier. Your only job is to work."
)
# Define the configuration as a Python dictionary to ensure
# correct YAML structure and avoid indentation issues.
config_data = {
"version": "1.0",
"logging": {"level": "INFO"},
"agents": {
"classifier": {
"type": "llm",
"config": {
"provider": "openai",
"model": "gpt-3.5-turbo",
"api_key": "some_key",
"system_message": context.expected_system_message,
"temperature": 0.0,
},
}
},
"router": {
"name": "main_router",
"routes": [{"source": "input", "destination": "classifier"}],
},
}
# Create a temporary file for the configuration
with tempfile.NamedTemporaryFile(
mode="w", delete=False, suffix=".yaml", encoding="utf-8"
) as f:
yaml.dump(config_data, f)
context.config_file_path = f.name
@when("I run the network with a prompt")
def step_impl(context):
"""
Runs the CleverAgentsApp and mocks the aiohttp call to capture the request.
"""
# Mock `aiohttp.ClientSession` to intercept the HTTP call made by the LLMAgent.
with patch("aiohttp.ClientSession") as MockSession:
# The session's `post` method returns an async context manager.
mock_post_cm = AsyncMock()
# The context manager, when entered, yields a mock response object.
mock_response = MagicMock()
mock_response.status = 200
# Mimic the OpenAI API JSON payload.
expected_json = {"choices": [{"message": {"content": "Mocked response"}}]}
mock_response.json = AsyncMock(return_value=expected_json)
mock_post_cm.__aenter__.return_value = mock_response
# Configure the mock session instance.
mock_session_instance = MockSession.return_value
mock_session_instance.post.return_value = mock_post_cm
# Ensure the `ClientSession` itself behaves as an async context manager.
mock_session_instance.__aenter__.return_value = mock_session_instance
# Store the mock `post` call for verification in the 'then' step.
context.mock_post = mock_session_instance.post
# Initialize and run the app.
app = CleverAgentsApp(config_files=[Path(context.config_file_path)])
asyncio.run(app.run_single_shot("This is a test prompt."))
@then("the LLM provider should have been called with the specific system message")
def step_impl(context):
"""
Asserts that the mocked aiohttp `post` call was made with the correct system message.
"""
try:
context.mock_post.assert_called_once()
# Extract the keyword arguments the mock was called with.
_, kwargs = context.mock_post.call_args
json_payload = kwargs.get("json", {})
messages = json_payload.get("messages", [])
assert len(messages) > 0, "No messages were sent to the LLM."
system_message = next((m for m in messages if m.get("role") == "system"), None)
assert (
system_message is not None
), "No system message was sent in the call to the LLM."
actual_system_message = system_message.get("content")
assert actual_system_message == context.expected_system_message, (
f"Incorrect system message. Expected: '{context.expected_system_message}', "
f"Got: '{actual_system_message}'"
)
finally:
# Clean up the temporary file.
if hasattr(context, "config_file_path"):
os.remove(context.config_file_path)
+22 -16
View File
@@ -1,50 +1,56 @@
import asyncio
import os
from pathlib import Path
from unittest.mock import AsyncMock
from unittest.mock import patch
import yaml
from behave import given
from behave import then
from behave import when
from cleveragents.network import AgentNetwork
from cleveragents.core.application import CleverAgentsApp
@given('a configuration file "{config_file}"')
def step_impl_config_file(context, config_file):
# Create a minimal configuration file for AgentNetwork
# Create a minimal but valid CleverAgents configuration file
config_data = {
"version": "1.0",
"agents": {
"dummy": {
"type": "llm",
"name": "dummy",
"config": { # Add the missing config field
"config": {
"provider": "openai",
"model": "test-model",
"api_key": "mock-api-key",
},
}
}
},
"router": {
"name": "default_router",
"routes": [
{"source": "input", "destination": "dummy"},
{"source": "dummy", "destination": "output"},
],
},
}
config_path = Path(config_file)
config_path.parent.mkdir(parents=True, exist_ok=True)
with open(config_path, "w") as f:
with open(config_path, "w", encoding="utf-8") as f:
yaml.dump(config_data, f)
context.config_file_path = config_file
@when('I process the message "{message}"')
def step_impl_process_message(context, message):
# Initialize AgentNetwork with the created configuration file
network = AgentNetwork([Path(context.config_file_path)], verbose=False)
# Monkey-patch the router's process_message to return a fixed response asynchronously
async def dummy_process_message(msg, ctx):
return "Test Response"
network.router.process_message = dummy_process_message
context.response = asyncio.run(network.process(message))
# Patch the router's process_message method to isolate the app logic
with patch(
"cleveragents.routing.router.Router.process_message",
new=AsyncMock(return_value="Test Response"),
):
app = CleverAgentsApp([Path(context.config_file_path)], verbose=False)
context.response = asyncio.run(app.run_single_shot(message))
@then("I should receive a non-empty network response")
@@ -0,0 +1,99 @@
import asyncio
from unittest.mock import AsyncMock
from unittest.mock import MagicMock
from behave import given
from behave import then
from behave import when
from cleveragents.agents.base import Agent
from cleveragents.routing.router import Router
from cleveragents.templates.renderer import TemplateEngine
from cleveragents.templates.renderer import TemplateRenderer
@given("a router with a buggy multi-step configuration")
def step_impl(context):
"""
Sets up a router configuration that mimics the user's failing workflow.
This includes a classifier agent, a responder agent, and routes that rely on
both transforms with context and string-based conditions.
"""
# Use a real TemplateRenderer to expose issues with template syntax.
# The SIMPLE engine does not support method calls like .get(), which is one bug.
context.template_renderer = TemplateRenderer(engine_type=TemplateEngine.SIMPLE)
# Mock Agent A (Classifier) - returns a keyword that should trigger a condition
mock_agent_a = MagicMock(spec=Agent)
mock_agent_a.process = AsyncMock(return_value="SEARCH")
# Mock Agent B (Responder) - processes the original message
async def process_b(message, context=None):
return f"search result for: {message}"
mock_agent_b = MagicMock(spec=Agent)
mock_agent_b.process = AsyncMock(side_effect=process_b)
context.agents = {
"classifier": mock_agent_a,
"responder": mock_agent_b,
}
context.router = Router(
name="test_router",
agents=context.agents,
template_renderer=context.template_renderer,
)
routes = [
{
"source": "input",
"destination": "classifier",
# This transform uses invalid syntax for the SIMPLE engine,
# reproducing the first observed bug from the logs.
"transform": "{{ context.get('initial_message', message) }}",
},
{
"source": "classifier",
"destination": "responder",
# This condition fails in the user's log, reproducing the second bug.
"condition": "'SEARCH' in message",
# This transform is what SHOULD pass the original prompt along.
"transform": "{{ context.initial_message }}",
},
{
"source": "responder",
"destination": "output",
},
]
context.router.add_routes(routes)
@when('I process the initial prompt "{prompt}"')
def step_impl(context, prompt):
"""
Runs the router's end-to-end processing with the initial prompt.
"""
context.initial_prompt = prompt
try:
# Run the async process_message method
context.final_output = asyncio.run(
context.router.process_message(context.initial_prompt)
)
context.error = None
except Exception as e:
context.final_output = None
context.error = e
@then('the buggy workflow output should be "{expected_output}"')
def step_impl(context, expected_output):
"""
Asserts that the final output matches the expected result, which would
only happen if all routing steps, transforms, and conditions worked correctly.
"""
# This assertion will fail with the current buggy code.
assert context.error is None, f"An unexpected error occurred: {context.error}"
assert (
context.final_output == expected_output
), f"Expected output '{expected_output}', but got '{context.final_output}'"
@@ -92,17 +92,27 @@ def step_impl(context):
assert context.responses["search"] == "Response from agent2"
assert context.responses["long"] == "Response from agent3"
# Check that each agent's process method was called with the right message
# Check that each agent's process method was called with the right message and context
context.agents["agent1"].process.assert_called_once_with(
"I need help with something", {"message": "I need help with something"}
"I need help with something",
{
"message": "I need help with something",
"initial_message": "I need help with something",
},
)
context.agents["agent2"].process.assert_called_once_with(
"Can you search for information?",
{"message": "Can you search for information?"},
{
"message": "Can you search for information?",
"initial_message": "Can you search for information?",
},
)
context.agents["agent3"].process.assert_called_once_with(
"This is a longer message with more than five words",
{"message": "This is a longer message with more than five words"},
{
"message": "This is a longer message with more than five words",
"initial_message": "This is a longer message with more than five words",
},
)
@@ -147,8 +157,8 @@ def step_impl(context):
context.process_spy.assert_called_once()
args, kwargs = context.process_spy.call_args
assert (
args[0] == "Transformed: {message}"
), f"Expected 'Transformed: {{message}}' but got '{args[0]}'"
args[0] == "Transformed: Original message"
), f"Expected 'Transformed: Original message' but got '{args[0]}'"
@given("a router with invalid routes")
+22 -63
View File
@@ -69,24 +69,29 @@ def step_impl(context):
# Create mock agents
agent1 = MockAgent("agent1", {}, context.template_renderer)
agent2 = MockAgent("agent2", {}, context.template_renderer)
context.agents = {"agent1": agent1, "agent2": agent2}
# Create router
context.router = Router(
"test_router", {"agent1": agent1, "agent2": agent2}, context.template_renderer
)
context.router = Router("test_router", context.agents, context.template_renderer)
# Add conditional routes
# Add conditional routes using the 'python' condition type
context.router.add_routes(
[
{
"from": "input",
"to": "agent1",
"condition": "message.startswith('A')",
"condition": {
"type": "python",
"code": "result = message.startswith('A')",
},
},
{
"from": "input",
"to": "agent2",
"condition": "not message.startswith('A')",
"condition": {
"type": "python",
"code": "result = not message.startswith('A')",
},
},
{"from": "agent1", "to": "output"},
{"from": "agent2", "to": "output"},
@@ -96,47 +101,32 @@ def step_impl(context):
@when("I send a message that matches a specific condition")
def step_impl(context):
# Mock the route_message method to avoid the actual routing logic
original_route_message = context.router.route_message
async def mock_route_message(message, source="input", ctx=None):
if message.startswith("A"):
agent1 = context.router.agents["agent1"]
return await agent1.process(message, ctx)
else:
agent2 = context.router.agents["agent2"]
return await agent2.process(message, ctx)
# Replace with mock
context.router.route_message = mock_route_message
# Now process the message
context.message_a = "A test message"
context.response_a = asyncio.run(context.router.process_message(context.message_a))
# Restore original method
context.router.route_message = original_route_message
@then("the message should be routed to the appropriate agent")
def step_impl(context):
agent1 = context.router.agents["agent1"]
agent2 = context.router.agents["agent2"]
agent1 = context.agents["agent1"]
agent2 = context.agents["agent2"]
assert context.message_a in agent1.processed_messages
assert context.message_a not in agent2.processed_messages
assert "Processed by agent1" in context.response_a
assert agent1.processed_messages[-1] == "A test message"
assert "B test message" not in agent1.processed_messages
assert "A test message" not in agent2.processed_messages
@when("I send a message that doesn't match, it should be routed differently")
@then("when I send a message that doesn't match, it should be routed differently")
def step_impl(context):
context.message_b = "B test message"
context.response_b = asyncio.run(context.router.process_message(context.message_b))
agent1 = context.router.agents["agent1"]
agent2 = context.router.agents["agent2"]
agent1 = context.agents["agent1"]
agent2 = context.agents["agent2"]
assert "Processed by agent2" in context.response_b
assert context.message_b not in agent1.processed_messages
assert context.message_b in agent2.processed_messages
assert agent2.processed_messages[-1] == "B test message"
@given("a network with message transformation rules")
@@ -170,37 +160,6 @@ def step_impl(context):
assert any(msg.startswith("TRANSFORMED:") for msg in agent.processed_messages)
@then("when I send a message that doesn't match, it should be routed differently")
def step_impl(context):
# Store the original method for later restoration
original_route_message = context.router.route_message
async def mock_route_message(message, source="input", ctx=None):
if not message.startswith("A"):
agent2 = context.router.agents["agent2"]
return await agent2.process(message, ctx)
else:
agent1 = context.router.agents["agent1"]
return await agent1.process(message, ctx)
# Replace with mock
context.router.route_message = mock_route_message
# Process a message that doesn't match the condition
context.message_b = "B test message"
context.response_b = asyncio.run(context.router.process_message(context.message_b))
# Restore original method
context.router.route_message = original_route_message
# Verify routing
agent1 = context.router.agents["agent1"]
agent2 = context.router.agents["agent2"]
assert context.message_b not in agent1.processed_messages
assert context.message_b in agent2.processed_messages
@when("I send a message through the network")
def step_impl(context):
context.message = "Test message for transformation"