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.
In modern software development, designing architectures manually introduces several challenges:
Manually creating UML diagrams and translating them into structured code is inefficient and slows down development.
Human interpretation of diagrams can result in inconsistencies in implementation.
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.
Our framework consists of two AI agents, orchestrated using CrewAI, to ensure a sequential and structured workflow:
a. Converts business requirements into PlantUML class and sequence diagrams.
b. Ensures structured design principles, including proper function names, relationships, and standardized UML components.
a. Extracts class names, methods, and relationships from the UML diagrams.
b. Generates modular Python code following the extracted UML structure.
The system follows a sequential agent-based workflow:
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!
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.
The system produces standardized class and sequence diagrams.
Extracts key elements (class names, function names, relationships) from the UML diagrams.
Generates structured Python code files.
A fully structured codebase is generated based on the UML diagrams.
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 )
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)
Defines two tasks:
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' )
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()
The primary reason for using SerperDevTool is enhancing the AI agent's ability to produce high-quality, structured output. Specifically, it helps in:
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.
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.
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.
| Component | Purpose |
|---|---|
| UML Generator Agent | Uses SerperDevTool to search for best practices in PlantUML diagrams. |
| Code Generator Agent | Queries coding patterns to ensure generated Python code follows industry standards. |
| Secure API Key Handling | Ensures API keys are safely loaded from environment variables. |
For the above code block, this was what the agent has produced as an output:
@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
# 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> );
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
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.