AgentForge — Multi-Step AI Agent Workflow Platform

import pytest
from langchain.agents import initialize_agent, AgentExecutor
from pydantic import BaseModel, ValidationError

class ToolCall(BaseModel):
    """Expected tool call in agent workflow."""
    tool_name: str
    expected_params: dict
    
class ToolCallValidator:
    def __init__(self):
        self.actual_calls = []
    
    def record_tool_call(self, tool_name: str, params: dict):
        """Record each tool call made by agent."""
        self.actual_calls.append({
            "tool": tool_name,
            "params": params
        })
    
    def validate_sequence(self, expected_calls: list[ToolCall]):
        """Validate tool calls match expected sequence."""
        assert len(self.actual_calls) == len(expected_calls), \
            f"Call count mismatch: expected {len(expected_calls)}, got {len(self.actual_calls)}"
        
        for i, (actual, expected) in enumerate(zip(self.actual_calls, expected_calls)):
            assert actual["tool"] == expected.tool_name, \
                f"Step {i}: expected tool {expected.tool_name}, got {actual['tool']}"
            
            # Validate key parameters match
            for param, value in expected.expected_params.items():
                assert param in actual["params"], \
                    f"Step {i}: missing parameter {param}"
                assert actual["params"][param] == value, \
                    f"Step {i}: param {param} mismatch"

@pytest.mark.asyncio
async def test_agent_research_workflow():
    """Test multi-step agent workflow for research task."""
    query = "Find recent AI safety regulations and summarize them."
    
    validator = ToolCallValidator()
    
    # Execute agent with tool call recording
    agent = initialize_agent(
        tools=research_tools,
        llm=gpt4,
        agent="zero-shot-react-description",
        callbacks=[ToolCallValidator.Callback(validator)]
    )
    
    result = await agent.arun(query)
    
    # Validate tool-call sequence
    expected_sequence = [
        ToolCall(tool_name="web_search", expected_params={"query": "AI safety regulations 2025"}),
        ToolCall(tool_name="fetch_webpage", expected_params={"url": "..."})  # Dynamic URLs
    ]
    
    validator.validate_sequence(expected_sequence)
    assert "regulation" in result.lower()

import pytest
import asyncio
from typing import Dict, Any

class AgentStateValidator:
    def __init__(self, max_steps: int = 20):
        self.max_steps = max_steps
        self.step_count = 0
        self.state_history = []
        self.tool_call_history = []
    
    async def validate_workflow_state(self, agent, query: str, expected_final_state: Dict[str, Any]):
        """Execute agent and validate state consistency throughout workflow."""
        self.step_count = 0
        self.state_history = []
        
        try:
            # Execute with step limit
            result = await asyncio.wait_for(
                agent.arun(query),
                timeout=60  # Prevent infinite loops
            )
        except asyncio.TimeoutError:
            pytest.fail(f"Agent exceeded timeout - possible infinite loop after {self.step_count} steps")
        
        # Verify step count reasonable
        assert self.step_count <= self.max_steps, \
            f"Agent took {self.step_count} steps (max {self.max_steps}) - possible loop"
        
        # Check for repeated tool calls (indicator of loop)
        tool_calls = [call["tool"] for call in self.tool_call_history]
        unique_calls = set(tool_calls)
        repeat_count = len(tool_calls) - len(unique_calls)
        
        assert repeat_count < 3, \
            f"Excessive tool call repetition detected: {repeat_count} duplicates"
        
        # Validate final state matches expectations
        for key, expected_value in expected_final_state.items():
            assert key in agent.memory.variables, f"Missing state key: {key}"
            assert agent.memory.variables[key] == expected_value, \
                f"State mismatch for {key}: expected {expected_value}, got {agent.memory.variables[key]}"
        
        return result

@pytest.mark.asyncio
async def test_agent_loop_detection():
    """Test that agent detects and breaks infinite loops."""
    validator = AgentStateValidator(max_steps=15)
    
    result = await validator.validate_workflow_state(
        agent=code_analysis_agent,
        query="Analyze this code and find bugs",
        expected_final_state={"analysis_complete": True, "bugs_found": 3}
    )
    
    assert "bug" in result.lower()
    assert validator.step_count <= 15