brent.edwards
17fe46d925
CI / lint (pull_request) Successful in 18s
CI / typecheck (pull_request) Successful in 26s
CI / security (pull_request) Successful in 17s
CI / quality (pull_request) Successful in 14s
CI / build (pull_request) Successful in 12s
CI / behave (3.13) (pull_request) Failing after 3m50s
CI / docker (pull_request) Has been skipped
CI / helm (pull_request) Has been skipped
CI / coverage (pull_request) Failing after 4m20s
There had been over 100 behave tests failing. There should be none failing now.
300 lines
9.4 KiB
Python
300 lines
9.4 KiB
Python
import asyncio
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.agents.base import Agent
|
|
from cleveragents.langgraph.nodes import Edge, Node, NodeConfig, NodeType
|
|
from cleveragents.langgraph.state import GraphState
|
|
|
|
|
|
class DummyAgent(Agent):
|
|
def __init__(self, name: str, response: str):
|
|
super().__init__(name)
|
|
self._response = response
|
|
self.received_inputs = []
|
|
|
|
async def process_message(self, message, context=None):
|
|
self.received_inputs.append(message)
|
|
return self._response
|
|
|
|
def get_capabilities(self):
|
|
return []
|
|
|
|
|
|
class DummyToolAgent(Agent):
|
|
def __init__(self, name: str, response: str):
|
|
super().__init__(name)
|
|
self._response = response
|
|
self.received_inputs = []
|
|
|
|
async def process_message(self, message, context=None):
|
|
self.received_inputs.append(message)
|
|
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 history limit 2 messages")
|
|
def step_node_with_history_limit(context):
|
|
config = NodeConfig(
|
|
name="history", type=NodeType.AGENT, metadata={"max_history_messages": 2}
|
|
)
|
|
context.node = Node(config)
|
|
|
|
|
|
@given("a conversation history of 3 short messages")
|
|
def step_history_three_messages(context):
|
|
context.messages = [
|
|
{"role": "user", "content": "hi"},
|
|
{"role": "assistant", "content": "there"},
|
|
{"role": "user", "content": "friend"},
|
|
]
|
|
|
|
|
|
@when("I prepare the conversation history")
|
|
def step_prepare_history(context):
|
|
trimmed, truncated = context.node._prepare_conversation_history(context.messages)
|
|
context.trimmed_history = trimmed
|
|
context.truncated = truncated
|
|
|
|
|
|
@then("it should return 2 messages and indicate truncation")
|
|
def step_assert_history(context):
|
|
assert len(context.trimmed_history) == 2
|
|
assert context.truncated is True
|
|
assert context.trimmed_history[0]["content"] == "there"
|
|
assert context.trimmed_history[1]["content"] == "friend"
|
|
|
|
|
|
@given("an agent node referencing a missing agent")
|
|
def step_agent_missing(context):
|
|
config = NodeConfig(name="agent", type=NodeType.AGENT, agent="missing")
|
|
context.node = Node(config, agents={})
|
|
context.state = _make_state(messages=[{"role": "user", "content": "hi"}])
|
|
|
|
|
|
@when("I execute the agent node")
|
|
def step_execute_agent(context):
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.result = loop.run_until_complete(context.node.execute(context.state))
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
@then("it should return a not found assistant message")
|
|
def step_assert_missing_agent(context):
|
|
messages = context.result.get("messages", [])
|
|
assert messages
|
|
assert messages[0]["content"] == "Agent missing not found"
|
|
assert messages[0]["agent"] == "missing"
|
|
|
|
|
|
@given('a tool agent node with a current_message "{text}" and long history')
|
|
def step_tool_agent_with_current(context, text):
|
|
tool_agent = DummyToolAgent("tool", response="ok")
|
|
config = NodeConfig(
|
|
name="tool_node",
|
|
type=NodeType.AGENT,
|
|
agent="tool",
|
|
metadata={"max_history_messages": 1},
|
|
)
|
|
context.node = Node(config, agents={"tool": tool_agent})
|
|
long_history = [
|
|
{"role": "user", "content": "first"},
|
|
{"role": "assistant", "content": "second"},
|
|
]
|
|
context.state = _make_state(
|
|
messages=long_history, metadata={"current_message": text}
|
|
)
|
|
|
|
|
|
@when("I execute the tool agent node")
|
|
def step_execute_tool_agent(context):
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.result = loop.run_until_complete(context.node.execute(context.state))
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
@then('the tool agent should receive input "{text}"')
|
|
def step_assert_tool_input(context, text):
|
|
agent = context.node.agents["tool"]
|
|
assert agent.received_inputs[-1] == text
|
|
|
|
|
|
@then("the response metadata should include last_agent_node")
|
|
def step_assert_last_agent_metadata(context):
|
|
metadata = context.result.get("metadata", {})
|
|
assert metadata.get("last_agent_node") == "tool_node"
|
|
|
|
|
|
@given("a function node with an async function")
|
|
def step_async_function_node(context):
|
|
async def async_fn(state):
|
|
return "async-result"
|
|
|
|
config = NodeConfig(name="fn", type=NodeType.FUNCTION, function="async_fn")
|
|
context.node = Node(config, agents={"async_fn": async_fn})
|
|
context.state = _make_state()
|
|
|
|
|
|
@given("a function node with a sync function that returns a coroutine")
|
|
def step_sync_function_coroutine(context):
|
|
async def inner():
|
|
return "sync-inner"
|
|
|
|
def sync_fn(state):
|
|
return inner()
|
|
|
|
config = NodeConfig(name="fn_sync", type=NodeType.FUNCTION, function="sync_fn")
|
|
context.node = Node(config, agents={"sync_fn": sync_fn})
|
|
context.state = _make_state()
|
|
|
|
|
|
@when("I execute the function node")
|
|
def step_execute_function_node(context):
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.result = loop.run_until_complete(context.node.execute(context.state))
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
@then('the function result message should be "{expected}"')
|
|
def step_assert_function_result(context, expected):
|
|
messages = context.result.get("messages", [])
|
|
assert messages
|
|
assert messages[0]["content"] == expected
|
|
|
|
|
|
@given('a conditional node looking for field "{field}" equals "{value}"')
|
|
def step_conditional_node(context, field, value):
|
|
config = NodeConfig(
|
|
name="cond",
|
|
type=NodeType.CONDITIONAL,
|
|
condition={"field": field, "equals": value},
|
|
)
|
|
context.node = Node(config)
|
|
context.state = _make_state(
|
|
messages=[{"role": "assistant", field: value, "content": "reply"}]
|
|
)
|
|
|
|
|
|
@when("I execute the conditional node with matching assistant message")
|
|
def step_execute_conditional(context):
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.result = loop.run_until_complete(context.node.execute(context.state))
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
@then("the condition_result should be true and metadata flagged")
|
|
def step_assert_conditional(context):
|
|
assert context.result.get("condition_result") is True
|
|
metadata = context.result.get("metadata", {})
|
|
assert metadata.get("conditional_edges_used") is True
|
|
|
|
|
|
@given('a message router node with a rule matching field "{field}" value "{value}"')
|
|
def step_message_router_node(context, field, value):
|
|
rules = [{"condition": {"field": field, "value": value}, "target": "urgent_node"}]
|
|
config = NodeConfig(
|
|
name="router", type=NodeType.MESSAGE_ROUTER, metadata={"rules": rules}
|
|
)
|
|
context.node = Node(config)
|
|
context.state = _make_state(metadata={"current_message": {field: value}})
|
|
|
|
|
|
@when('I execute the message router with current_message containing topic "urgent"')
|
|
def step_execute_message_router(context):
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
context.result = loop.run_until_complete(context.node.execute(context.state))
|
|
finally:
|
|
loop.close()
|
|
|
|
|
|
@then('it should route to "urgent_node"')
|
|
def step_assert_router(context):
|
|
metadata = context.result.get("metadata", {})
|
|
assert metadata.get("routed_to") == "urgent_node"
|
|
|
|
|
|
@given('an edge with equals condition on field "{field}" value "{value}"')
|
|
def step_edge_condition(context, field, value):
|
|
context.edge = Edge(
|
|
source="a", target="b", condition={"field": field, "equals": value}
|
|
)
|
|
context.node = Node(NodeConfig(name="edge", type=NodeType.END))
|
|
|
|
|
|
@when("I evaluate the edge condition with matching last message")
|
|
def step_eval_edge(context):
|
|
state = _make_state(messages=[{"role": "assistant", "content": "done"}])
|
|
context.edge_result = context.node.evaluate_edge_condition(context.edge, state)
|
|
|
|
|
|
@then("the edge condition should be true")
|
|
def step_assert_edge_true(context):
|
|
assert context.edge_result is True
|
|
|
|
|
|
@then("evaluating with no messages should also default to true")
|
|
def step_assert_edge_default_true(context):
|
|
empty_state = _make_state()
|
|
assert context.node.evaluate_edge_condition(context.edge, empty_state) is True
|
|
|
|
|
|
@given('I evaluate equals condition on message "{text}"')
|
|
def step_eval_equals(context, text):
|
|
context.node = Node(NodeConfig(name="eval", type=NodeType.END))
|
|
context.eval_equals = context.node._eval_condition(text, {"equals": text})
|
|
|
|
|
|
@then("the equals evaluation should return true")
|
|
def step_assert_eval_true(context):
|
|
assert context.eval_equals is True
|
|
|
|
|
|
@when('I evaluate field condition on message dict with foo value "bar"')
|
|
def step_eval_field_true(context):
|
|
context.eval_field_true = context.node._eval_condition(
|
|
{"foo": "bar"}, {"field": "foo", "value": "bar"}
|
|
)
|
|
|
|
|
|
@then("the field evaluation should return true")
|
|
def step_assert_field_true(context):
|
|
assert context.eval_field_true is True
|
|
|
|
|
|
@when("I evaluate field condition on non-dict message")
|
|
def step_eval_field_false(context):
|
|
context.eval_field_false = context.node._eval_condition(
|
|
"not a dict", {"field": "foo", "value": "bar"}
|
|
)
|
|
|
|
|
|
@then("it should return false")
|
|
def step_assert_field_false(context):
|
|
assert context.eval_field_false is False
|
|
|
|
|
|
@given("a node configured for parallel execution")
|
|
def step_node_parallel(context):
|
|
config = NodeConfig(name="parallel", type=NodeType.END, parallel=True)
|
|
context.node = Node(config)
|
|
|
|
|
|
@then("can_execute_parallel should return true")
|
|
def step_assert_parallel(context):
|
|
assert context.node.can_execute_parallel() is True
|