# 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. ```python 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. ```python 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 ```python 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. ```python 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. | 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. | ## Example output: For the above code block, this was what the agent has produced as an output: ## uml_diagram.txt ```plantuml @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 } 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 } 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) ``` ```javascript // 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 (

Login

); }; export default Login; ``` ```javascript // 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 (

Signup

); }; export default Signup; ``` ```javascript // src/App.js import React from 'react'; import Login from './components/Login'; import Signup from './components/Signup'; function App() { return (

); } export default App; ``` ```javascript // 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( ); ``` # 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.