Abstract

Software development often requires structured architectural documentation, such as UML diagrams, to ensure clarity and maintainability. However, manually creating UML diagrams and translating them into code is a time-consuming process prone to inconsistencies. This paper presents an AI-driven, agentic framework leveraging CrewAI to automate the generation of PlantUML class and sequence diagrams based on business requirements and transform these diagrams into structured programming code. By integrating Large Language Models (LLMs) with AI agents, this approach streamlines the development lifecycle, ensuring consistency, efficiency, and adherence to software design principles.

1. Introduction: The Problem Statement

In modern software development, designing architectures manually introduces several challenges:

1. Time-Intensive Process:

Manually creating UML diagrams and translating them into structured code is inefficient and slows down development.

2. Inconsistencies Between Design and Implementation:

Human interpretation of diagrams can result in inconsistencies in implementation.

3. Lack of Automation:

Without an automated link between architectural design and code generation, manual intervention is required, increasing the risk of errors and inefficiencies.

To address these challenges, we propose an AI-powered, agentic approach to fully automate the process from business requirement gathering to UML diagram creation and structured code generation.

2. Proposed Solution

Our framework consists of two AI agents, orchestrated using CrewAI, to ensure a sequential and structured workflow:

UML Generator Agent:

a. Converts business requirements into PlantUML class and sequence diagrams.
b. Ensures structured design principles, including proper function names, relationships, and standardized UML components.

Code Generator Agent:

a. Extracts class names, methods, and relationships from the UML diagrams.
b. Generates modular Python code following the extracted UML structure.

Key Features of the Framework:

1. End-to-End Automation: Seamlessly converts textual business requirements into UML diagrams and corresponding code.
2. Standardized Software Design Practices: Ensures uniformity in function names, class structures, and flow.
3. Structured Code Organization: Generates modular files (app.py, controller.py, services.py, repository.py) for clean and maintainable architecture.

3. System Architecture

The system follows a sequential agent-based workflow:

User Input:

The user provides a high-level business requirement, such as "Create a To-do React application with a FastAPI backend." or it can be anything that a user can ask for!

PlantUML Generator Agent:

Converts the input into structured class and sequence diagrams.
Ensures the correct usage of alt, group, and activate/deactivate for proper representation of try-except logic.

Generated UML Diagrams:

The system produces standardized class and sequence diagrams.

Code Generator Agent:

Extracts key elements (class names, function names, relationships) from the UML diagrams.
Generates structured Python code files.

Final Output:

A fully structured codebase is generated based on the UML diagrams.

4. Implementation Details

4.1 Agent Definition (agents.py)

This file defines two agents:

PlantUML Generator Agent: Responsible for generating class and sequence diagrams.
Code Generator Agent: Responsible for parsing the UML diagrams and generating corresponding Python code.

from crewai import Agent, LLM
from tools import tool
from dotenv import load_dotenv
import os
import plantuml

load_dotenv()
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")

llm = LLM(model="gemini/gemini-2.0-flash-exp", temperature=0.2, verbose=True)

uml_generator = Agent(
    role="PlantUML Designer",
    goal="Generate PlantUML class and sequence diagrams for {topic}",
    verbose=True,
    memory=True,
    backstory="An AI-driven UML expert skilled at designing detailed architecture diagrams.",
    tools=[tool],
    llm=llm,
    allow_delegation=True
)

def code_generator_agent(context, *args, **kwargs):
    uml_output = context.output
    print("Generated UML Output:\n", uml_output)

    try:
        diagram = plantuml.PlantUML(uml_output)
        class_names = [c.name for c in diagram.classes]
        function_names = [m.name for c in diagram.classes for m in c.methods]

        print("Extracted Class Names:", class_names)
        print("Extracted Function Names:", function_names)

    except plantuml.PlantUMLParseError as e:
        print(f"Error parsing PlantUML: {e}")
        return "Error: Could not parse PlantUML"

    generated_code = ""
    for class_name in class_names:
        generated_code += f"class {class_name}:\n"
        for function_name in function_names:
            generated_code += f"    def {function_name}(self):\n"
            generated_code += f"        try:\n"
            generated_code += f"            # Implement function logic here\n"
            generated_code += f"            pass\n"
            generated_code += f"        except Exception as e:\n"
            generated_code += f"            print(f'Error in {function_name}: {{e}}')\n"
            generated_code += f"            raise\n"
        generated_code += "\n"

    return generated_code

code_generator = Agent(
    role="Senior Software Engineer",
    goal="Generate structured Python code based on UML diagrams for {topic}",
    verbose=True,
    memory=True,
    tools=[tool],
    llm=llm,
    agent_function=code_generator_agent
)

4.2 Crew Orchestration (crew.py)

The agents are orchestrated sequentially using CrewAI.

from crewai import Crew, Process
from tasks import uml_task, code_task
from agents import uml_generator, code_generator

crew = Crew(
    agents=[uml_generator, code_generator],
    tasks=[uml_task, code_task],
    process=Process.sequential,
)

result = crew.kickoff(inputs={'topic': 'Create a To-do React application with FastAPI backend'})
print(result)

4.3 Task Definition (tasks.py)

Defines two tasks:

1. Generating UML diagrams
2. Generating corresponding code

from crewai import Task
from tools import tool
from agents import uml_generator, code_generator

uml_task = Task(
    description=(
        "Generate Plantuml class and sequence diagram {topic}."
    "The class diagram should have the classes App, Controller, Services, Repository,"
    "The function names should follow camelCasing,"
    "The same set of functions generated in the class diagram should be used for generating the sequence diagram,"
    "The sequence diagram should contain the actor user and participants as app, controller, services and repository,"
    "Use autonumber 1.0 so that the count of sequence lines will be there, use group keyword for functions, use alt keyword for each function's try and except block, use proper activate and deactivate,"
    "Each sequence line should have more than 20 words describing the entire logic (code-wise and it's explanation as well) for each and every sequence lines,"
    "Mandatory use of alt try and except for all the grouped functions,"
    "The sequence diagram should start from the user level and the diagram should cover every logic till the repository and how the data/output gets returned back to the user,"
    "App file logic: has the routing function (just gets the input and passes it to controller function),"
    "Controller file logic: Receives the input and validates it and if so, passes it to services function,"
    "Services file logic: This is where the actual business logic happens and passes to repository (if there any any database related operations),"
    "Repository file logic: Database related operations will happen here and returns the result back to services function."
    ""
    ),
    expected_output="PlantUML code",
    tools=[tool],
    agent=uml_generator,
    output_file='uml_diagram.txt'
)

code_task = Task(
    description=(
        "Generate Python code for {topic} using the UML diagram structure. "
        "Follow proper try-except handling and maintain function integrity."
    ),
    context=[uml_task],
    expected_output="Structured Python code",
    tools=[tool],
    agent=code_generator,
    output_file='code.txt'
)

4.4 Tool Definition (tools.py)

This file sets up an external search tool to assist the agents in retrieving relevant information when generating PlantUML diagrams and corresponding code.

from crewai_tools import SerperDevTool
from dotenv import load_dotenv
load_dotenv()
import os

os.environ['SERPER_API_KEY'] = os.getenv('SERPER_API_KEY')
tool = SerperDevTool()

Why is SerperDevTool Used in this Project?

The primary reason for using SerperDevTool is enhancing the AI agent's ability to produce high-quality, structured output. Specifically, it helps in:

1. Improving UML Generation

When the UML Generator Agent creates class and sequence diagrams, it may need external references to ensure:
a. Correct UML syntax.
b. Standardized class structures.
c. Best practices in software design.

2. Enhancing Code Generation

The Code Generator Agent ensures that the generated code adheres to industry standards.
It can retrieve:
a. Common design patterns for structuring
application layers.
b. Optimized function implementations.
c. Error handling best practices in Python.

3. Maintaining Accuracy and Relevance

Unlike traditional AI-generated output, which relies only on pre-trained knowledge, SerperDevTool allows agents to:
a. Query live data and fetch up-to-date knowledge.
b. Avoid hallucinations in UML and code generation.
c. Ensure consistency with modern development practices.

Example output:

For the above code block, this was what the agent has produced as an output:

uml_diagram.txt

@startuml
class App {
  +configureRoutes()
}

class Controller {
  +registerUser(request: Request) : Response
  +loginUser(request: Request) : Response
  +createTask(request: Request) : Response
  +updateTask(request: Request) : Response
  +deleteTask(taskId: int) : Response
  +getTasks(userId: int) : Response
}

class Services {
  +registerUser(userData: dict) : User
  +loginUser(userData: dict) : User
  +createTask(taskData: dict) : Task
  +updateTask(taskId: int, taskData: dict) : Task
  +deleteTask(taskId: int) : bool
  +getTasks(userId: int) : list<Task>
}

class Repository {
  +createUser(userData: dict) : User
  +getUserByEmail(email: str) : User
  +createTask(taskData: dict) : Task
  +updateTask(taskId: int, taskData: dict) : Task
  +deleteTask(taskId: int) : bool
  +getTasksByUserId(userId: int) : list<Task>
}

App -- Controller
Controller -- Services
Services -- Repository

@enduml

@startuml
autonumber 1.0
actor User
participant App
participant Controller
participant Services
participant Repository

User -> App: Initiates the task creation process by sending a request with task details (title, description, due date) through the user interface.
activate App
App -> Controller: Routes the request to the createTask function within the Controller, passing the task details received from the user.
activate Controller
Controller -> Controller: Validates the received task details, checking for required fields and data types to ensure data integrity before processing.
Controller -> Services: If the validation is successful, the Controller invokes the createTask function in the Services layer, forwarding the validated task details.
activate Services

group Function Logic: createTask
    Services -> Services: Performs business logic, such as checking user permissions, sanitizing input, and preparing the data for persistence in the database.
    Services -> Repository: Calls the createTask function in the Repository layer, providing the prepared task data to be stored in the database.
    activate Repository
    Repository -> Repository: Executes the database insertion operation, creating a new task record with the provided data and handling potential database errors.
    alt Successful Task Creation
        Repository --> Services: Returns the newly created task object, including the generated task ID and creation timestamp, confirming successful persistence.
        deactivate Repository
    else Database Error
        Repository -->> Services: Returns an error message or exception indicating the failure to create the task in the database due to connectivity or data integrity issues.
        deactivate Repository
    end

    Services --> Controller: Returns the created task object or the error message back to the Controller, depending on the outcome of the repository operation.
    deactivate Services
end

alt Task Creation Success
    Controller --> App: Returns the created task object to the App layer, indicating that the task was successfully created and stored in the database.
    deactivate Controller
    App --> User: Updates the user interface with the newly created task details, providing visual confirmation of the successful task creation.
    deactivate App
else Task Creation Failure
    Controller -->> App: Returns an error message to the App layer, indicating that the task creation failed due to validation errors or database issues.
    deactivate Controller
    App -->> User: Displays an error message to the user, informing them about the failure to create the task and providing guidance on how to resolve the issue.
    deactivate App
end

@enduml

code.txt

# repository.py

from typing import List, Optional
from fastapi import HTTPException
# Assuming you are using SQLAlchemy for database interaction
from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import func
import os

# Database URL (replace with your actual database URL)
DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///./test.db")

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

# Define User model

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    tasks = relationship("Task", back_populates="owner")


# Define Task model

class Task(Base):
    __tablename__ = "tasks"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String)
    description = Column(String, nullable=True)
    due_date = Column(DateTime, nullable=True)
    completed = Column(Boolean, default=False)
    owner_id = Column(Integer, ForeignKey("users.id"))
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    owner = relationship("User", back_populates="tasks")


Base.metadata.create_all(bind=engine)


class Repository:
    def __init__(self, db: SessionLocal):
        self.db = db

    def createUser(self, userData: dict) -> User:
        try:
            hashed_password = userData["hashed_password"]  # Assuming password hashing is done in services
            db_user = User(email=userData["email"], hashed_password=hashed_password)
            self.db.add(db_user)
            self.db.commit()
            self.db.refresh(db_user)
            return db_user
        except Exception as e:
            self.db.rollback()
            raise HTTPException(status_code=500, detail=f"Failed to create user: {str(e)}")

    def getUserByEmail(self, email: str) -> Optional[User]:
        try:
            return self.db.query(User).filter(User.email == email).first()
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Failed to get user by email: {str(e)}")

    def createTask(self, taskData: dict) -> Task:
        try:
            db_task = Task(**taskData)
            self.db.add(db_task)
            self.db.commit()
            self.db.refresh(db_task)
            return db_task
        except Exception as e:
            self.db.rollback()
            raise HTTPException(status_code=500, detail=f"Failed to create task: {str(e)}")

    def updateTask(self, taskId: int, taskData: dict) -> Task:
        try:
            db_task = self.db.query(Task).filter(Task.id == taskId).first()
            if db_task is None:
                raise HTTPException(status_code=404, detail="Task not found")

            for key, value in taskData.items():
                setattr(db_task, key, value)

            self.db.commit()
            self.db.refresh(db_task)
            return db_task
        except Exception as e:
            self.db.rollback()
            raise HTTPException(status_code=500, detail=f"Failed to update task: {str(e)}")

    def deleteTask(self, taskId: int) -> bool:
        try:
            db_task = self.db.query(Task).filter(Task.id == taskId).first()
            if db_task is None:
                raise HTTPException(status_code=404, detail="Task not found")

            self.db.delete(db_task)
            self.db.commit()
            return True
        except Exception as e:
            self.db.rollback()
            raise HTTPException(status_code=500, detail=f"Failed to delete task: {str(e)}")

    def getTasksByUserId(self, userId: int) -> List[Task]:
        try:
            return self.db.query(Task).filter(Task.owner_id == userId).all()
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Failed to get tasks: {str(e)}")

# services.py

from typing import List
from fastapi import HTTPException
from repository import Repository, User, Task
from passlib.context import CryptContext

# Password hashing context
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class Services:
    def __init__(self, repository: Repository):
        self.repository = repository

    def registerUser(self, userData: dict) -> User:
        try:
            # Hash the password
            hashed_password = pwd_context.hash(userData["password"])
            userData["hashed_password"] = hashed_password
            del userData["password"]  # Remove plain password

            # Create user in the repository
            return self.repository.createUser(userData)
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Failed to register user: {str(e)}")

    def loginUser(self, userData: dict) -> User:
        try:
            email = userData["email"]
            password = userData["password"]
            user = self.repository.getUserByEmail(email)

            if not user:
                raise HTTPException(status_code=400, detail="Incorrect email or password")

            # Verify password
            if not pwd_context.verify(password, user.hashed_password):
                raise HTTPException(status_code=400, detail="Incorrect email or password")

            return user
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Failed to login user: {str(e)}")

    def createTask(self, taskData: dict) -> Task:
        try:
            return self.repository.createTask(taskData)
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Failed to create task: {str(e)}")

    def updateTask(self, taskId: int, taskData: dict) -> Task:
        try:
            return self.repository.updateTask(taskId, taskData)
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Failed to update task: {str(e)}")

    def deleteTask(self, taskId: int) -> bool:
        try:
            return self.repository.deleteTask(taskId)
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Failed to delete task: {str(e)}")

    def getTasks(self, userId: int) -> List[Task]:
        try:
            return self.repository.getTasksByUserId(userId)
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Failed to get tasks: {str(e)}")

# controller.py

from typing import List
from fastapi import APIRouter, Depends, HTTPException, Request, Response
from sqlalchemy.orm import Session
from database import SessionLocal
from services import Services
from repository import User, Task
from pydantic import BaseModel

# Pydantic models for request validation
class UserSchema(BaseModel):
    email: str
    password: str

class TaskSchema(BaseModel):
    title: str
    description: str = None
    due_date: str = None
    owner_id: int

# Dependency to get the database session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

router = APIRouter()

@router.post("/register", response_model=User)
async def registerUser(request: Request, user: UserSchema, db: Session = Depends(get_db)):
    try:
        services = Services(repository=Repository(db))
        user_data = user.dict()
        return services.registerUser(user_data)
    except HTTPException as e:
        raise e
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Registration failed: {str(e)}")

@router.post("/login", response_model=User)
async def loginUser(request: Request, user: UserSchema, db: Session = Depends(get_db)):
    try:
        services = Services(repository=Repository(db))
        user_data = user.dict()
        return services.loginUser(user_data)
    except HTTPException as e:
        raise e
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Login failed: {str(e)}")

@router.post("/tasks", response_model=Task)
async def createTask(request: Request, task: TaskSchema, db: Session = Depends(get_db)):
    try:
        services = Services(repository=Repository(db))
        task_data = task.dict()
        return services.createTask(task_data)
    except HTTPException as e:
        raise e
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Task creation failed: {str(e)}")

@router.put("/tasks/{taskId}", response_model=Task)
async def updateTask(taskId: int, request: Request, task: TaskSchema, db: Session = Depends(get_db)):
    try:
        services = Services(repository=Repository(db))
        task_data = task.dict()
        return services.updateTask(taskId, task_data)
    except HTTPException as e:
        raise e
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Task update failed: {str(e)}")

@router.delete("/tasks/{taskId}")
async def deleteTask(taskId: int, db: Session = Depends(get_db)):
    try:
        services = Services(repository=Repository(db))
        result = services.deleteTask(taskId)
        if result:
            return {"message": "Task deleted successfully"}
        else:
            raise HTTPException(status_code=404, detail="Task not found")
    except HTTPException as e:
        raise e
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Task deletion failed: {str(e)}")

@router.get("/tasks/{userId}", response_model=List[Task])
async def getTasks(userId: int, db: Session = Depends(get_db)):
    try:
        services = Services(repository=Repository(db))
        return services.getTasks(userId)
    except HTTPException as e:
        raise e
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to get tasks: {str(e)}")

# app.py
from fastapi import FastAPI
from controller import router as api_router
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# Configure CORS
origins = [
    "http://localhost:3000",  # React app address
    "http://localhost:8000",  # FastAPI address
    "*", # Allow all origins (for development purposes only!)
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Include the API router
app.include_router(api_router)

# Configure routes (this function is not strictly necessary with APIRouter, but kept for compatibility)
def configureRoutes():
    pass

if __name__ == "__main__":
    # Run the app using Uvicorn
    import uvicorn
    configureRoutes()
    uvicorn.run(app, host="0.0.0.0", port=8000)

// src/components/Login.js
import React, { useState } from 'react';

const Login = () => {
  const [email, setEmail] = useState('');
  const [password, setPassword] = useState('');

  const handleSubmit = async (e) => {
    e.preventDefault();
    try {
      const response = await fetch('http://localhost:8000/login', {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
        },
        body: JSON.stringify({ email, password }),
      });

      if (response.ok) {
        const data = await response.json();
        console.log('Login successful:', data);
        // Store the token or user info in local storage/context
      } else {
        console.error('Login failed:', response.statusText);
        // Display an error message to the user
      }
    } catch (error) {
      console.error('Login error:', error);
      // Display an error message to the user
    }
  };

  return (
    <div>
      <h2>Login</h2>
      <form onSubmit={handleSubmit}>
        <div>
          <label htmlFor="email">Email:</label>
          <input
            type="email"
            id="email"
            value={email}
            onChange={(e) => setEmail(e.target.value)}
            required
          />
        </div>
        <div>
          <label htmlFor="password">Password:</label>
          <input
            type="password"
            id="password"
            value={password}
            onChange={(e) => setPassword(e.target.value)}
            required
          />
        </div>
        <button type="submit">Log In</button>
      </form>
    </div>
  );
};

export default Login;

// src/components/Signup.js
import React, { useState } from 'react';

const Signup = () => {
  const [email, setEmail] = useState('');
  const [password, setPassword] = useState('');

  const handleSubmit = async (e) => {
    e.preventDefault();
    try {
      const response = await fetch('http://localhost:8000/register', {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
        },
        body: JSON.stringify({ email, password }),
      });

      if (response.ok) {
        const data = await response.json();
        console.log('Signup successful:', data);
        // Redirect to login page or store user info
      } else {
        console.error('Signup failed:', response.statusText);
        // Display an error message to the user
      }
    } catch (error) {
      console.error('Signup error:', error);
      // Display an error message to the user
    }
  };

  return (
    <div>
      <h2>Signup</h2>
      <form onSubmit={handleSubmit}>
        <div>
          <label htmlFor="email">Email:</label>
          <input
            type="email"
            id="email"
            value={email}
            onChange={(e) => setEmail(e.target.value)}
            required
          />
        </div>
        <div>
          <label htmlFor="password">Password:</label>
          <input
            type="password"
            id="password"
            value={password}
            onChange={(e) => setPassword(e.target.value)}
            required
          />
        </div>
        <button type="submit">Sign Up</button>
      </form>
    </div>
  );
};

export default Signup;

// src/App.js
import React from 'react';
import Login from './components/Login';
import Signup from './components/Signup';

function App() {
  return (
    <div className="App">
      <Login />
      <Signup />
    </div>
  );
}

export default App;

// index.js
import React from 'react';
import ReactDOM from 'react-dom/client';
import App from './App';

const root = ReactDOM.createRoot(document.getElementById('root'));
root.render(
  <React.StrictMode>
    <App />
  </React.StrictMode>
);

Conclusion and Future Work

This framework successfully automates the transition from textual business requirements to structured UML diagrams and well-organized codebases. It ensures:

a. Consistency between software design and implementation
b. Reduced manual effort through automation
Standardized, maintainable, and modular code generation

Future enhancements could include:

a. Support for additional programming languages (e.g., Java, C++).
b. Enhanced validation mechanisms to improve UML-to-code accuracy.
c. Integration with DevOps pipelines for continuous deployment.

This agentic approach paves the way for intelligent, automated software engineering, minimizing human intervention and maximizing efficiency.