forked from HAL9000/cleveragents-core
brent.edwards
17fe46d925
There had been over 100 behave tests failing. There should be none failing now.
223 lines
7.2 KiB
Python
223 lines
7.2 KiB
Python
import asyncio
|
|
from copy import deepcopy
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.agents.base import Agent
|
|
from cleveragents.langgraph.nodes import Node, NodeConfig, NodeType
|
|
from cleveragents.langgraph.state import GraphState
|
|
|
|
|
|
class FailingAgent(Agent):
|
|
def __init__(self, name: str, exc: Exception):
|
|
super().__init__(name)
|
|
self.exc = exc
|
|
self.received_inputs = []
|
|
|
|
async def process_message(self, message, context=None):
|
|
self.received_inputs.append(message)
|
|
raise self.exc
|
|
|
|
def get_capabilities(self):
|
|
return []
|
|
|
|
|
|
class RecordingAgent(Agent):
|
|
def __init__(self, name: str, response: str):
|
|
super().__init__(name)
|
|
self.response = response
|
|
self.received_inputs = []
|
|
self.received_contexts = []
|
|
|
|
async def process_message(self, message, context=None):
|
|
self.received_inputs.append(message)
|
|
self.received_contexts.append(deepcopy(context))
|
|
return self.response
|
|
|
|
def get_capabilities(self):
|
|
return []
|
|
|
|
|
|
def _make_state(messages=None, metadata=None):
|
|
return GraphState(messages=messages or [], metadata=metadata or {})
|
|
|
|
|
|
@given("a node with default history limits")
|
|
def step_default_history(context):
|
|
config = NodeConfig(name="history_default", type=NodeType.AGENT)
|
|
context.node = Node(config)
|
|
|
|
|
|
@when("I prepare the conversation history for empty input")
|
|
def step_prepare_empty_history(context):
|
|
trimmed, truncated = context.node._prepare_conversation_history([])
|
|
context.trimmed_history = trimmed
|
|
context.truncated = truncated
|
|
|
|
|
|
@then("it should return empty history without truncation")
|
|
def step_assert_empty_history(context):
|
|
assert context.trimmed_history == []
|
|
assert context.truncated is False
|
|
|
|
|
|
@given("a node with invalid history metadata and two messages")
|
|
def step_invalid_history_metadata(context):
|
|
config = NodeConfig(
|
|
name="history_invalid",
|
|
type=NodeType.AGENT,
|
|
metadata={"max_history_messages": "oops", "max_history_chars": "nope"},
|
|
)
|
|
context.node = Node(config)
|
|
context.messages = [
|
|
{"role": "user", "content": "first"},
|
|
{"role": "assistant", "content": "second"},
|
|
]
|
|
|
|
|
|
@when("I prepare the conversation history with invalid limits")
|
|
def step_prepare_invalid_history(context):
|
|
trimmed, truncated = context.node._prepare_conversation_history(context.messages)
|
|
context.trimmed_history = trimmed
|
|
context.truncated = truncated
|
|
|
|
|
|
@then("it should return full history without truncation")
|
|
def step_assert_invalid_history(context):
|
|
assert context.trimmed_history == context.messages
|
|
assert context.truncated is False
|
|
|
|
|
|
@given("an agent node without an agent configured")
|
|
def step_agent_missing_config(context):
|
|
config = NodeConfig(name="agent_missing", type=NodeType.AGENT)
|
|
context.node = Node(config)
|
|
context.state = _make_state(messages=[{"role": "user", "content": "hi"}])
|
|
|
|
|
|
@when("I execute the node expecting an agent configuration error")
|
|
def step_execute_missing_agent(context):
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.result = loop.run_until_complete(context.node.execute(context.state))
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
@then("the result should include failed_node and error text")
|
|
def step_assert_missing_agent_error(context):
|
|
assert context.result.get("failed_node") == "agent_missing"
|
|
assert "no agent specified" in context.result.get("error", "")
|
|
|
|
|
|
@given("an agent node with no prior messages")
|
|
def step_agent_no_messages(context):
|
|
agent = RecordingAgent("rec", response="ok")
|
|
config = NodeConfig(name="agent_empty", type=NodeType.AGENT, agent="rec")
|
|
context.node = Node(config, agents={"rec": agent})
|
|
context.state = _make_state(messages=[])
|
|
|
|
|
|
@when("I execute the agent node with empty history")
|
|
def step_execute_agent_empty_history(context):
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.result = loop.run_until_complete(context.node.execute(context.state))
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
@then("the agent should receive empty input and track node metadata")
|
|
def step_assert_empty_agent_input(context):
|
|
agent = context.node.agents["rec"]
|
|
assert agent.received_inputs[-1] == ""
|
|
metadata = context.result.get("metadata", {})
|
|
assert metadata.get("last_agent_node") == "agent_empty"
|
|
|
|
|
|
@given("an agent node with nested metadata context and long history")
|
|
def step_agent_nested_context(context):
|
|
agent = RecordingAgent("rec", response="nested-ok")
|
|
config = NodeConfig(
|
|
name="agent_nested",
|
|
type=NodeType.AGENT,
|
|
agent="rec",
|
|
metadata={"max_history_messages": 1},
|
|
)
|
|
context.node = Node(config, agents={"rec": agent})
|
|
history = [
|
|
{"role": "user", "content": "short"},
|
|
{"role": "assistant", "content": "long message that exceeds limit"},
|
|
]
|
|
nested_metadata = {"current_message": "now", "context": {"foo": "bar"}}
|
|
context.state = _make_state(messages=history, metadata=nested_metadata)
|
|
|
|
|
|
@when("I execute the agent node with nested context")
|
|
def step_execute_agent_nested(context):
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.result = loop.run_until_complete(context.node.execute(context.state))
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
@then("the agent context should merge nested values and mark truncation")
|
|
def step_assert_agent_context_merge(context):
|
|
agent = context.node.agents["rec"]
|
|
received_context = agent.received_contexts[-1]
|
|
assert received_context["foo"] == "bar"
|
|
assert received_context["_history_truncated"] is True
|
|
assert received_context["_history_original_length"] == 2
|
|
|
|
|
|
@given("an agent node whose handler raises an exception")
|
|
def step_agent_raises(context):
|
|
failing_agent = FailingAgent("fail", RuntimeError("boom"))
|
|
config = NodeConfig(name="agent_fail", type=NodeType.AGENT, agent="fail")
|
|
context.node = Node(config, agents={"fail": failing_agent})
|
|
context.state = _make_state(messages=[{"role": "user", "content": "hello"}])
|
|
|
|
|
|
@when("I execute the failing agent node")
|
|
def step_execute_failing_agent(context):
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.result = loop.run_until_complete(context.node.execute(context.state))
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
@then("the agent response should contain the error string")
|
|
def step_assert_agent_error_path(context):
|
|
messages = context.result.get("messages", [])
|
|
assert messages
|
|
assert "Error processing message: boom" in messages[0]["content"]
|
|
|
|
|
|
@given("a function node missing a target with retry policy")
|
|
def step_function_missing_with_retry(context):
|
|
config = NodeConfig(
|
|
name="fn_missing",
|
|
type=NodeType.FUNCTION,
|
|
function="missing_fn",
|
|
retry_policy={"max_retries": 1, "delay": 0},
|
|
)
|
|
context.node = Node(config, agents={})
|
|
context.state = _make_state()
|
|
|
|
|
|
@when("I execute the function node with retries")
|
|
def step_execute_function_missing(context):
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.result = loop.run_until_complete(context.node.execute(context.state))
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
@then("it should report failure after retries")
|
|
def step_assert_function_missing(context):
|
|
assert context.result.get("failed_node") == "fn_missing"
|
|
assert "not found" in context.result.get("error", "")
|