Command the Crew

I'm a huge sci-fi fan, but while I'm fully onboard the Star Wars train (apologies to my fellow Trekkies!),
but I've always appreciated the classic episodes of Star Trek from my childhood.
The diverse crew of the USS Enterprise, each masterminding their unique roles, is a perfect metaphor for understanding AI agents and their power in projects like Facilis.
So, let's embark on an intergalactic mission, leveraging AI as our ship's crew and  boldly go where no man has gone before! 
This teamwork concept is a wonderful analogy to illustrate how AI agents work and how we use them in our DC-Facilis project. So, let’s dive in and assume the role of a starship captain, leading an AI crew into unexplored territories!

Welcome to CrewAI!

To manage our AI crew, we use a fantastic framework called CrewAI. It's lean, lightning-fast, and operates as a multi-agent Python platform. One of the reasons we love it, besides the fact that it was created by another Brazilian, is its incredible flexibility and role-based design.

from crewai import Agent, Task, Crew

Meet the Planners

In Facilis, our AI agents are divided into two groups. Let's start with the first one I like to call "The Planners."

The Extraction Agent

The main role of Facilis is to take a natural language description of a REST service and auto-magically create all the necessary interoperability. So, our first crew member is the Extraction Agent. This agent is tasked with "extracting" API specifications from a user's prompt description.

Here's what the Extraction Agent looks out for:

• Host (required)
• Endpoint (required)
• Params (optional)
• Port (if available)
• JSON model (for POST/PUT/PATCH/DELETE)
• Authentication (if applicable)

    def create_extraction_agent(self) -> Agent:
        return Agent(
            role='API Specification Extractor',
            goal='Extract API specifications from natural language descriptions',
            backstory=dedent("""
                You are specialized in interpreting natural language descriptions
                and extracting structured API specifications.
            """),
            allow_delegation=True,
            llm=self.llm
        )

    def extract_api_specs(self, descriptions: List[str]) -> Task:
        return Task(
            description=dedent(f"""
                Extract API specifications from the following descriptions:
                {json.dumps(descriptions, indent=2)}

                For each description, extract:
                - host (required)
                - endpoint (required)
                - HTTP_Method (required)
                - params (optional)
                - port (if available)
                - json_model (for POST/PUT/PATCH/DELETE)
                - authentication (if applicable)

                Mark any missing required fields as 'missing'.
                Return results in JSON format as an array of specifications.
            """),
            expected_output="""A JSON array containing extracted API specifications with all required and optional fields""",
            agent=self.extraction_agent
        )

The Validation Agent

Next up, the Validation Agent! Their mission is to ensure that the API specifications gathered by the Extraction Agent are correct and consistent. They check:

1. Valid host format
2. Endpoint starting with '/'
3. Valid HTTP methods (GET, POST, PUT, DELETE, PATCH)
4. Valid port number (if provided)
5. JSON model presence for applicable methods.

    def create_validation_agent(self) -> Agent:
        return Agent(
            role='API Validator',
            goal='Validate API specifications for correctness and consistency',
            backstory=dedent("""
                You are an expert in API validation, ensuring all specifications
                meet the required standards and format.
            """),
            allow_delegation=False,
            llm=self.llm
        )

 def validate_api_spec(self, extracted_data: Dict) -> Task:
        return Task(
            description=dedent(f"""
                Validate the following API specification:
                {json.dumps(extracted_data, indent=2)}

                Check for:
                1. Valid host format
                2. Endpoint starts with '/'
                3. Valid HTTP method (GET, POST, PUT, DELETE, PATCH)
                4. Valid port number (if provided)
                5. JSON model presence for POST/PUT/PATCH/DELETE methods

                Return validation results in JSON format.
            """),
            expected_output="""A JSON object containing validation results with any errors or confirmation of validity""",
            agent=self.validation_agent
        )

The Interaction Agent

Moving on, we meet the Interaction Agent, our User Interaction Specialist. Their role is to obtain any missing API specification fields that were marked by the Extraction Agent and validate them based on the Validation Agent's findings. They interact directly with users to fill any gaps.

The Production Agent

We need two crucial pieces of information to create the necessary interoperability: namespace and production name. The Production Agent engages with users to gather this information, much like the Interaction Agent.

The Documentation Transformation Agent

Once the specifications are ready, it's time to convert them into OpenAPI documentation. The Documentation Transformation Agent, an OpenAPI specialist, takes care of this.

    def create_transformation_agent(self) -> Agent:
        return Agent(
            role='OpenAPI Transformation Specialist',
            goal='Convert API specifications into OpenAPI documentation',
            backstory=dedent("""
                You are an expert in OpenAPI specifications and documentation.
                Your role is to transform validated API details into accurate
                and comprehensive OpenAPI 3.0 documentation.
            """),
            allow_delegation=False,
            llm=self.llm
        )

    def transform_to_openapi(self, validated_endpoints: List[Dict], production_info: Dict) -> Task:
        return Task(
            description=dedent(f"""
                Transform the following validated API specifications into OpenAPI 3.0 documentation:

                Production Information:
                {json.dumps(production_info, indent=2)}

                Validated Endpoints:
                {json.dumps(validated_endpoints, indent=2)}

                Requirements:
                1. Generate complete OpenAPI 3.0 specification
                2. Include proper request/response schemas
                3. Document all parameters and request bodies
                4. Include authentication if specified
                5. Ensure proper path formatting

                Return the OpenAPI specification in both JSON and YAML formats.
            """),
            expected_output="""A JSON object containing the complete OpenAPI 3.0 specification with all endpoints and schemas""",
            agent=self.transformation_agent
        )

The Review Agent

After transformation, the OpenAPI documentation undergoes a meticulous review to ensure compliance and quality. The Review Agent follows this checklist:

1. OpenAPI 3.0 Compliance
   • Correct version specification
   • Required root elements
   • Schema structure validation
2. Completeness
   • All endpoints documented
   • Fully specified parameters
   • Defined request/response schemas
   • Properly configured security schemes
3. Quality Checks
   • Consistent naming conventions
   • Clear descriptions
   • Proper use of data types
   • Meaningful response codes
4. Best Practices
   • Proper tag usage
   • Consistent parameter naming
   • Appropriate security definitions

Finally, if everything looks good, the Review Agent reports a healthy JSON object with the following structure:

{
 "is_valid": boolean,
 "approved_spec": object (the reviewed and possibly corrected OpenAPI spec),
 "issues": [array of strings describing any issues found],
 "recommendations": [array of improvement suggestions]
}

    def create_reviewer_agent(self) -> Agent:
        return Agent(
            role='OpenAPI Documentation Reviewer',
            goal='Ensure OpenAPI documentation compliance and quality',
            backstory=dedent("""
                You are the final authority on OpenAPI documentation quality and compliance.
                With extensive experience in OpenAPI 3.0 specifications, you meticulously
                review documentation for accuracy, completeness, and adherence to standards.
            """),
            allow_delegation=True,
            llm=self.llm
        )


    def review_openapi_spec(self, openapi_spec: Dict) -> Task:
        return Task(
            description=dedent(f"""
                Review the following OpenAPI specification for compliance and quality:

                {json.dumps(openapi_spec, indent=2)}

                Review Checklist:
                1. OpenAPI 3.0 Compliance
                - Verify correct version specification
                - Check required root elements
                - Validate schema structure

                2. Completeness
                - All endpoints properly documented
                - Parameters fully specified
                - Request/response schemas defined
                - Security schemes properly configured

                3. Quality Checks
                - Consistent naming conventions
                - Clear descriptions
                - Proper use of data types
                - Meaningful response codes

                4. Best Practices
                - Proper tag usage
                - Consistent parameter naming
                - Appropriate security definitions

                You must return a JSON object with the following structure:
                {{
                    "is_valid": boolean,
                    "approved_spec": object (the reviewed and possibly corrected OpenAPI spec),
                    "issues": [array of strings describing any issues found],
                    "recommendations": [array of improvement suggestions]
                }}
            """),
            expected_output="""A JSON object containing: is_valid (boolean), approved_spec (object), issues (array), and recommendations (array)""",
            agent=self.reviewer_agent
        )

The Iris Agent

The last agent in the planner group is the Iris Agent, who sends the finalized OpenAPI documentation to Iris.

    def create_iris_i14y_agent(self) -> Agent:
        return Agent(
            role='Iris I14y Integration Specialist',
            goal='Integrate API specifications with Iris I14y service',
            backstory=dedent("""
                You are responsible for ensuring smooth integration between the API
                documentation system and the Iris I14y service. You handle the
                communication with Iris, validate responses, and ensure successful
                integration of API specifications.
            """),
            allow_delegation=False,
            llm=self.llm
        )

    def send_to_iris(self, openapi_spec: Dict, production_info: Dict, review_result: Dict) -> Task:
        return Task(
            description=dedent(f"""
                Send the approved OpenAPI specification to Iris I14y service:

                Production Information:
                - Name: {production_info['production_name']}
                - Namespace: {production_info['namespace']}
                - Is New: {production_info.get('create_new', False)}

                Review Status:
                - Approved: {review_result['is_valid']}

                Return the integration result in JSON format.
            """),
            expected_output="""A JSON object containing the integration result with Iris I14y service, including success status and response details""",
            agent=self.iris_i14y_agent
        )

class IrisI14yService:
    def __init__(self):
        self.logger = logging.getLogger('facilis.IrisI14yService')
        self.base_url = os.getenv("FACILIS_URL", "http://dc-facilis-iris-1:52773") 
        self.headers = {
            "Content-Type": "application/json"
        }
        self.timeout = int(os.getenv("IRIS_TIMEOUT", "504"))  # in milliseconds
        self.max_retries = int(os.getenv("IRIS_MAX_RETRIES", "3"))
        self.logger.info("IrisI14yService initialized")

    async def send_to_iris_async(self, payload: Dict) -> Dict:
        """
        Send payload to Iris generate endpoint asynchronously
        """
        self.logger.info("Sending payload to Iris generate endpoint")
        if isinstance(payload, str):
            try:
                json.loads(payload)  
            except json.JSONDecodeError:
                raise ValueError("Invalid JSON string provided")

        retry_count = 0
        last_error = None

        # Create timeout for aiohttp
        timeout = aiohttp.ClientTimeout(total=self.timeout / 1000)  # Convert ms to seconds

        while retry_count < self.max_retries:
            try:
                self.logger.info(f"Attempt {retry_count + 1}/{self.max_retries}: Sending request to {self.base_url}/facilis/api/generate")

                async with aiohttp.ClientSession(timeout=timeout) as session:
                    async with session.post(
                        f"{self.base_url}/facilis/api/generate",
                        json=payload,
                        headers=self.headers
                    ) as response:
                        if response.status == 200:
                            return await response.json()
                        response.raise_for_status()

            except asyncio.TimeoutError as e:
                retry_count += 1
                last_error = e
                error_msg = f"Timeout occurred (attempt {retry_count}/{self.max_retries})"
                self.logger.warning(error_msg)

                if retry_count < self.max_retries:
                    wait_time = 2 ** (retry_count - 1)
                    self.logger.info(f"Waiting {wait_time} seconds before retry...")
                    await asyncio.sleep(wait_time)
                continue

            except aiohttp.ClientError as e:
                error_msg = f"Failed to send to Iris: {str(e)}"
                self.logger.error(error_msg)
                raise IrisIntegrationError(error_msg)

        error_msg = f"Failed to send to Iris after {self.max_retries} attempts due to timeout"
        self.logger.error(error_msg)
        raise IrisIntegrationError(error_msg, last_error)

Meet the Generators

Our second set of agents are - the Generators. They are here to transform the OpenAPI specifications into InterSystems IRIS interoperability.
There are eight of them in this group.

The first one is the Analyzer Agent. He's like the planner, mapping out the route.
Its job is to delve into the OpenAPI specs and figure out what IRIS Interoperability components are needed.

    def create_analyzer_agent():
        return Agent(
            role="OpenAPI Specification Analyzer",
            goal="Thoroughly analyze OpenAPI specifications and plan IRIS Interoperability components",
            backstory="""You are an expert in both OpenAPI specifications and InterSystems IRIS Interoperability. 
            Your job is to analyze OpenAPI documents and create a detailed plan for how they should be 
            implemented as IRIS Interoperability components.""",
            verbose=False,
            allow_delegation=False,
            tools=[analyze_openapi_tool],
            llm=get_facilis_llm()
        )

   analysis_task = Task(
        description="""Analyze the OpenAPI specification and plan the necessary IRIS Interoperability components. 
        Include a list of all components that should be in the Production class.""",
        agent=analyzer,
        expected_output="A detailed analysis of OpenAPI spec and plan for IRIS components, including Production components list",
        input={
            "openapi_spec": openApiSpec,
            "production_name": "${production_name}" 
        }
    )

Next up, the Business Services (BS) and Business Operations (BO) Agents take over.
They generate the Business Services and Business Operations based on the OpenAPI endpoints.
They use a handy tool called MessageClassTool to generate the perfect message classes, ensuring the communication.

    def create_bs_generator_agent():
        return Agent(
            role="IRIS Production and Business Service Generator",
            goal="Generate properly formatted IRIS Production and Business Service classes from OpenAPI specifications",
            backstory="""You are an experienced InterSystems IRIS developer specializing in Interoperability Productions.
            Your expertise is in creating Business Services and Productions that can receive and process incoming requests based on
            API specifications.""",
            verbose=False,
            allow_delegation=True,
            tools=[generate_production_class_tool, generate_business_service_tool],
            llm=get_facilis_llm()
        )

    def create_bo_generator_agent():
        return Agent(
            role="IRIS Business Operation Generator",
            goal="Generate properly formatted IRIS Business Operation classes from OpenAPI specifications",
            backstory="""You are an experienced InterSystems IRIS developer specializing in Interoperability Productions.
            Your expertise is in creating Business Operations that can send requests to external systems
            based on API specifications.""",
            verbose=False,
            allow_delegation=True,
            tools=[generate_business_operation_tool, generate_message_class_tool],
            llm=get_facilis_llm()
        )

    bs_generation_task = Task(
        description="Generate Business Service classes based on the OpenAPI endpoints",
        agent=bs_generator,
        expected_output="IRIS Business Service class definitions",
        context=[analysis_task]
    )

    bo_generation_task = Task(
        description="Generate Business Operation classes based on the OpenAPI endpoints",
        agent=bo_generator,
        expected_output="IRIS Business Operation class definitions",
        context=[analysis_task]
    )

    class GenerateMessageClassTool(BaseTool):
        name: str = "generate_message_class"
        description: str = "Generate an IRIS Message class"
        input_schema: Type[BaseModel] = GenerateMessageClassToolInput

        def _run(self, message_name: str, schema_info: Union[str, Dict[str, Any]]) -> str:
            writer = IRISClassWriter()
            try:
                if isinstance(schema_info, str):
                    try:
                        schema_dict = json.loads(schema_info)
                    except json.JSONDecodeError:
                        return "Error: Invalid JSON format for schema info"
                else:
                    schema_dict = schema_info

                class_content = writer.write_message_class(message_name, schema_dict)
                # Store the generated class
                writer.generated_classes[f"MSG.{message_name}"] = class_content
                return class_content
            except Exception as e:
                return f"Error generating message class: {str(e)}"

Once BS and BO have done their thing, it's time for the Production Agent to shine!
This agent pulls everything together to create a cohesive production environment.

After everything is set up, next in line is the Validation Agent.
This one makes comes in for a final checkup, making sure each Iris class is ok.

Then we have the Export Agent and the Collection Agent. The Export Agent generates the .cls files while the Collection Agent gathers all the file names.
Everything gets passed along to the importer, which compiles everything into InterSystems Iris.

    def create_exporter_agent():
        return Agent(
            role="IRIS Class Exporter",
            goal="Export and validate IRIS class definitions to proper .cls files",
            backstory="""You are an InterSystems IRIS deployment specialist. Your job is to ensure 
            that generated IRIS class definitions are properly exported as valid .cls files that 
            can be directly imported into an IRIS environment.""",
            verbose=False,
            allow_delegation=False,
            tools=[export_iris_classes_tool, validate_iris_classes_tool],
            llm=get_facilis_llm()
        )

    def create_collector_agent():
        return Agent(
            role="IRIS Class Collector",
            goal="Collect all generated IRIS class files into a JSON collection",
            backstory="""You are a file system specialist responsible for gathering and 
            organizing generated IRIS class files into a structured collection.""",
            verbose=False,
            allow_delegation=False,
            tools=[CollectGeneratedFilesTool()],
            llm=get_facilis_llm()
        )

    export_task = Task(
        description="Export all generated IRIS classes as valid .cls files",
        agent=exporter,
        expected_output="Valid IRIS .cls files saved to output directory",
        context=[bs_generation_task, bo_generation_task],
        input={
            "output_dir": "/home/irisowner/dev/output/iris_classes"  # Optional
        }
    )

    collection_task = Task(
        description="Collect all generated IRIS class files into a JSON collection",
        agent=collector,
        expected_output="JSON collection of all generated .cls files",
        context=[export_task, validate_task],
        input={
            "directory": "./output/iris_classes"
        }
    )

Limitations and Challenges

Our project started as an exciting experiment, where my fellow musketeers and I aimed to create a fully automated tool using agents.
It was a wild ride!
Our main focus was on REST API integrations. It's always a joy to get a task with an OpenAPI specification to integrate; however, legacy systems can be a whole different story.
We thought automating these tasks could be incredibly useful.
But every adventure has its twists:
One of the biggest challenges was instructing the AI to convert OpenAPI to Iris Interoperability.
We started with openAI GPT3.5-turbo model, on initial tests proved difficult with debugging and preventing breaks.
Switching to Anthropic Claude 3.7 Sonnet showed better results for the Generator group but not so much for the Planners...
This led us to split our environment configurations, using different LLM providers for flexibility.
We used GPT3.5-turbo for planning and Claude sonnet for generation, great combo!
This combination worked well but we did encounter issues with hallucinations.
Moving to GT4o improved results, yet we still faced hallucinations creating Iris classes and sometimes unnecessary OpenAPI specifications like the renowned Pet Store OpenAPI example.
we had a blast learning along the way, and I'm super excited about the amazing future in this field with countless possibilities!