INTRODUCTION: THE IMPERATIVE FOR INTELLIGENT CODE REPOSITORY ANALYSIS
The software engineering landscape in 2026 presents unprecedented challenges for organizations managing large, complex codebases. Modern enterprise systems often span millions of lines of code across multiple programming languages, incorporate diverse architectural patterns, and evolve through contributions from distributed teams over many years.
Traditional code analysis tools, while valuable, struggle to provide the holistic understanding that stakeholders across the organization require to make informed decisions about system evolution, technical debt management, and strategic planning.
Contemporary AI-powered coding assistants such as GitHub Copilot, Cursor, Anthropic Claude Code, and Windsurf have revolutionized individual developer productivity by offering intelligent code completion, generation, and refactoring suggestions. These tools represent a significant leap forward from earlier integrated development environments. However, they exhibit fundamental limitations that prevent them from serving as comprehensive repository analysis solutions. Their context windows, while expanding to hundreds of thousands of tokens in models like Gemini 3 Pro, still cannot encompass entire enterprise codebases that may contain tens of millions of tokens. These assistants primarily focus on immediate coding tasks rather than cross-cutting concerns like architectural consistency, requirements traceability, or business goal alignment. They operate within constrained scopes, typically analyzing individual files or small clusters of related code, making them unsuitable for understanding system-wide patterns, dependencies, and quality attributes.
Furthermore, current AI coding assistants face persistent challenges with hallucinations, where they generate plausible but factually incorrect code or architectural suggestions. They lack the ability to validate their outputs against comprehensive factual knowledge about the specific codebase, its historical evolution, its architectural decisions, and its business context.
Security vulnerabilities in AI-generated code have become a documented concern, with studies indicating that code produced by AI assistants may contain more security flaws than human-written code, and conventional security scanning tools struggle to detect these AI-specific vulnerabilities.
The probabilistic nature of large language models means they excel at pattern matching but can fail at deep reasoning about complex system interactions, leading to what practitioners call "architectural hallucination" where suggested designs appear sound but are contextually inappropriate.
The limitations extend beyond technical capabilities. These tools provide no systematic approach to analyzing the alignment between business strategy, requirements, architecture, implementation, and testing. They cannot automatically generate comprehensive SWOT analyses that evaluate strengths, weaknesses, opportunities, and threats across all dimensions of a software system. They lack the ability to trace how business goals flow through requirements specifications into architectural decisions and ultimately into code implementations and test coverage. For senior management seeking to understand technical debt's business impact, for architects evaluating whether the system can support new strategic initiatives, for project managers assessing risk, and for operations teams understanding deployment complexities, current AI coding assistants provide insufficient insight.
This article presents a comprehensive approach to building a multi-agent AI system specifically designed to analyze large, polyglot code repositories with the depth and breadth that enterprise stakeholders require. Unlike single-purpose coding assistants, this system orchestrates specialized agents that collaboratively examine every facet of a software system, from its business context and strategic alignment through its requirements, architecture, design, implementation, testing, and operational characteristics. The system addresses the fundamental challenges of limited context windows through hierarchical summarization and intelligent information retrieval. It mitigates hallucination risks through rigorous fact-checking using retrieval-augmented generation with both vector databases and knowledge graphs. It provides transparency and traceability by maintaining explicit links between business goals, requirements, architectural decisions, code artifacts, and test cases.
THE FUNDAMENTAL ARCHITECTURE: ORCHESTRATING SPECIALIZED INTELLIGENCE
The multi-agent architecture represents a paradigm shift from monolithic analysis tools to a collaborative ecosystem of specialized intelligences. Each agent within the system possesses deep expertise in a specific domain, whether that domain is business strategy analysis, requirements engineering, architectural pattern recognition, code quality assessment, or operational readiness evaluation. This specialization allows each agent to employ domain-specific reasoning strategies, utilize tailored language models optimized for its particular tasks, and maintain focused knowledge bases that would be impractical to combine into a single monolithic system.
At the heart of this architecture lies an orchestration layer that coordinates agent activities, manages information flow between agents, handles error recovery, and optimizes resource allocation across the system. This orchestration layer does not simply execute agents sequentially but rather understands dependencies between different analysis tasks and can parallelize independent work streams to maximize efficiency. When the architecture analysis agent identifies a component with high cyclomatic complexity, the orchestrator can simultaneously dispatch the code quality agent to perform detailed analysis of that component while the test coverage agent examines whether adequate tests exist for those complex code paths.
The orchestration layer implements sophisticated communication protocols that allow agents to exchange not just raw data but rich contextual information. When the requirements analysis agent identifies a functional requirement that appears unimplemented, it communicates this finding to the architecture agent with full context about the requirement's source, its relationship to business goals, and its priority. The architecture agent can then investigate whether architectural provisions exist to support this requirement even if implementation is incomplete, providing a more nuanced analysis than either agent could achieve independently.
The system employs a hybrid architecture that combines local and remote large language model deployments to optimize for cost, latency, privacy, and capability. Certain tasks, particularly those involving sensitive proprietary code or requiring rapid iteration, execute using local models deployed on organizational infrastructure. These local deployments leverage the diverse GPU ecosystem available in 2026, including Nvidia CUDA-based systems that continue to dominate high-performance AI workloads, AMD ROCm platforms that offer compelling open-source alternatives and substantial memory capacity through MI355X and MI455X GPUs, Intel solutions utilizing IPEX-LLM for optimized inference on Arc, Flex, and Max GPUs, and Apple Silicon devices with their unified memory architecture that proves particularly effective for managing large models on M5 Ultra chips expected in 2026.
For tasks requiring the most advanced reasoning capabilities or the largest context windows, the system can selectively invoke remote models such as GPT-5.1 with its integrated reasoning engine and 200,000 token context window, Gemini 3 Pro with its exceptional multimodal understanding and up to one million token context capacity, or Claude Opus 4.5 Sonnet with its sustained performance on complex multi-step tasks and strong software engineering capabilities. The orchestration layer makes intelligent decisions about model selection based on task requirements, balancing capability against cost and latency constraints.
ADDRESSING CONTEXT LIMITATIONS THROUGH HIERARCHICAL KNOWLEDGE REPRESENTATION
The challenge of analyzing codebases that vastly exceed any model's context window requires a sophisticated approach to knowledge representation and retrieval. The system addresses this through a multi-layered strategy that combines hierarchical summarization, retrieval-augmented generation, and hybrid knowledge storage using both vector databases and knowledge graphs.
Hierarchical summarization begins at the most granular level of individual functions and methods. Each function is analyzed to extract its purpose, inputs, outputs, side effects, and key algorithmic approach. These function-level summaries are then aggregated to create class or module summaries that describe the overall responsibility, public interface, internal state management, and relationships to other components. Class summaries aggregate into component summaries that capture architectural intent, provided services, dependencies, and quality characteristics. Component summaries further aggregate into subsystem summaries that describe major architectural building blocks, and ultimately into a system-level summary that provides an executive overview of the entire codebase.
This hierarchical structure allows agents to navigate the codebase at appropriate levels of abstraction. When analyzing whether the system architecture supports a new business requirement, an agent can begin with subsystem-level summaries to identify relevant architectural areas, then drill down through component and class summaries to specific code implementations only when necessary. This approach dramatically reduces the amount of information that must fit within a context window at any given time while preserving the ability to access detailed information when needed.
The system stores code artifacts in a retrieval-augmented generation database using vector embeddings that capture semantic meaning. When an agent needs to find code related to a particular concept, such as authentication mechanisms or data validation logic, it can query the vector database using natural language descriptions. The database returns semantically similar code chunks ranked by relevance, allowing the agent to focus its analysis on the most pertinent code without exhaustively scanning the entire repository.
Complementing the vector database, a knowledge graph provides structured representation of code entities and their relationships. The graph contains nodes representing subsystems, components, classes, interfaces, services, functions, and methods. Edges capture relationships such as inheritance, composition, method invocation, data flow, and architectural dependencies. This graph structure enables powerful traversal queries that answer questions like "What are all the components that depend on this authentication service?" or "Which business requirements trace to code in this subsystem?" The knowledge graph also maintains links to the vector database, allowing seamless navigation from high-level structural understanding to detailed code analysis.
The integration of vector databases and knowledge graphs creates what practitioners call GraphRAG, a hybrid approach that combines the semantic search capabilities of vector embeddings with the relational reasoning power of graph structures. When analyzing a complex architectural question, an agent might first use vector search to identify relevant code areas based on semantic similarity, then leverage graph traversal to explore the relationships and dependencies of those areas, and finally retrieve detailed code from the vector database for in-depth analysis. This hybrid approach provides both the flexibility to handle unstructured queries and the precision to navigate complex structural relationships.
MITIGATING HALLUCINATIONS THROUGH RIGOROUS FACT-CHECKING AND VALIDATION
Hallucinations represent one of the most significant risks in LLM-based analysis systems. When an AI model generates a plausible but factually incorrect statement about a codebase, stakeholders making decisions based on that misinformation can suffer serious consequences. The multi-agent system employs multiple complementary strategies to detect and prevent hallucinations.
Every factual claim made by an agent must be grounded in verifiable evidence from the codebase or associated documentation. When the architecture analysis agent claims that a particular component implements a specific design pattern, it must reference the actual code structures that exhibit that pattern. The system maintains explicit links between analytical conclusions and the evidence supporting those conclusions, allowing human reviewers to verify claims and providing transparency into the reasoning process.
The system implements a validation agent whose specific responsibility is fact-checking the outputs of other agents. When an agent produces an analysis, the validation agent examines the claims made, retrieves the purported evidence from the knowledge graph and vector database, and verifies that the evidence actually supports the conclusions. If the validation agent detects inconsistencies, it flags them for human review and requests that the original agent revise its analysis with corrected information.
Cross-agent verification provides another layer of protection against hallucinations. When multiple agents analyze related aspects of the system, their conclusions should be mutually consistent. If the requirements agent identifies a functional requirement as fully implemented while the test coverage agent finds no tests for that functionality, this inconsistency triggers a review process where both agents re-examine their analyses and reconcile the discrepancy. This cross-checking leverages the principle that hallucinations are unlikely to be consistent across multiple independent reasoning processes.
The system employs chain-of-thought and tree-of-thought reasoning strategies that make the reasoning process explicit and reviewable. Rather than directly generating conclusions, agents articulate their reasoning steps, consider alternative interpretations, and evaluate the strength of evidence for different conclusions. This transparent reasoning allows both automated validation agents and human reviewers to identify logical flaws or unsupported leaps in reasoning that might indicate hallucinations.
For critical analyses where accuracy is paramount, the system implements a human-in-the-loop workflow where agents present their findings along with supporting evidence to human experts for validation before incorporating those findings into the final report. This approach recognizes that while AI can dramatically accelerate analysis, human judgment remains essential for high-stakes decisions.
SPECIALIZED AGENTS: DEEP EXPERTISE ACROSS THE SOFTWARE ENGINEERING LIFECYCLE
The power of the multi-agent architecture emerges from the specialized capabilities of individual agents, each designed to analyze a specific facet of the software system with depth that would be impossible in a generalist tool.
The business strategy agent analyzes documentation, meeting notes, and strategic planning materials to understand the organization's business goals, competitive positioning, and strategic initiatives. This agent identifies how the software system supports or constrains business objectives, evaluates alignment between technical capabilities and business needs, and highlights opportunities where technical improvements could enable new business capabilities. By understanding the business context, this agent provides the foundation for evaluating whether technical decisions serve organizational goals.
The requirements analysis agent processes requirements specifications, user stories, feature requests, and customer feedback to build a comprehensive understanding of what the system should do. This agent identifies functional requirements, quality attribute requirements, and constraints. It traces requirements to their sources, whether those sources are regulatory mandates, customer requests, or internal business needs. The agent evaluates requirements for completeness, consistency, and testability, flagging ambiguities or conflicts that could lead to implementation problems.
The architecture analysis agent examines architectural documentation, design documents, and Architecture Decision Records to understand the system's intended structure. It analyzes the code itself to determine the implemented architecture, comparing intended versus actual structure to identify architectural drift. This agent recognizes architectural patterns, evaluates whether the architecture supports required quality attributes like scalability and maintainability, and identifies architectural weaknesses such as circular dependencies, excessive coupling, or missing abstraction layers. It also analyzes how the architecture has evolved over time, identifying areas of architectural erosion where incremental changes have degraded structural integrity.
The code quality agent performs deep analysis of implementation quality using both static analysis and semantic understanding. It calculates complexity metrics, identifies code smells and anti-patterns, evaluates adherence to coding standards, and assesses maintainability characteristics. Unlike traditional static analysis tools that operate purely on syntactic patterns, this agent leverages language models to understand semantic issues like unclear variable names, inadequate comments, or convoluted logic that technically compiles but is difficult for humans to understand and maintain.
The domain modeling agent analyzes how the code represents domain concepts from both the problem domain and the solution domain. It identifies domain entities, their relationships, and the business rules governing them. This agent evaluates whether the domain model accurately reflects the business domain, whether it is implemented consistently across the codebase, and whether it provides appropriate abstractions. Understanding domain modeling is crucial because misalignment between the code's domain model and the actual business domain leads to bugs, maintenance difficulties, and limitations in system evolution.
The test and quality agent examines test suites, test results, and quality assurance processes. It evaluates test coverage, identifying code paths that lack adequate testing. It analyzes test quality, determining whether tests actually validate meaningful behaviors or merely achieve superficial coverage metrics. This agent correlates test results with code changes to identify fragile areas where small modifications frequently break tests, suggesting brittle designs. It also evaluates whether testing practices align with the system's quality attribute requirements.
The operations and deployment agent analyzes deployment configurations, infrastructure as code, monitoring setups, and operational documentation. It evaluates whether the system can be reliably deployed, monitored, and maintained in production environments. This agent identifies operational risks such as missing health checks, inadequate logging, or deployment processes that require manual intervention. It also analyzes operational data like incident reports and performance metrics to identify runtime issues that may not be apparent from code analysis alone.
The repository metadata agent analyzes version control history, issue tracking systems, and project management tools to understand how the codebase evolves. It identifies hotspots where code changes frequently, suggesting either active development areas or problematic code that requires repeated fixes. It analyzes commit patterns to understand team dynamics and knowledge distribution. It correlates code changes with issues and tickets to understand what problems drive modifications and whether fixes actually resolve the underlying issues.
The knowledge graph construction agent processes all information gathered by other agents to build and maintain the comprehensive knowledge graph that serves as the system's central knowledge repository. This agent performs named entity recognition on diverse document types, extracting entities like requirements, architectural components, business goals, and test cases. It identifies relationships between entities, creating the rich web of connections that enables sophisticated queries and traceability.
The report generation agent synthesizes analyses from all other agents into comprehensive, stakeholder-appropriate reports. This agent understands the information needs of different audiences, generating executive summaries for senior management, detailed technical analyses for architects and developers, and focused reports for specific concerns like security vulnerabilities or technical debt. It creates visualizations that make complex information accessible, generates traceability matrices that show relationships between requirements, architecture, code, and tests, and produces SWOT analyses that evaluate the system from multiple perspectives.
POLYGLOT CODE ANALYSIS: UNIFIED UNDERSTANDING ACROSS PROGRAMMING LANGUAGES
Enterprise codebases rarely limit themselves to a single programming language. A typical system might combine Python for data processing and machine learning components, Java for backend services, JavaScript or TypeScript for frontend applications, C++ for performance-critical components, Go for infrastructure tools, and SQL for database schemas. Analyzing such polyglot repositories requires parsing and understanding code across all these languages with consistent semantic representation.
The system leverages Tree-sitter, a parser generator and incremental parsing library that provides uniform abstract syntax tree representations across more than forty programming languages. Tree-sitter's error-tolerant parsing algorithm continues to produce useful results even when code contains syntax errors, which is essential for analyzing codebases under active development. Its incremental parsing capability allows the system to efficiently update its understanding when code changes, re-parsing only the modified portions rather than the entire codebase.
For Python code, the system uses Tree-sitter's Python grammar to parse source files into abstract syntax trees. These trees represent the hierarchical structure of the code, with nodes for modules, classes, functions, statements, and expressions. The system traverses these trees to extract semantic information like function signatures, class hierarchies, import dependencies, and control flow structures. By understanding the AST rather than treating code as text, the system can accurately identify constructs even when coding style varies across the codebase.
Java code undergoes similar parsing using Tree-sitter's Java grammar. The system extracts information about packages, classes, interfaces, methods, annotations, and type hierarchies. It understands Java-specific constructs like generics, lambda expressions, and stream operations. The system recognizes common Java frameworks and libraries, understanding how Spring dependency injection affects component relationships or how JPA annotations define data models.
For C++ code, the system parses headers and implementation files to understand class definitions, template instantiations, namespace organization, and complex type systems. It handles C++ specific challenges like template metaprogramming and multiple inheritance, building an accurate model of the code's structure despite the language's complexity.
JavaScript and TypeScript parsing extracts information about modules, functions, classes, and the prototype-based inheritance system. The system understands modern JavaScript features like async/await, destructuring, and arrow functions. For TypeScript, it additionally processes type annotations, interfaces, and type guards, using this type information to enhance its understanding of component contracts and data flow.
The system employs a query language based on Tree-sitter's S-expression syntax to identify specific patterns across all languages. A query might search for all function definitions that take more than a certain number of parameters, regardless of whether those functions are written in Python, Java, or C++. This uniform query capability enables cross-language analysis, such as identifying architectural patterns that span multiple languages or detecting inconsistencies in how similar concepts are implemented across different parts of the codebase.
Named entity recognition operates on the parsed AST representations to extract domain-specific entities. Different programming languages and frameworks require specialized NER implementations. When analyzing a Spring-based Java application, the NER system recognizes Spring annotations to identify services, repositories, and controllers. When analyzing a Python Flask application, it recognizes Flask decorators to identify routes and view functions. These language-specific NER implementations feed into a unified knowledge graph where entities from all languages coexist and relate to each other.
The system maintains separate but interconnected knowledge graphs for each programming language's specific constructs while also maintaining a language-agnostic architectural view. This allows agents to reason about language-specific details when necessary while also performing cross-language architectural analysis. An agent analyzing data flow might trace a request from a JavaScript frontend through a Java backend service to a Python data processing component, understanding how data transforms as it crosses language boundaries.
DOMAIN MODELING ANALYSIS: BRIDGING PROBLEM AND SOLUTION DOMAINS
Understanding how a software system models its domain is fundamental to evaluating its quality and evolution potential. The multi-agent system performs sophisticated domain modeling analysis that examines both the problem domain, which represents the real-world business concepts the system addresses, and the solution domain, which represents the technical constructs used to implement the system.
The problem domain agent analyzes requirements documents, business glossaries, and domain expert interviews to build a conceptual model of the business domain. This model identifies core domain entities like Customer, Order, Product, or Invoice in a business application, or Patient, Diagnosis, Treatment, and Medication in a healthcare system. The agent identifies relationships between entities, such as "a Customer places multiple Orders" or "a Treatment addresses a Diagnosis." It captures business rules like "an Order cannot be fulfilled until payment is confirmed" or "a Patient must have a primary care physician assigned."
The solution domain agent analyzes the codebase to understand how these problem domain concepts are represented in the implementation. It identifies classes, database tables, API resources, and other technical constructs that correspond to problem domain entities. It examines how relationships are implemented, whether through object references, foreign keys, or message passing. It evaluates how business rules are enforced, whether through validation logic, database constraints, or workflow orchestration.
The domain alignment agent compares the problem domain model with the solution domain implementation to identify mismatches. When a problem domain entity exists without corresponding implementation, this suggests missing functionality. When implementation constructs exist without clear problem domain correspondence, this suggests either technical infrastructure concerns or potentially unnecessary complexity. When the structure of relationships differs between problem and solution domains, this can indicate impedance mismatches that lead to bugs or make the system difficult to evolve.
Domain-driven design principles emphasize maintaining a ubiquitous language where the same terminology is used consistently across requirements, design, code, and communication. The domain modeling agents evaluate whether the codebase adheres to this principle. If requirements documents refer to "Customers" but the code uses "Users" or "Accounts," this terminology inconsistency can lead to confusion and errors. The agents identify such discrepancies and recommend standardization.
The system analyzes bounded contexts, which are explicit boundaries within which a particular domain model applies. In complex systems, the same term might mean different things in different contexts. "Customer" might mean something different to the sales team than to the support team. The agents identify these bounded contexts in the code, evaluating whether they are explicitly represented through modules, namespaces, or services, or whether they are implicit and potentially confused.
Strategic domain-driven design distinguishes between core domains that provide competitive advantage, supporting domains that are necessary but not differentiating, and generic domains that could be replaced with off-the-shelf solutions. The domain modeling agents classify components according to this taxonomy, helping stakeholders understand where to focus improvement efforts. Technical debt in core domain implementations poses greater risk than technical debt in generic domains that might eventually be replaced.
The agents also analyze how the domain model has evolved over time by examining version control history. They identify when new domain concepts were introduced, when existing concepts were modified or removed, and whether these changes aligned with evolving business requirements. Rapid changes to core domain models might indicate business uncertainty or requirements volatility, while stagnation in areas where the business is evolving suggests the code is falling behind business needs.
IMPLEMENTING RETRIEVAL-AUGMENTED GENERATION WITH GRAPHRAG
The system's ability to provide factually accurate, well-grounded analysis depends critically on its retrieval-augmented generation implementation. Rather than relying solely on the language model's parametric knowledge, which can be outdated or incorrect, the system retrieves relevant factual information from the codebase and documentation before generating analytical conclusions.
The vector database stores embeddings of code chunks, documentation sections, requirements statements, and other textual artifacts. These embeddings are generated using specialized models that understand code semantics, not just natural language. When a model like OpenAI's Codex or a code-specific variant of Gemini or Claude generates embeddings for code, it captures semantic meaning that goes beyond simple text similarity. Two functions that perform similar operations using different implementations will have similar embeddings, allowing the system to find semantically related code even when textual similarity is low.
The system chunks code at semantically meaningful boundaries rather than arbitrary character or line limits. A function or method forms a natural chunk, as does a class definition or a module. For longer classes or modules, the system may create overlapping chunks that preserve context across boundaries. Each chunk is embedded and stored in the vector database along with metadata indicating its location in the codebase, its programming language, and its relationships to other chunks.
When an agent needs to analyze a particular aspect of the system, it formulates a query describing what it seeks. This query might be "authentication and authorization mechanisms" or "data validation logic for user inputs." The vector database performs a similarity search, comparing the query embedding against all stored chunk embeddings and returning the most semantically similar chunks. The agent then examines these retrieved chunks to understand how the codebase implements the relevant functionality.
The knowledge graph complements vector search by providing structured navigation of relationships. While vector search excels at finding semantically similar content, graph traversal excels at following explicit relationships. If an agent identifies a service class through vector search, it can use the knowledge graph to find all classes that depend on that service, all interfaces it implements, all methods it provides, and all components that invoke those methods. This graph-based navigation provides comprehensive understanding of a component's role in the system architecture.
The GraphRAG approach integrates both retrieval mechanisms. An agent might begin with a vector search to identify relevant components, then use graph traversal to explore their relationships and dependencies, then retrieve additional code chunks for detailed analysis, and finally synthesize this information into analytical conclusions. Throughout this process, the agent maintains explicit links between its conclusions and the evidence retrieved from the vector database and knowledge graph.
The system implements community detection algorithms on the knowledge graph to identify clusters of closely related components. These communities often correspond to architectural modules or subsystems. By summarizing each community, the system can provide module-level abstractions that fit within context windows while preserving the ability to drill down into detailed component analysis when necessary.
The knowledge graph also maintains temporal information, recording when entities were created, modified, or deprecated. This temporal dimension allows agents to analyze how the system has evolved, identifying areas of rapid change that might indicate instability or active development, and areas of stagnation that might indicate technical debt or abandoned features.
MANAGING BIAS AND ENSURING FAIRNESS IN ANALYSIS
Large language models can exhibit various forms of bias stemming from their training data, which may include biased code examples, documentation, or technical discussions. When these models analyze codebases, their biases can manifest in subtle ways that affect the fairness and accuracy of analysis.
The system addresses bias through multiple complementary strategies. First, it employs diverse models from different providers for different analysis tasks. GPT-5.1, Gemini 3 Pro, and Claude Opus 4.5 Sonnet are trained on different datasets using different methodologies, leading to different bias profiles. By using multiple models and comparing their analyses, the system can identify cases where bias might be influencing conclusions. When different models reach inconsistent conclusions about the same code, this triggers additional scrutiny to determine which analysis is more accurate and whether bias might be affecting any of the models.
The system maintains awareness of common bias patterns in technical analysis. Language models may exhibit bias toward certain programming paradigms, favoring object-oriented approaches over functional programming or vice versa. They may favor certain frameworks or libraries based on their prevalence in training data rather than their appropriateness for the specific context. The validation agent is specifically trained to recognize these bias patterns and flag analyses that might be unduly influenced by them.
For analyses that involve human factors, such as evaluating code quality or architectural elegance, the system recognizes that these judgments can be subjective and potentially biased. Rather than presenting a single model's opinion as objective truth, the system presents multiple perspectives and acknowledges the subjective nature of certain evaluations. It grounds subjective assessments in objective metrics where possible, such as supporting a claim that code is complex by citing specific complexity metrics rather than relying solely on the model's judgment.
The system also addresses bias in its training and fine-tuning processes.
When fine-tuning local models on organization-specific code and documentation, it ensures that training data represents diverse coding styles, architectural approaches, and problem-solving strategies. It actively seeks to include examples that counter common biases, such as including high-quality functional programming examples to balance object-oriented bias.
Human-in-the-loop workflows provide a critical check on bias. When the system generates analyses, human reviewers from diverse backgrounds and with different areas of expertise review the findings. These reviewers are specifically asked to consider whether bias might be influencing conclusions and to provide alternative perspectives when appropriate.
REASONING STRATEGIES: CHAIN-OF-THOUGHT, TREE-OF-THOUGHT, AND SELF-REFLECTION
The quality of analysis depends not just on what information the system retrieves but on how it reasons about that information. The multi-agent system employs sophisticated reasoning strategies that make the reasoning process explicit, consider alternative interpretations, and critically evaluate conclusions.
Chain-of-thought reasoning requires agents to articulate their reasoning steps explicitly rather than jumping directly to conclusions. When analyzing whether a component adheres to SOLID principles, an agent might reason: "This class has three public methods that serve different purposes. The first method handles user authentication, the second manages session state, and the third generates audit logs. These responsibilities are distinct and could be separated. This suggests a violation of the Single Responsibility Principle. Let me examine whether these responsibilities are cohesive or whether they should be split into separate classes. Authentication and session management are closely related, as sessions are created upon successful authentication. However, audit logging is a cross-cutting concern that applies to many operations beyond authentication. Therefore, I conclude that audit logging should be extracted to a separate component, while authentication and session management can reasonably remain together."
This explicit reasoning allows both automated validation and human review to follow the agent's logic, identify potential flaws, and evaluate whether conclusions are well-supported. It also makes the reasoning process more robust by forcing the agent to consider each step carefully rather than relying on pattern matching that might miss important nuances.
Tree-of-thought reasoning extends chain-of-thought by exploring multiple reasoning paths in parallel. When faced with ambiguous situations, an agent might pursue several different interpretations simultaneously, evaluating each path's plausibility before committing to a conclusion. For example, when analyzing an architectural pattern that could be interpreted as either a layered architecture or a hexagonal architecture, the agent explores both interpretations, examines evidence supporting each, and evaluates which interpretation better explains the observed structure. This multi-path exploration reduces the risk of premature commitment to an incorrect interpretation.
Self-reflection involves agents critically evaluating their own conclusions and actively seeking disconfirming evidence. After reaching a preliminary conclusion, an agent asks itself: "What evidence would contradict this conclusion? Let me specifically search for such evidence." This adversarial approach to self-evaluation helps identify cases where initial conclusions were based on incomplete information or faulty reasoning.
The system implements debate-style reasoning for complex or ambiguous analyses. Multiple agents or multiple instances of the same agent argue for different conclusions, each presenting evidence and reasoning to support their position. A judge agent or human reviewer then evaluates the arguments and determines which is most convincing. This debate approach surfaces multiple perspectives and ensures that alternative interpretations receive serious consideration.
Agents also employ analogical reasoning, comparing the current codebase to known patterns, anti-patterns, and best practices. When an agent recognizes a structure similar to a documented design pattern, it can leverage knowledge about that pattern's typical benefits and drawbacks. However, the system is careful to validate that the analogy is appropriate rather than forcing the code into a pattern that doesn't quite fit.
CONTINUOUS SYNCHRONIZATION: KEEPING ANALYSIS CURRENT AS CODE EVOLVES
Codebases are not static artifacts but living systems that evolve continuously. The multi-agent analysis system must maintain current understanding as developers commit changes, refactor code, add features, and fix bugs. A static analysis that becomes outdated within days or weeks provides limited value.
The system implements a change detection agent that monitors the version control repository for commits. When changes occur, this agent analyzes the modified files to determine what has changed. Rather than triggering a complete re-analysis of the entire codebase for every commit, the system performs incremental updates that focus on affected areas.
When a developer modifies a function, the change detection agent identifies that specific function and triggers re-parsing and re-analysis of it. The system updates the function's summary, recalculates relevant metrics, and updates its embedding in the vector database. It then examines the knowledge graph to identify components that depend on the modified function, triggering re-analysis of those components if the function's interface or behavior changed in ways that might affect them.
For architectural changes that affect multiple components, the system performs broader re-analysis. If a developer refactors a module by splitting it into multiple smaller modules, the change detection agent recognizes this structural change and triggers re-analysis of the affected subsystem. The architecture analysis agent updates its understanding of the system structure, and the knowledge graph is modified to reflect the new component organization.
The system maintains a change log that records what has been modified and when, allowing stakeholders to understand how the system has evolved since the last comprehensive analysis. This change log integrates with the reporting system, so that generated reports can highlight recent changes and their implications.
Incremental analysis is significantly more efficient than full re-analysis, allowing the system to maintain current understanding even in rapidly evolving codebases. However, the system periodically performs full re-analysis to ensure that incremental updates have not introduced inconsistencies or missed subtle changes that accumulate over time.
The synchronization system also monitors changes to non-code artifacts like requirements documents, architectural decision records, and test results. When a requirements document is updated, the requirements analysis agent re-processes it, identifies new or modified requirements, and updates the knowledge graph. The system then checks whether these requirement changes affect the alignment between requirements and implementation, potentially identifying new gaps or validating that recent code changes address previously unmet requirements.
GENERATING COMPREHENSIVE SWOT ANALYSIS REPORTS
The ultimate output of the multi-agent system is a comprehensive analysis report that synthesizes findings from all agents into actionable insights for stakeholders. The centerpiece of this report is a SWOT analysis that evaluates the system's strengths, weaknesses, opportunities, and threats across all dimensions of software engineering.
The report begins with an executive summary that provides senior management with a high-level overview of the system's status, highlighting the most critical findings and their business implications. This summary avoids technical jargon, instead focusing on business impacts like risk to strategic initiatives, cost implications of technical debt, or opportunities to leverage existing capabilities for new features.
The business analysis section evaluates how well the system supports organizational strategy and business goals. Strengths might include robust implementations of core business capabilities that provide competitive advantage, or flexible architectures that allow rapid adaptation to changing business needs. Weaknesses might include missing functionality for planned strategic initiatives, or technical limitations that prevent the system from scaling to support business growth. Opportunities might include underutilized capabilities that could support new business models, or architectural foundations that could be extended to new markets. Threats might include technical debt that risks system reliability, or dependencies on obsolete technologies that could become unsupported.
The requirements analysis section evaluates the completeness, consistency, and implementation status of requirements. Strengths include well-documented requirements with clear traceability to business goals and comprehensive test coverage. Weaknesses include ambiguous or conflicting requirements, requirements that lack implementation, or implemented features that do not trace to documented requirements.
Opportunities include requirements that are partially implemented and could be completed with modest effort, or emerging requirements that align well with existing architectural capabilities. Threats include requirements volatility that suggests business uncertainty, or regulatory requirements that are not adequately addressed.
The architecture and design section evaluates structural quality, pattern adherence, and architectural integrity. Strengths include well-defined architectural layers with appropriate separation of concerns, effective use of design patterns that enhance maintainability, and architectural decisions that successfully address quality attribute requirements. Weaknesses include architectural drift where implementation deviates from intended structure, circular dependencies that create tight coupling, or missing abstraction layers that lead to code duplication. Opportunities include architectural foundations that could be leveraged for new capabilities, or refactoring opportunities that could significantly improve structure with manageable effort. Threats include architectural erosion where incremental changes are degrading structural integrity, or fundamental architectural limitations that will impede future evolution.
The code quality section evaluates implementation quality, maintainability, and adherence to coding standards. Strengths include well-structured code with appropriate abstraction, comprehensive error handling, and clear documentation. Weaknesses include high-complexity functions that are difficult to understand and maintain, code smells that indicate design problems, or inconsistent coding styles that impede readability.
Opportunities include refactoring opportunities that could improve quality, or opportunities to extract reusable components from duplicated code. Threats include increasing complexity trends that suggest maintainability is degrading, or security vulnerabilities in critical code paths.
The testing and quality section evaluates test coverage, test quality, and quality assurance processes. Strengths include comprehensive test suites with high coverage of critical functionality, effective integration tests that validate component interactions, and automated testing that catches regressions quickly. Weaknesses include inadequate test coverage of complex code paths, brittle tests that break frequently due to minor changes, or missing test categories like performance tests or security tests. Opportunities include areas where adding tests would significantly reduce risk, or opportunities to improve test efficiency through better test design. Threats include declining test coverage trends, or increasing defect rates that suggest quality is degrading.
The operations and deployment section evaluates production readiness, operational observability, and deployment reliability. Strengths include robust deployment automation, comprehensive monitoring and alerting, and well-documented operational procedures. Weaknesses include manual deployment steps that risk errors, inadequate logging that impedes troubleshooting, or missing health checks that prevent effective monitoring. Opportunities include opportunities to improve operational efficiency through better automation, or opportunities to enhance reliability through improved monitoring. Threats include operational complexity that risks outages, or dependencies on infrastructure that may become unavailable.
Each finding in the SWOT analysis includes a detailed rationale that explains the basis for the conclusion, cites specific evidence from the codebase or documentation, and provides context about why the finding matters. For weaknesses and threats, the report includes specific recommendations for remediation, including estimated effort, priority based on risk and impact, and dependencies between recommendations.
The report includes visualizations that make complex information accessible. Architectural diagrams show the system's structure, with visual indicators highlighting areas of concern like high-complexity components or components with inadequate test coverage. Dependency graphs show relationships between components, with visual indicators for problematic dependencies like circular references. Traceability matrices show relationships between business goals, requirements, architectural components, code modules, and tests, highlighting gaps where requirements lack implementation or code lacks requirements justification.
For stakeholders who need to explore findings interactively, the system can generate a web-based application that allows navigation through the knowledge graph, drilling down from high-level findings to specific code examples, and exploring relationships between different aspects of the system. This interactive exploration capability is particularly valuable for architects and senior developers who need to understand findings in depth.
IMPLEMENTATION APPROACH: BUILDING THE MULTI-AGENT SYSTEM
Implementing a multi-agent analysis system of this sophistication requires careful attention to architecture, technology selection, and integration. The following sections provide detailed guidance on building such a system, including concrete code examples that illustrate key concepts.
The system is built using Python as the primary implementation language due to its rich ecosystem of libraries for machine learning, natural language processing, graph processing, and data analysis. Python's extensive support for LLM integration through libraries like LangChain, LlamaIndex, and direct API access to OpenAI, Anthropic, and Google makes it an ideal choice. However, the system's architecture allows for polyglot implementation where appropriate, such as using Go for high-performance concurrent processing or Rust for performance-critical parsing operations.
The foundation of the system is an agent framework that provides common infrastructure for all specialized agents. This framework handles model interaction, prompt management, result caching, error handling, and logging. Each specialized agent extends this base framework with domain-specific logic.
The orchestration layer coordinates agent execution using a workflow engine that understands dependencies between analysis tasks. When a comprehensive analysis is requested, the orchestrator creates a directed acyclic graph of tasks, where nodes represent agent analyses and edges represent dependencies. The business strategy analysis must complete before requirements analysis can fully evaluate requirement alignment with business goals. Architecture analysis depends on code parsing completing. Test coverage analysis depends on both code analysis and test parsing. The orchestrator executes independent tasks in parallel to maximize efficiency while respecting dependencies.
The vector database implementation uses a high-performance vector store like Qdrant or Weaviate, which provide efficient similarity search over large embedding collections. These databases support filtering based on metadata, allowing queries like "find code chunks related to authentication that are written in Java and modified within the last month." The system generates embeddings using models specifically trained for code understanding, such as code-specific variants of embedding models or specialized models like CodeBERT.
The knowledge graph is implemented using a graph database like Neo4j, which provides powerful query capabilities through the Cypher query language. The graph schema defines node types for different entity categories like Subsystem, Component, Class, Function, Requirement, BusinessGoal, and Test. Edge types define relationships like DEPENDS_ON, IMPLEMENTS, TESTS, TRACES_TO, and CALLS. This rich schema enables sophisticated queries that traverse multiple relationship types to answer complex questions.
The integration between vector database and knowledge graph is bidirectional. Each node in the knowledge graph includes references to related embeddings in the vector database, allowing graph traversal to retrieve detailed code for analysis. Each embedding in the vector database includes metadata linking it to corresponding knowledge graph nodes, allowing vector search results to be enriched with structural information from the graph.
CODE EXAMPLE: PARSING AND KNOWLEDGE GRAPH CONSTRUCTION
The following example demonstrates how the system parses Python code using Tree-sitter and populates the knowledge graph with extracted entities and relationships. This example focuses on clarity and completeness rather than brevity, showing a realistic implementation that handles real-world code complexity.
import tree_sitter
from tree_sitter import Language, Parser
import tree_sitter_python
from typing import List, Dict, Set, Optional
from dataclasses import dataclass
from neo4j import GraphDatabase
import hashlib
@dataclass
class FunctionInfo:
"""Represents extracted information about a function."""
name: str
parameters: List[str]
return_type: Optional[str]
docstring: Optional[str]
start_line: int
end_line: int
complexity: int
calls: Set[str]
@dataclass
class ClassInfo:
"""Represents extracted information about a class."""
name: str
bases: List[str]
methods: List[FunctionInfo]
attributes: List[str]
docstring: Optional[str]
start_line: int
end_line: int
class PythonCodeParser:
"""Parses Python code using Tree-sitter and extracts semantic information."""
def __init__(self):
"""Initialize the parser with Python language grammar."""
self.language = Language(tree_sitter_python.language())
self.parser = Parser(self.language)
def parse_file(self, file_path: str) -> Dict:
"""Parse a Python file and extract comprehensive information.
Args:
file_path: Path to the Python source file
Returns:
Dictionary containing extracted classes, functions, and imports
"""
with open(file_path, 'r', encoding='utf-8') as f:
source_code = f.read()
tree = self.parser.parse(bytes(source_code, 'utf8'))
root_node = tree.root_node
result = {
'file_path': file_path,
'imports': self._extract_imports(root_node, source_code),
'classes': self._extract_classes(root_node, source_code),
'functions': self._extract_functions(root_node, source_code),
'module_docstring': self._extract_module_docstring(root_node, source_code)
}
return result
def _extract_imports(self, node, source_code: str) -> List[Dict]:
"""Extract import statements from the AST."""
imports = []
query = self.language.query("""
(import_statement) @import
(import_from_statement) @import_from
""")
captures = query.captures(node)
for capture_node, capture_name in captures:
import_text = source_code[capture_node.start_byte:capture_node.end_byte]
imports.append({
'type': capture_name,
'text': import_text,
'line': capture_node.start_point[0] + 1
})
return imports
def _extract_classes(self, node, source_code: str) -> List[ClassInfo]:
"""Extract class definitions with comprehensive information."""
classes = []
query = self.language.query("(class_definition) @class")
captures = query.captures(node)
for class_node, _ in captures:
class_info = self._parse_class(class_node, source_code)
if class_info:
classes.append(class_info)
return classes
def _parse_class(self, class_node, source_code: str) -> Optional[ClassInfo]:
"""Parse detailed information about a class."""
name_node = class_node.child_by_field_name('name')
if not name_node:
return None
class_name = source_code[name_node.start_byte:name_node.end_byte]
bases = []
superclasses_node = class_node.child_by_field_name('superclasses')
if superclasses_node:
for child in superclasses_node.children:
if child.type == 'identifier':
base_name = source_code[child.start_byte:child.end_byte]
bases.append(base_name)
body_node = class_node.child_by_field_name('body')
methods = []
attributes = []
docstring = None
if body_node:
docstring = self._extract_docstring(body_node, source_code)
for child in body_node.children:
if child.type == 'function_definition':
method_info = self._parse_function(child, source_code)
if method_info:
methods.append(method_info)
elif child.type == 'expression_statement':
attr = self._extract_attribute_assignment(child, source_code)
if attr:
attributes.append(attr)
return ClassInfo(
name=class_name,
bases=bases,
methods=methods,
attributes=attributes,
docstring=docstring,
start_line=class_node.start_point[0] + 1,
end_line=class_node.end_point[0] + 1
)
def _extract_functions(self, node, source_code: str) -> List[FunctionInfo]:
"""Extract module-level function definitions."""
functions = []
for child in node.children:
if child.type == 'function_definition':
func_info = self._parse_function(child, source_code)
if func_info:
functions.append(func_info)
return functions
def _parse_function(self, func_node, source_code: str) -> Optional[FunctionInfo]:
"""Parse detailed information about a function."""
name_node = func_node.child_by_field_name('name')
if not name_node:
return None
func_name = source_code[name_node.start_byte:name_node.end_byte]
parameters = []
params_node = func_node.child_by_field_name('parameters')
if params_node:
for child in params_node.children:
if child.type == 'identifier':
param_name = source_code[child.start_byte:child.end_byte]
parameters.append(param_name)
elif child.type == 'typed_parameter':
param_name_node = child.child_by_field_name('name')
if param_name_node:
param_name = source_code[param_name_node.start_byte:param_name_node.end_byte]
parameters.append(param_name)
return_type = None
return_type_node = func_node.child_by_field_name('return_type')
if return_type_node:
return_type = source_code[return_type_node.start_byte:return_type_node.end_byte]
body_node = func_node.child_by_field_name('body')
docstring = None
calls = set()
complexity = 1
if body_node:
docstring = self._extract_docstring(body_node, source_code)
calls = self._extract_function_calls(body_node, source_code)
complexity = self._calculate_complexity(body_node)
return FunctionInfo(
name=func_name,
parameters=parameters,
return_type=return_type,
docstring=docstring,
start_line=func_node.start_point[0] + 1,
end_line=func_node.end_point[0] + 1,
complexity=complexity,
calls=calls
)
def _extract_docstring(self, node, source_code: str) -> Optional[str]:
"""Extract docstring from a function or class body."""
if node.child_count == 0:
return None
first_child = node.children[0]
if first_child.type == 'expression_statement':
string_node = first_child.children[0] if first_child.child_count > 0 else None
if string_node and string_node.type == 'string':
docstring = source_code[string_node.start_byte:string_node.end_byte]
return docstring.strip('"\'')
return None
def _extract_function_calls(self, node, source_code: str) -> Set[str]:
"""Extract all function calls within a node."""
calls = set()
query = self.language.query("(call function: (identifier) @func_name)")
captures = query.captures(node)
for call_node, _ in captures:
func_name = source_code[call_node.start_byte:call_node.end_byte]
calls.add(func_name)
return calls
def _calculate_complexity(self, node) -> int:
"""Calculate cyclomatic complexity of a function."""
complexity = 1
query = self.language.query("""
(if_statement) @decision
(while_statement) @decision
(for_statement) @decision
(except_clause) @decision
(boolean_operator) @decision
""")
captures = query.captures(node)
complexity += len(captures)
return complexity
def _extract_attribute_assignment(self, node, source_code: str) -> Optional[str]:
"""Extract attribute assignments like self.attribute = value."""
if node.child_count == 0:
return None
assignment = node.children[0]
if assignment.type == 'assignment':
left = assignment.child_by_field_name('left')
if left and left.type == 'attribute':
attr_text = source_code[left.start_byte:left.end_byte]
if attr_text.startswith('self.'):
return attr_text[5:]
return None
def _extract_module_docstring(self, node, source_code: str) -> Optional[str]:
"""Extract module-level docstring."""
if node.child_count == 0:
return None
first_child = node.children[0]
if first_child.type == 'expression_statement':
return self._extract_docstring(first_child.parent, source_code)
return None
class KnowledgeGraphBuilder:
"""Builds and maintains the knowledge graph from parsed code information."""
def __init__(self, uri: str, user: str, password: str):
"""Initialize connection to Neo4j graph database.
Args:
uri: Neo4j database URI
user: Database username
password: Database password
"""
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def close(self):
"""Close database connection."""
self.driver.close()
def create_module_node(self, file_path: str, module_info: Dict):
"""Create a module node in the knowledge graph.
Args:
file_path: Path to the source file
module_info: Parsed module information
"""
with self.driver.session() as session:
session.execute_write(self._create_module, file_path, module_info)
@staticmethod
def _create_module(tx, file_path: str, module_info: Dict):
"""Transaction function to create module node."""
module_id = hashlib.sha256(file_path.encode()).hexdigest()[:16]
query = """
MERGE (m:Module {id: $module_id})
SET m.file_path = $file_path,
m.docstring = $docstring,
m.last_updated = datetime()
RETURN m
"""
tx.run(query,
module_id=module_id,
file_path=file_path,
docstring=module_info.get('module_docstring'))
for class_info in module_info.get('classes', []):
KnowledgeGraphBuilder._create_class(tx, module_id, class_info)
for func_info in module_info.get('functions', []):
KnowledgeGraphBuilder._create_function(tx, module_id, func_info, None)
for import_info in module_info.get('imports', []):
KnowledgeGraphBuilder._create_import(tx, module_id, import_info)
@staticmethod
def _create_class(tx, module_id: str, class_info: ClassInfo):
"""Create a class node and its relationships."""
class_id = hashlib.sha256(f"{module_id}:{class_info.name}".encode()).hexdigest()[:16]
query = """
MATCH (m:Module {id: $module_id})
MERGE (c:Class {id: $class_id})
SET c.name = $name,
c.docstring = $docstring,
c.start_line = $start_line,
c.end_line = $end_line,
c.attributes = $attributes
MERGE (m)-[:CONTAINS]->(c)
RETURN c
"""
tx.run(query,
module_id=module_id,
class_id=class_id,
name=class_info.name,
docstring=class_info.docstring,
start_line=class_info.start_line,
end_line=class_info.end_line,
attributes=class_info.attributes)
for base in class_info.bases:
base_query = """
MATCH (c:Class {id: $class_id})
MERGE (b:Class {name: $base_name})
MERGE (c)-[:INHERITS_FROM]->(b)
"""
tx.run(base_query, class_id=class_id, base_name=base)
for method_info in class_info.methods:
KnowledgeGraphBuilder._create_function(tx, module_id, method_info, class_id)
@staticmethod
def _create_function(tx, module_id: str, func_info: FunctionInfo, class_id: Optional[str]):
"""Create a function or method node."""
if class_id:
func_id = hashlib.sha256(f"{class_id}:{func_info.name}".encode()).hexdigest()[:16]
parent_label = "Class"
parent_id = class_id
else:
func_id = hashlib.sha256(f"{module_id}:{func_info.name}".encode()).hexdigest()[:16]
parent_label = "Module"
parent_id = module_id
query = f"""
MATCH (p:{parent_label} {{id: $parent_id}})
MERGE (f:Function {{id: $func_id}})
SET f.name = $name,
f.parameters = $parameters,
f.return_type = $return_type,
f.docstring = $docstring,
f.start_line = $start_line,
f.end_line = $end_line,
f.complexity = $complexity
MERGE (p)-[:CONTAINS]->(f)
RETURN f
"""
tx.run(query,
parent_id=parent_id,
func_id=func_id,
name=func_info.name,
parameters=func_info.parameters,
return_type=func_info.return_type,
docstring=func_info.docstring,
start_line=func_info.start_line,
end_line=func_info.end_line,
complexity=func_info.complexity)
for called_func in func_info.calls:
call_query = """
MATCH (f:Function {id: $func_id})
MERGE (called:Function {name: $called_name})
MERGE (f)-[:CALLS]->(called)
"""
tx.run(call_query, func_id=func_id, called_name=called_func)
@staticmethod
def _create_import(tx, module_id: str, import_info: Dict):
"""Create import relationship."""
query = """
MATCH (m:Module {id: $module_id})
MERGE (imported:Module {name: $import_text})
MERGE (m)-[:IMPORTS]->(imported)
"""
tx.run(query, module_id=module_id, import_text=import_info['text'])
def query_function_complexity(self, threshold: int) -> List[Dict]:
"""Query functions exceeding complexity threshold.
Args:
threshold: Complexity threshold
Returns:
List of functions with high complexity
"""
with self.driver.session() as session:
result = session.run("""
MATCH (f:Function)
WHERE f.complexity > $threshold
RETURN f.name AS name, f.complexity AS complexity, f.start_line AS line
ORDER BY f.complexity DESC
""", threshold=threshold)
return [dict(record) for record in result]
def query_class_dependencies(self, class_name: str) -> List[str]:
"""Query all classes that depend on a given class.
Args:
class_name: Name of the class
Returns:
List of dependent class names
"""
with self.driver.session() as session:
result = session.run("""
MATCH (c:Class {name: $class_name})<-[:INHERITS_FROM|USES]-(dependent:Class)
RETURN DISTINCT dependent.name AS name
""", class_name=class_name)
return [record['name'] for record in result]
This code example demonstrates several key concepts. The PythonCodeParser class uses Tree-sitter to parse Python source files and extract comprehensive information about classes, functions, imports, and their relationships. It goes beyond simple syntactic parsing to extract semantic information like function complexity, docstrings, and call graphs. The parser handles real-world Python code complexity, including type annotations, class inheritance, and various statement types.
The KnowledgeGraphBuilder class takes the parsed information and populates a Neo4j graph database with nodes representing modules, classes, and functions, along with edges representing relationships like containment, inheritance, and function calls. Each entity receives a unique identifier generated by hashing its fully qualified name, ensuring consistent identification across multiple parsing runs. The builder supports incremental updates, merging new information with existing nodes rather than creating duplicates.
The integration between parsing and graph construction creates a foundation for sophisticated analysis. Once the knowledge graph is populated, agents can query it to answer complex questions about the codebase structure, identify problematic patterns, and trace relationships across the system.
PART ONE CONCLUSION
This concludes the first part of the article, which has covered the fundamental architecture, key challenges, reasoning strategies, and core implementation concepts for building LLM-based multi-agent AI systems for code repository analysis. The article has explained why such systems are necessary given the limitations of current AI coding assistants, how they address challenges like context windows and hallucinations, and how they orchestrate specialized agents to provide comprehensive analysis.
Would you like me to continue with the second part of the article, which will cover the complete running example implementation, detailed agent specifications, vector database integration, report generation, and deployment considerations across different GPU architectures?
PART TWO: COMPLETE IMPLEMENTATION, AGENT SPECIFICATIONS, AND DEPLOYMENT
VECTOR DATABASE INTEGRATION AND EMBEDDING GENERATION
The vector database serves as the semantic memory of the multi-agent system, enabling agents to retrieve relevant code and documentation based on meaning rather than exact keyword matches. This section details the implementation of vector database integration, embedding generation strategies, and retrieval mechanisms.
Embedding Strategy for Code and Documentation
Code embeddings differ fundamentally from natural language embeddings because code has both semantic meaning (what the code does) and structural properties (how it's organized). The system employs specialized embedding models that understand code semantics across multiple programming languages.
In 2026, several embedding models excel at code understanding. OpenAI's text-embedding-3-large and text-embedding-3-small models, while designed for general text, perform well on code when the code is properly formatted. However, specialized models like CodeBERT, GraphCodeBERT, and UniXcoder provide superior performance by pre-training on large code corpora and understanding code structure through abstract syntax trees.
The system generates embeddings at multiple granularities to support different types of queries. Function-level embeddings capture the semantics of individual functions or methods, including their purpose, algorithmic approach, and relationship to the broader system. Class-level embeddings represent the overall responsibility and interface of classes. Module-level embeddings capture the high-level purpose and exports of entire modules or files. Documentation embeddings represent requirements documents, architectural decision records, API documentation, and other textual artifacts.
Each embedding is stored with rich metadata that enables filtered retrieval. Metadata includes the programming language, file path, component or subsystem identifier, last modification date, author information, and links to corresponding knowledge graph nodes. This metadata allows queries like "find authentication-related code in Java services modified in the last quarter" or "find Python data processing functions in the analytics subsystem."
CODE EXAMPLE: Vector Database Integration with Qdrant
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue
from typing import List, Dict, Optional, Tuple
import openai
from dataclasses import dataclass
import hashlib
from datetime import datetime
import numpy as np
@dataclass
class CodeChunk:
"""Represents a chunk of code with metadata."""
content: str
file_path: str
language: str
chunk_type: str # 'function', 'class', 'module'
start_line: int
end_line: int
entity_name: str
component: str
subsystem: str
complexity: Optional[int] = None
last_modified: Optional[datetime] = None
author: Optional[str] = None
class CodeEmbeddingGenerator:
"""Generates embeddings for code chunks using specialized models."""
def __init__(self, api_key: str, model: str = "text-embedding-3-large"):
"""Initialize the embedding generator.
Args:
api_key: OpenAI API key
model: Embedding model to use
"""
self.client = openai.OpenAI(api_key=api_key)
self.model = model
self.embedding_dimension = 3072 if "large" in model else 1536
def generate_embedding(self, code_chunk: CodeChunk) -> np.ndarray:
"""Generate embedding for a code chunk.
Args:
code_chunk: Code chunk to embed
Returns:
Embedding vector as numpy array
"""
context = self._prepare_code_context(code_chunk)
response = self.client.embeddings.create(
input=context,
model=self.model
)
embedding = np.array(response.data[0].embedding)
return embedding
def _prepare_code_context(self, code_chunk: CodeChunk) -> str:
"""Prepare code with context for better embeddings."""
context_parts = [
f"Language: {code_chunk.language}",
f"Type: {code_chunk.chunk_type}",
f"Component: {code_chunk.component}",
f"Entity: {code_chunk.entity_name}",
"",
"Code:",
code_chunk.content
]
return "\n".join(context_parts)
def generate_batch_embeddings(self, code_chunks: List[CodeChunk],
batch_size: int = 100) -> List[np.ndarray]:
"""Generate embeddings for multiple code chunks efficiently."""
embeddings = []
for i in range(0, len(code_chunks), batch_size):
batch = code_chunks[i:i + batch_size]
contexts = [self._prepare_code_context(chunk) for chunk in batch]
response = self.client.embeddings.create(
input=contexts,
model=self.model
)
batch_embeddings = [np.array(item.embedding) for item in response.data]
embeddings.extend(batch_embeddings)
return embeddings
class VectorDatabaseManager:
"""Manages code embeddings in Qdrant vector database."""
def __init__(self, host: str = "localhost", port: int = 6333,
collection_name: str = "code_embeddings"):
"""Initialize connection to Qdrant."""
self.client = QdrantClient(host=host, port=port)
self.collection_name = collection_name
self.embedding_dimension = 3072
def initialize_collection(self):
"""Create the collection if it doesn't exist."""
collections = self.client.get_collections().collections
collection_names = [col.name for col in collections]
if self.collection_name not in collection_names:
self.client.create_collection(
collection_name=self.collection_name,
vectors_config=VectorParams(
size=self.embedding_dimension,
distance=Distance.COSINE
)
)
def store_code_chunk(self, code_chunk: CodeChunk, embedding: np.ndarray):
"""Store a code chunk with its embedding."""
point_id = self._generate_point_id(code_chunk)
payload = {
"content": code_chunk.content,
"file_path": code_chunk.file_path,
"language": code_chunk.language,
"chunk_type": code_chunk.chunk_type,
"start_line": code_chunk.start_line,
"end_line": code_chunk.end_line,
"entity_name": code_chunk.entity_name,
"component": code_chunk.component,
"subsystem": code_chunk.subsystem,
"complexity": code_chunk.complexity,
"last_modified": code_chunk.last_modified.isoformat() if code_chunk.last_modified else None,
"author": code_chunk.author
}
point = PointStruct(
id=point_id,
vector=embedding.tolist(),
payload=payload
)
self.client.upsert(
collection_name=self.collection_name,
points=[point]
)
def store_batch(self, code_chunks: List[CodeChunk], embeddings: List[np.ndarray]):
"""Store multiple code chunks efficiently."""
points = []
for code_chunk, embedding in zip(code_chunks, embeddings):
point_id = self._generate_point_id(code_chunk)
payload = {
"content": code_chunk.content,
"file_path": code_chunk.file_path,
"language": code_chunk.language,
"chunk_type": code_chunk.chunk_type,
"start_line": code_chunk.start_line,
"end_line": code_chunk.end_line,
"entity_name": code_chunk.entity_name,
"component": code_chunk.component,
"subsystem": code_chunk.subsystem,
"complexity": code_chunk.complexity,
"last_modified": code_chunk.last_modified.isoformat() if code_chunk.last_modified else None,
"author": code_chunk.author
}
point = PointStruct(
id=point_id,
vector=embedding.tolist(),
payload=payload
)
points.append(point)
self.client.upsert(
collection_name=self.collection_name,
points=points
)
def semantic_search(self, query_text: str, embedding_generator: CodeEmbeddingGenerator,
top_k: int = 10, filters: Optional[Dict] = None) -> List[Dict]:
"""Perform semantic search for code chunks."""
query_chunk = CodeChunk(
content=query_text,
file_path="",
language=filters.get("language", "python") if filters else "python",
chunk_type="query",
start_line=0,
end_line=0,
entity_name="query",
component=filters.get("component", "") if filters else "",
subsystem=filters.get("subsystem", "") if filters else ""
)
query_embedding = embedding_generator.generate_embedding(query_chunk)
filter_conditions = None
if filters:
conditions = []
if "language" in filters:
conditions.append(
FieldCondition(
key="language",
match=MatchValue(value=filters["language"])
)
)
if "component" in filters:
conditions.append(
FieldCondition(
key="component",
match=MatchValue(value=filters["component"])
)
)
if conditions:
filter_conditions = Filter(must=conditions)
results = self.client.search(
collection_name=self.collection_name,
query_vector=query_embedding.tolist(),
limit=top_k,
query_filter=filter_conditions
)
return [
{
"score": result.score,
"content": result.payload["content"],
"file_path": result.payload["file_path"],
"language": result.payload["language"],
"chunk_type": result.payload["chunk_type"],
"entity_name": result.payload["entity_name"],
"component": result.payload["component"],
"subsystem": result.payload["subsystem"],
"start_line": result.payload["start_line"],
"end_line": result.payload["end_line"],
"complexity": result.payload.get("complexity")
}
for result in results
]
def _generate_point_id(self, code_chunk: CodeChunk) -> str:
"""Generate a unique ID for a code chunk."""
unique_string = f"{code_chunk.file_path}:{code_chunk.start_line}:{code_chunk.end_line}:{code_chunk.entity_name}"
return hashlib.sha256(unique_string.encode()).hexdigest()[:16]
class HybridRetriever:
"""Combines vector search with knowledge graph traversal."""
def __init__(self, vector_db: VectorDatabaseManager,
knowledge_graph: 'KnowledgeGraphBuilder',
embedding_generator: CodeEmbeddingGenerator):
"""Initialize the hybrid retriever."""
self.vector_db = vector_db
self.knowledge_graph = knowledge_graph
self.embedding_generator = embedding_generator
def retrieve_with_context(self, query: str, top_k: int = 10) -> List[Dict]:
"""Retrieve code chunks with expanded context from knowledge graph."""
initial_results = self.vector_db.semantic_search(
query_text=query,
embedding_generator=self.embedding_generator,
top_k=top_k
)
enriched_results = []
for result in initial_results:
entity_name = result["entity_name"]
if result["chunk_type"] == "function":
callers = self._get_function_callers(entity_name)
callees = self._get_function_callees(entity_name)
result["callers"] = callers
result["callees"] = callees
elif result["chunk_type"] == "class":
subclasses = self._get_subclasses(entity_name)
superclasses = self._get_superclasses(entity_name)
result["subclasses"] = subclasses
result["superclasses"] = superclasses
enriched_results.append(result)
return enriched_results
def _get_function_callers(self, function_name: str) -> List[str]:
"""Get functions that call the specified function."""
with self.knowledge_graph.driver.session() as session:
result = session.run("""
MATCH (caller:Function)-[:CALLS]->(f:Function {name: $name})
RETURN caller.name AS caller_name
LIMIT 10
""", name=function_name)
return [record["caller_name"] for record in result]
def _get_function_callees(self, function_name: str) -> List[str]:
"""Get functions called by the specified function."""
with self.knowledge_graph.driver.session() as session:
result = session.run("""
MATCH (f:Function {name: $name})-[:CALLS]->(callee:Function)
RETURN callee.name AS callee_name
LIMIT 10
""", name=function_name)
return [record["callee_name"] for record in result]
def _get_subclasses(self, class_name: str) -> List[str]:
"""Get subclasses of the specified class."""
with self.knowledge_graph.driver.session() as session:
result = session.run("""
MATCH (subclass:Class)-[:INHERITS_FROM]->(c:Class {name: $name})
RETURN subclass.name AS subclass_name
""", name=class_name)
return [record["subclass_name"] for record in result]
def _get_superclasses(self, class_name: str) -> List[str]:
"""Get superclasses of the specified class."""
with self.knowledge_graph.driver.session() as session:
result = session.run("""
MATCH (c:Class {name: $name})-[:INHERITS_FROM]->(superclass:Class)
RETURN superclass.name AS superclass_name
""", name=class_name)
return [record["superclass_name"] for record in result]
This implementation demonstrates sophisticated capabilities for managing code embeddings and performing hybrid retrieval that combines semantic search with structural graph traversal.
SPECIALIZED AGENT IMPLEMENTATIONS
Architecture Analysis Agent
The architecture analysis agent examines system structure, identifies architectural patterns and anti-patterns, evaluates quality attributes, and detects architectural drift.
from typing import List, Dict, Optional, Set, Tuple
from dataclasses import dataclass
from enum import Enum
import anthropic
from collections import defaultdict
class ArchitecturalPattern(Enum):
"""Common architectural patterns."""
LAYERED = "Layered Architecture"
MICROSERVICES = "Microservices"
HEXAGONAL = "Hexagonal/Ports and Adapters"
EVENT_DRIVEN = "Event-Driven"
MVC = "Model-View-Controller"
CLEAN = "Clean Architecture"
UNKNOWN = "Unknown Pattern"
class ArchitecturalSmell(Enum):
"""Architectural anti-patterns and smells."""
CIRCULAR_DEPENDENCY = "Circular Dependency"
GOD_COMPONENT = "God Component"
DENSE_STRUCTURE = "Dense Structure"
UNSTABLE_DEPENDENCY = "Unstable Dependency"
AMBIGUOUS_INTERFACE = "Ambiguous Interface"
@dataclass
class ComponentMetrics:
"""Metrics for a component."""
name: str
lines_of_code: int
number_of_classes: int
afferent_coupling: int
efferent_coupling: int
instability: float
abstractness: float
distance_from_main: float
cyclomatic_complexity: int
dependencies: Set[str]
dependents: Set[str]
@dataclass
class ArchitecturalAnalysis:
"""Results of architectural analysis."""
identified_patterns: List[ArchitecturalPattern]
architectural_smells: List[Tuple[ArchitecturalSmell, str, str]]
component_metrics: Dict[str, ComponentMetrics]
layer_violations: List[Tuple[str, str, str]]
circular_dependencies: List[List[str]]
hotspots: List[Tuple[str, float, str]]
recommendations: List[str]
class ArchitectureAnalysisAgent:
"""Agent specialized in analyzing software architecture."""
def __init__(self, knowledge_graph: 'KnowledgeGraphBuilder',
vector_db: VectorDatabaseManager,
embedding_generator: CodeEmbeddingGenerator,
llm_api_key: str):
"""Initialize the architecture analysis agent."""
self.knowledge_graph = knowledge_graph
self.vector_db = vector_db
self.embedding_generator = embedding_generator
self.client = anthropic.Anthropic(api_key=llm_api_key)
def analyze_architecture(self) -> ArchitecturalAnalysis:
"""Perform comprehensive architectural analysis."""
print("Starting architectural analysis...")
component_metrics = self._extract_component_metrics()
patterns = self._identify_patterns(component_metrics)
smells = self._detect_architectural_smells(component_metrics)
layer_violations = self._detect_layer_violations(component_metrics)
circular_deps = self._detect_circular_dependencies(component_metrics)
hotspots = self._identify_hotspots(component_metrics)
recommendations = self._generate_recommendations(
patterns, smells, layer_violations, circular_deps, hotspots
)
return ArchitecturalAnalysis(
identified_patterns=patterns,
architectural_smells=smells,
component_metrics=component_metrics,
layer_violations=layer_violations,
circular_dependencies=circular_deps,
hotspots=hotspots,
recommendations=recommendations
)
def _extract_component_metrics(self) -> Dict[str, ComponentMetrics]:
"""Extract metrics for all components from knowledge graph."""
with self.knowledge_graph.driver.session() as session:
result = session.run("""
MATCH (m:Module)
OPTIONAL MATCH (m)-[:CONTAINS]->(c:Class)
OPTIONAL MATCH (m)-[:IMPORTS]->(dep:Module)
OPTIONAL MATCH (dependent:Module)-[:IMPORTS]->(m)
RETURN m.file_path AS component,
count(DISTINCT c) AS num_classes,
collect(DISTINCT dep.file_path) AS dependencies,
collect(DISTINCT dependent.file_path) AS dependents
""")
metrics = {}
for record in result:
component = record["component"]
dependencies = set(d for d in record["dependencies"] if d)
dependents = set(d for d in record["dependents"] if d)
efferent = len(dependencies)
afferent = len(dependents)
instability = efferent / (afferent + efferent) if (afferent + efferent) > 0 else 0
complexity = self._get_component_complexity(session, component)
abstractness = self._get_component_abstractness(session, component)
distance = abs(abstractness + instability - 1)
metrics[component] = ComponentMetrics(
name=component,
lines_of_code=0,
number_of_classes=record["num_classes"],
afferent_coupling=afferent,
efferent_coupling=efferent,
instability=instability,
abstractness=abstractness,
distance_from_main=distance,
cyclomatic_complexity=complexity,
dependencies=dependencies,
dependents=dependents
)
return metrics
def _get_component_complexity(self, session, component: str) -> int:
"""Calculate total cyclomatic complexity for a component."""
result = session.run("""
MATCH (m:Module {file_path: $component})-[:CONTAINS*]->(f:Function)
RETURN sum(f.complexity) AS total_complexity
""", component=component)
record = result.single()
return record["total_complexity"] if record and record["total_complexity"] else 0
def _get_component_abstractness(self, session, component: str) -> float:
"""Calculate abstractness metric for a component."""
result = session.run("""
MATCH (m:Module {file_path: $component})-[:CONTAINS]->(c:Class)
WITH count(c) AS total_classes
MATCH (m:Module {file_path: $component})-[:CONTAINS]->(c:Class)
WHERE c.is_abstract = true
RETURN total_classes, count(c) AS abstract_classes
""", component=component)
record = result.single()
if record and record["total_classes"] and record["total_classes"] > 0:
return record["abstract_classes"] / record["total_classes"]
return 0.0
def _identify_patterns(self, metrics: Dict[str, ComponentMetrics]) -> List[ArchitecturalPattern]:
"""Identify architectural patterns using LLM analysis."""
structure_summary = self._create_structure_summary(metrics)
prompt = f"""Analyze the following software architecture and identify architectural patterns.
Component Structure:
{structure_summary}
Identify which architectural patterns are present. Consider:
- Layered Architecture: Clear separation into layers
- Microservices: Independent, loosely coupled services
- Hexagonal: Core domain isolated from external concerns
- Event-Driven: Components communicate through events
- MVC: Separation of model, view, and controller
- Clean Architecture: Dependency rule with concentric layers
Format:
PATTERN: [Pattern Name]
CONFIDENCE: [High/Medium/Low]
EVIDENCE: [Specific evidence]"""
response = self.client.messages.create(
model="claude-opus-4-20250514",
max_tokens=2000,
messages=[{"role": "user", "content": prompt}]
)
patterns = self._parse_pattern_response(response.content[0].text)
return patterns
def _create_structure_summary(self, metrics: Dict[str, ComponentMetrics]) -> str:
"""Create a textual summary of the architectural structure."""
lines = []
for component, metric in list(metrics.items())[:20]: # Limit for context
lines.append(f"\nComponent: {component}")
lines.append(f" Classes: {metric.number_of_classes}")
lines.append(f" Complexity: {metric.cyclomatic_complexity}")
lines.append(f" Instability: {metric.instability:.2f}")
lines.append(f" Dependencies: {len(metric.dependencies)}")
return "\n".join(lines)
def _parse_pattern_response(self, response_text: str) -> List[ArchitecturalPattern]:
"""Parse LLM response to extract identified patterns."""
patterns = []
lines = response_text.split('\n')
for line in lines:
if line.startswith('PATTERN:'):
pattern_name = line.replace('PATTERN:', '').strip()
for pattern in ArchitecturalPattern:
if pattern_name.lower() in pattern.value.lower():
patterns.append(pattern)
break
return patterns if patterns else [ArchitecturalPattern.UNKNOWN]
def _detect_architectural_smells(self, metrics: Dict[str, ComponentMetrics]) -> List[Tuple[ArchitecturalSmell, str, str]]:
"""Detect architectural anti-patterns and smells."""
smells = []
for component, metric in metrics.items():
if metric.cyclomatic_complexity > 500 and metric.number_of_classes > 20:
smells.append((
ArchitecturalSmell.GOD_COMPONENT,
component,
f"Component has {metric.number_of_classes} classes and complexity of {metric.cyclomatic_complexity}"
))
if metric.instability < 0.3:
for dep in metric.dependencies:
if dep in metrics and metrics[dep].instability > 0.7:
smells.append((
ArchitecturalSmell.UNSTABLE_DEPENDENCY,
component,
f"Stable component depends on unstable component {dep}"
))
if metric.distance_from_main > 0.7:
smells.append((
ArchitecturalSmell.AMBIGUOUS_INTERFACE,
component,
f"Component in zone of pain (distance: {metric.distance_from_main:.2f})"
))
return smells
def _detect_layer_violations(self, metrics: Dict[str, ComponentMetrics]) -> List[Tuple[str, str, str]]:
"""Detect violations of layered architecture principles."""
violations = []
layers = self._infer_layers(metrics)
layer_order = ["presentation", "application", "domain", "infrastructure", "data"]
for component, metric in metrics.items():
component_layer = self._get_component_layer(component, layers)
if component_layer and component_layer in layer_order:
component_level = layer_order.index(component_layer)
for dep in metric.dependencies:
dep_layer = self._get_component_layer(dep, layers)
if dep_layer and dep_layer in layer_order:
dep_level = layer_order.index(dep_layer)
if component_level > dep_level:
violations.append((
component,
dep,
f"{component_layer} layer depends on {dep_layer} layer"
))
return violations
def _infer_layers(self, metrics: Dict[str, ComponentMetrics]) -> Dict[str, str]:
"""Infer architectural layers from component names."""
layers = {}
for component in metrics.keys():
component_lower = component.lower()
if any(term in component_lower for term in ['ui', 'view', 'controller', 'presentation']):
layers[component] = "presentation"
elif any(term in component_lower for term in ['service', 'application']):
layers[component] = "application"
elif any(term in component_lower for term in ['domain', 'model', 'entity']):
layers[component] = "domain"
elif any(term in component_lower for term in ['repository', 'dao', 'data']):
layers[component] = "data"
elif any(term in component_lower for term in ['infrastructure', 'config']):
layers[component] = "infrastructure"
return layers
def _get_component_layer(self, component: str, layers: Dict[str, str]) -> Optional[str]:
"""Get the layer a component belongs to."""
return layers.get(component)
def _detect_circular_dependencies(self, metrics: Dict[str, ComponentMetrics]) -> List[List[str]]:
"""Detect circular dependencies using cycle detection."""
cycles = []
graph = {component: list(metric.dependencies) for component, metric in metrics.items()}
visited = set()
rec_stack = set()
path = []
def dfs(node):
visited.add(node)
rec_stack.add(node)
path.append(node)
if node in graph:
for neighbor in graph[node]:
if neighbor not in visited:
if dfs(neighbor):
return True
elif neighbor in rec_stack:
cycle_start = path.index(neighbor)
cycle = path[cycle_start:] + [neighbor]
cycles.append(cycle)
return True
path.pop()
rec_stack.remove(node)
return False
for component in graph:
if component not in visited:
dfs(component)
return cycles
def _identify_hotspots(self, metrics: Dict[str, ComponentMetrics]) -> List[Tuple[str, float, str]]:
"""Identify architectural hotspots that need attention."""
hotspots = []
for component, metric in metrics.items():
score = 0.0
reasons = []
if metric.cyclomatic_complexity > 300:
score += 0.3
reasons.append(f"High complexity ({metric.cyclomatic_complexity})")
total_coupling = metric.afferent_coupling + metric.efferent_coupling
if total_coupling > 10:
score += 0.3
reasons.append(f"High coupling ({total_coupling})")
if metric.distance_from_main > 0.5:
score += 0.2
reasons.append(f"Far from main sequence ({metric.distance_from_main:.2f})")
if metric.number_of_classes > 15:
score += 0.2
reasons.append(f"Many classes ({metric.number_of_classes})")
if score > 0.5:
hotspots.append((component, score, "; ".join(reasons)))
hotspots.sort(key=lambda x: x[1], reverse=True)
return hotspots
def _generate_recommendations(self, patterns, smells, violations, circular_deps, hotspots) -> List[str]:
"""Generate architectural recommendations using LLM."""
summary = f"""Architectural Analysis Summary:
Patterns: {', '.join(p.value for p in patterns)}
Smells: {len(smells)} found
Layer Violations: {len(violations)} found
Circular Dependencies: {len(circular_deps)} found
Hotspots: {len(hotspots)} identified
Provide specific, actionable recommendations to improve the architecture."""
response = self.client.messages.create(
model="claude-opus-4-20250514",
max_tokens=3000,
messages=[{"role": "user", "content": summary}]
)
recommendations_text = response.content[0].text
recommendations = [line.strip('- ').strip() for line in recommendations_text.split('\n')
if line.strip().startswith('-')]
return recommendations
Requirements Traceability Agent
from typing import List, Dict, Set, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
import openai
class RequirementStatus(Enum):
"""Status of requirement implementation."""
FULLY_IMPLEMENTED = "Fully Implemented"
PARTIALLY_IMPLEMENTED = "Partially Implemented"
NOT_IMPLEMENTED = "Not Implemented"
UNKNOWN = "Unknown"
class RequirementType(Enum):
"""Types of requirements."""
FUNCTIONAL = "Functional"
NON_FUNCTIONAL = "Non-Functional"
QUALITY_ATTRIBUTE = "Quality Attribute"
@dataclass
class Requirement:
"""Represents a software requirement."""
id: str
title: str
description: str
type: RequirementType
priority: str
source: str
acceptance_criteria: List[str]
@dataclass
class TraceabilityLink:
"""Represents a traceability link between artifacts."""
from_artifact: str
from_type: str
to_artifact: str
to_type: str
link_type: str
confidence: float
@dataclass
class RequirementAnalysis:
"""Analysis results for a requirement."""
requirement: Requirement
status: RequirementStatus
implementing_components: List[str]
test_coverage: List[str]
gaps: List[str]
traceability_links: List[TraceabilityLink]
class RequirementsTraceabilityAgent:
"""Agent specialized in requirements analysis and traceability."""
def __init__(self, knowledge_graph, vector_db, embedding_generator, llm_api_key: str):
"""Initialize the requirements traceability agent."""
self.knowledge_graph = knowledge_graph
self.vector_db = vector_db
self.embedding_generator = embedding_generator
self.client = openai.OpenAI(api_key=llm_api_key)
def analyze_requirements(self, requirements: List[Requirement]) -> List[RequirementAnalysis]:
"""Analyze implementation status and traceability for all requirements."""
analyses = []
for req in requirements:
print(f"Analyzing requirement: {req.id} - {req.title}")
analysis = self._analyze_single_requirement(req)
analyses.append(analysis)
return analyses
def _analyze_single_requirement(self, requirement: Requirement) -> RequirementAnalysis:
"""Analyze a single requirement's implementation and traceability."""
implementing_components = self._find_implementing_components(requirement)
test_coverage = self._find_test_coverage(requirement, implementing_components)
status = self._determine_status(requirement, implementing_components, test_coverage)
gaps = self._identify_gaps(requirement, implementing_components, test_coverage, status)
traceability_links = self._build_traceability_links(
requirement, implementing_components, test_coverage
)
return RequirementAnalysis(
requirement=requirement,
status=status,
implementing_components=implementing_components,
test_coverage=test_coverage,
gaps=gaps,
traceability_links=traceability_links
)
def _find_implementing_components(self, requirement: Requirement) -> List[str]:
"""Find code components that implement a requirement."""
search_query = f"{requirement.title}. {requirement.description}"
results = self.vector_db.semantic_search(
query_text=search_query,
embedding_generator=self.embedding_generator,
top_k=20
)
components = self._llm_filter_implementations(requirement, results)
return components
def _llm_filter_implementations(self, requirement: Requirement,
search_results: List[Dict]) -> List[str]:
"""Use LLM to filter search results and identify actual implementations."""
code_snippets = []
for i, result in enumerate(search_results[:10]):
code_snippets.append(f"""
Component {i+1}: {result['entity_name']}
Code:
{result['content'][:500]}
""")
prompt = f"""Analyze whether these code components implement this requirement:
Requirement: {requirement.title}
Description: {requirement.description}
Code Components:
{chr(10).join(code_snippets)}
For each component, determine if it implements the requirement.
Format: Component X: YES/NO - [explanation]"""
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
implementing_components = []
response_text = response.choices[0].message.content
for i, result in enumerate(search_results[:10]):
if f"Component {i+1}: YES" in response_text.upper():
implementing_components.append(result['entity_name'])
return implementing_components
def _find_test_coverage(self, requirement: Requirement,
implementing_components: List[str]) -> List[str]:
"""Find tests that cover a requirement."""
tests = []
test_query = f"test {requirement.title}"
test_results = self.vector_db.semantic_search(
query_text=test_query,
embedding_generator=self.embedding_generator,
top_k=15
)
for result in test_results:
entity_name = result['entity_name']
if 'test' in entity_name.lower():
tests.append(entity_name)
return list(set(tests))
def _determine_status(self, requirement, implementing_components, test_coverage) -> RequirementStatus:
"""Determine the implementation status of a requirement."""
if not implementing_components:
return RequirementStatus.NOT_IMPLEMENTED
if len(implementing_components) >= 2 and len(test_coverage) >= 2:
return RequirementStatus.FULLY_IMPLEMENTED
if implementing_components:
return RequirementStatus.PARTIALLY_IMPLEMENTED
return RequirementStatus.UNKNOWN
def _identify_gaps(self, requirement, implementing_components, test_coverage, status) -> List[str]:
"""Identify gaps in requirement implementation."""
gaps = []
if status == RequirementStatus.NOT_IMPLEMENTED:
gaps.append("Requirement has no implementation")
if len(test_coverage) == 0:
gaps.append("No tests found for this requirement")
if len(implementing_components) == 0:
gaps.append("No implementing components identified")
return gaps
def _build_traceability_links(self, requirement, implementing_components, test_coverage) -> List[TraceabilityLink]:
"""Build traceability links between requirement and artifacts."""
links = []
for component in implementing_components:
links.append(TraceabilityLink(
from_artifact=requirement.id,
from_type="Requirement",
to_artifact=component,
to_type="Component",
link_type="implements",
confidence=0.8
))
for test in test_coverage:
links.append(TraceabilityLink(
from_artifact=requirement.id,
from_type="Requirement",
to_artifact=test,
to_type="Test",
link_type="tests",
confidence=0.7
))
return links
SWOT ANALYSIS REPORT GENERATION
from typing import List, Dict
from dataclasses import dataclass
from datetime import datetime
import json
@dataclass
class SWOTCategory:
"""Represents one category of SWOT analysis."""
category: str
domain: str
findings: List[Dict]
class SWOTReportGenerator:
"""Generates comprehensive SWOT analysis reports."""
def __init__(self, architecture_agent, requirements_agent, llm_api_key: str):
"""Initialize the report generator."""
self.architecture_agent = architecture_agent
self.requirements_agent = requirements_agent
self.client = openai.OpenAI(api_key=llm_api_key)
def generate_comprehensive_report(self, requirements: List[Requirement],
business_context: str) -> Dict:
"""Generate a comprehensive SWOT analysis report."""
print("Generating comprehensive SWOT analysis...")
arch_analysis = self.architecture_agent.analyze_architecture()
req_analyses = self.requirements_agent.analyze_requirements(requirements)
swot_categories = []
swot_categories.extend(self._generate_architecture_swot(arch_analysis))
swot_categories.extend(self._generate_requirements_swot(req_analyses))
swot_categories.extend(self._generate_business_swot(business_context, arch_analysis, req_analyses))
executive_summary = self._generate_executive_summary(swot_categories, arch_analysis, req_analyses)
recommendations = self._generate_prioritized_recommendations(swot_categories)
report = {
"metadata": {
"generated_at": datetime.now().isoformat(),
"analysis_scope": "Complete Repository Analysis",
"version": "1.0"
},
"executive_summary": executive_summary,
"swot_analysis": self._organize_swot(swot_categories),
"detailed_findings": {
"architecture": self._format_architecture_findings(arch_analysis),
"requirements": self._format_requirements_findings(req_analyses)
},
"recommendations": recommendations,
"metrics": self._generate_metrics_summary(arch_analysis, req_analyses)
}
return report
def _generate_architecture_swot(self, analysis) -> List[SWOTCategory]:
"""Generate SWOT categories for architecture analysis."""
categories = []
# Strengths
strengths = []
if analysis.identified_patterns:
strengths.append({
"title": "Well-Defined Architectural Patterns",
"description": f"System implements: {', '.join(p.value for p in analysis.identified_patterns)}",
"evidence": "Pattern analysis",
"priority": "Medium",
"impact": "Positive"
})
if strengths:
categories.append(SWOTCategory(
category="Strengths",
domain="Architecture",
findings=strengths
))
# Weaknesses
weaknesses = []
if analysis.architectural_smells:
for smell, component, description in analysis.architectural_smells[:5]:
weaknesses.append({
"title": f"Architectural Smell: {smell.value}",
"description": description,
"evidence": f"Component: {component}",
"priority": "High",
"impact": "Negative"
})
if weaknesses:
categories.append(SWOTCategory(
category="Weaknesses",
domain="Architecture",
findings=weaknesses
))
# Opportunities
opportunities = []
if analysis.hotspots:
refactorable = [h for h in analysis.hotspots if h[1] > 0.7]
if refactorable:
opportunities.append({
"title": "High-Impact Refactoring Opportunities",
"description": f"{len(refactorable)} components identified for refactoring",
"evidence": f"Top: {refactorable[0][0]}",
"priority": "High",
"impact": "Positive"
})
if opportunities:
categories.append(SWOTCategory(
category="Opportunities",
domain="Architecture",
findings=opportunities
))
# Threats
threats = []
if analysis.circular_dependencies:
threats.append({
"title": "Circular Dependencies Detected",
"description": f"{len(analysis.circular_dependencies)} cycles found",
"evidence": f"Example: {' -> '.join(analysis.circular_dependencies[0])}",
"priority": "High",
"impact": "Negative"
})
if threats:
categories.append(SWOTCategory(
category="Threats",
domain="Architecture",
findings=threats
))
return categories
def _generate_requirements_swot(self, analyses: List[RequirementAnalysis]) -> List[SWOTCategory]:
"""Generate SWOT categories for requirements analysis."""
categories = []
total_reqs = len(analyses)
fully_implemented = sum(1 for a in analyses if a.status == RequirementStatus.FULLY_IMPLEMENTED)
not_implemented = sum(1 for a in analyses if a.status == RequirementStatus.NOT_IMPLEMENTED)
# Strengths
strengths = []
if fully_implemented / total_reqs > 0.7:
strengths.append({
"title": "High Requirements Implementation Rate",
"description": f"{fully_implemented}/{total_reqs} requirements fully implemented",
"evidence": "Requirements traceability analysis",
"priority": "High",
"impact": "Positive"
})
if strengths:
categories.append(SWOTCategory(
category="Strengths",
domain="Requirements",
findings=strengths
))
# Weaknesses
weaknesses = []
if not_implemented > 0:
weaknesses.append({
"title": "Unimplemented Requirements",
"description": f"{not_implemented} requirements have no implementation",
"evidence": "Requirements analysis",
"priority": "High",
"impact": "Negative"
})
if weaknesses:
categories.append(SWOTCategory(
category="Weaknesses",
domain="Requirements",
findings=weaknesses
))
return categories
def _generate_business_swot(self, business_context, arch_analysis, req_analyses) -> List[SWOTCategory]:
"""Generate SWOT categories for business alignment."""
prompt = f"""Analyze business alignment:
Business Context: {business_context}
Technical Summary:
- Patterns: {', '.join(p.value for p in arch_analysis.identified_patterns)}
- Requirements: {len(req_analyses)} total
- Issues: {len(arch_analysis.architectural_smells)} smells
Generate SWOT findings for business alignment.
Format each as:
CATEGORY: [Strengths/Weaknesses/Opportunities/Threats]
TITLE: [Title]
DESCRIPTION: [Description]
PRIORITY: [High/Medium/Low]
---"""
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.4
)
categories = self._parse_business_swot_response(response.choices[0].message.content)
return categories
def _parse_business_swot_response(self, response_text: str) -> List[SWOTCategory]:
"""Parse LLM response into SWOT categories."""
findings_by_category = {
"Strengths": [],
"Weaknesses": [],
"Opportunities": [],
"Threats": []
}
current_category = None
current_finding = {}
for line in response_text.split('\n'):
line = line.strip()
if line.startswith('CATEGORY:'):
if current_finding and 'title' in current_finding:
findings_by_category[current_category].append(current_finding)
current_category = line.replace('CATEGORY:', '').strip()
current_finding = {}
elif line.startswith('TITLE:'):
current_finding['title'] = line.replace('TITLE:', '').strip()
elif line.startswith('DESCRIPTION:'):
current_finding['description'] = line.replace('DESCRIPTION:', '').strip()
elif line.startswith('PRIORITY:'):
current_finding['priority'] = line.replace('PRIORITY:', '').strip()
current_finding['evidence'] = "Business alignment analysis"
current_finding['impact'] = "Positive" if current_category in ["Strengths", "Opportunities"] else "Negative"
if current_finding and 'title' in current_finding:
findings_by_category[current_category].append(current_finding)
categories = []
for category, findings in findings_by_category.items():
if findings:
categories.append(SWOTCategory(
category=category,
domain="Business",
findings=findings
))
return categories
def _generate_executive_summary(self, swot_categories, arch_analysis, req_analyses) -> str:
"""Generate executive summary using LLM."""
key_findings = []
for category in swot_categories:
for finding in category.findings[:2]:
key_findings.append(f"{category.category}: {finding['title']}")
prompt = f"""Generate an executive summary for a software analysis report.
Key Findings:
{chr(10).join(f"- {f}" for f in key_findings)}
Write 3-4 paragraphs for senior management covering:
1. Overall system health
2. Critical strengths and weaknesses
3. Top improvement priorities
4. Business risk and opportunity"""
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.5
)
return response.choices[0].message.content
def _generate_prioritized_recommendations(self, swot_categories) -> List[Dict]:
"""Generate prioritized recommendations from SWOT analysis."""
issues = []
for category in swot_categories:
if category.category in ["Weaknesses", "Threats"]:
for finding in category.findings:
issues.append({
"domain": category.domain,
"title": finding['title'],
"priority": finding['priority']
})
issues_text = "\n".join(
f"{i+1}. [{issue['priority']}] {issue['domain']}: {issue['title']}"
for i, issue in enumerate(issues[:15])
)
prompt = f"""Generate actionable recommendations for these issues:
{issues_text}
For each major issue provide:
RECOMMENDATION: [Action]
IMPACT: [Expected improvement]
EFFORT: [High/Medium/Low]
PRIORITY: [Critical/High/Medium/Low]
---"""
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.4,
max_tokens=3000
)
recommendations = self._parse_recommendations(response.choices[0].message.content)
priority_order = {"Critical": 0, "High": 1, "Medium": 2, "Low": 3}
recommendations.sort(key=lambda r: priority_order.get(r.get('priority', 'Low'), 3))
return recommendations
def _parse_recommendations(self, response_text: str) -> List[Dict]:
"""Parse LLM response into structured recommendations."""
recommendations = []
current_rec = {}
for line in response_text.split('\n'):
line = line.strip()
if line.startswith('RECOMMENDATION:'):
current_rec['recommendation'] = line.replace('RECOMMENDATION:', '').strip()
elif line.startswith('IMPACT:'):
current_rec['impact'] = line.replace('IMPACT:', '').strip()
elif line.startswith('EFFORT:'):
current_rec['effort'] = line.replace('EFFORT:', '').strip()
elif line.startswith('PRIORITY:'):
current_rec['priority'] = line.replace('PRIORITY:', '').strip()
elif line == '---':
if current_rec and 'recommendation' in current_rec:
recommendations.append(current_rec)
current_rec = {}
if current_rec and 'recommendation' in current_rec:
recommendations.append(current_rec)
return recommendations
def _organize_swot(self, categories: List[SWOTCategory]) -> Dict:
"""Organize SWOT categories into structured format."""
swot = {
"Strengths": {},
"Weaknesses": {},
"Opportunities": {},
"Threats": {}
}
for category in categories:
if category.domain not in swot[category.category]:
swot[category.category][category.domain] = []
swot[category.category][category.domain].extend(category.findings)
return swot
def _format_architecture_findings(self, analysis) -> Dict:
"""Format architecture analysis findings."""
return {
"patterns": [p.value for p in analysis.identified_patterns],
"smells_count": len(analysis.architectural_smells),
"circular_dependencies_count": len(analysis.circular_dependencies),
"hotspots_count": len(analysis.hotspots)
}
def _format_requirements_findings(self, analyses) -> Dict:
"""Format requirements analysis findings."""
return {
"total_requirements": len(analyses),
"fully_implemented": sum(1 for a in analyses if a.status == RequirementStatus.FULLY_IMPLEMENTED),
"not_implemented": sum(1 for a in analyses if a.status == RequirementStatus.NOT_IMPLEMENTED)
}
def _generate_metrics_summary(self, arch_analysis, req_analyses) -> Dict:
"""Generate summary metrics."""
return {
"architecture": {
"total_components": len(arch_analysis.component_metrics),
"architectural_smells": len(arch_analysis.architectural_smells),
"hotspots": len(arch_analysis.hotspots)
},
"requirements": {
"total": len(req_analyses),
"fully_implemented_percent": sum(1 for a in req_analyses if a.status == RequirementStatus.FULLY_IMPLEMENTED) / len(req_analyses) * 100 if req_analyses else 0
}
}
def export_report(self, report: Dict, output_path: str, format: str = "json"):
"""Export report to file."""
if format == "json":
with open(output_path, 'w') as f:
json.dump(report, f, indent=2)
elif format == "markdown":
md_content = self._generate_markdown_report(report)
with open(output_path, 'w') as f:
f.write(md_content)
def _generate_markdown_report(self, report: Dict) -> str:
"""Generate Markdown formatted report."""
md = []
md.append("# Software System Analysis Report")
md.append(f"\nGenerated: {report['metadata']['generated_at']}\n")
md.append("## Executive Summary\n")
md.append(report['executive_summary'])
md.append("\n")
md.append("## SWOT Analysis\n")
for category in ["Strengths", "Weaknesses", "Opportunities", "Threats"]:
md.append(f"### {category}\n")
for domain, findings in report['swot_analysis'][category].items():
md.append(f"#### {domain}\n")
for finding in findings:
md.append(f"**{finding['title']}** (Priority: {finding['priority']})")
md.append(f"\n{finding['description']}\n")
md.append("## Recommendations\n")
for i, rec in enumerate(report['recommendations'], 1):
md.append(f"### {i}. {rec['recommendation']}")
md.append(f"\n- **Impact:** {rec['impact']}")
md.append(f"\n- **Effort:** {rec['effort']}")
md.append(f"\n- **Priority:** {rec['priority']}\n")
return "\n".join(md)
DEPLOYMENT CONSIDERATIONS AND GPU ARCHITECTURE OPTIMIZATION
Local Model Deployment Strategies
For organizations requiring on-premises deployment, the choice of GPU architecture significantly impacts performance and cost-effectiveness.
NVIDIA CUDA-Based Deployment:
from vllm import LLM, SamplingParams
class LocalLLMDeployment:
"""Manages local LLM deployment on NVIDIA GPUs."""
def __init__(self, model_name: str = "meta-llama/Llama-3.1-70B-Instruct",
tensor_parallel_size: int = 4):
"""Initialize local LLM deployment."""
self.llm = LLM(
model=model_name,
tensor_parallel_size=tensor_parallel_size,
gpu_memory_utilization=0.9,
dtype="bfloat16"
)
self.sampling_params = SamplingParams(
temperature=0.3,
top_p=0.9,
max_tokens=2048
)
def generate(self, prompts: List[str]) -> List[str]:
"""Generate responses for prompts."""
outputs = self.llm.generate(prompts, self.sampling_params)
return [output.outputs[0].text for output in outputs]
Hybrid Deployment Architecture:
from enum import Enum
class ModelTier(Enum):
"""Model deployment tiers."""
LOCAL_SMALL = "local_small"
LOCAL_LARGE = "local_large"
REMOTE_ADVANCED = "remote_advanced"
class HybridModelOrchestrator:
"""Orchestrates model selection based on task requirements."""
def __init__(self, local_deployment=None, openai_key=None, anthropic_key=None):
"""Initialize hybrid orchestrator."""
self.local_deployment = local_deployment
if openai_key:
self.openai_client = openai.OpenAI(api_key=openai_key)
if anthropic_key:
self.anthropic_client = anthropic.Anthropic(api_key=anthropic_key)
def select_model(self, task_type: str, context_length: int,
requires_reasoning: bool, privacy_sensitive: bool) -> ModelTier:
"""Select appropriate model tier for a task."""
if privacy_sensitive:
return ModelTier.LOCAL_LARGE if context_length > 50000 else ModelTier.LOCAL_SMALL
if context_length > 200000 or requires_reasoning:
return ModelTier.REMOTE_ADVANCED
return ModelTier.LOCAL_LARGE if context_length > 50000 else ModelTier.LOCAL_SMALL
def generate(self, prompt: str, task_type: str,
context_length: int = 4000,
requires_reasoning: bool = False,
privacy_sensitive: bool = False) -> str:
"""Generate response using appropriate model."""
tier = self.select_model(task_type, context_length, requires_reasoning, privacy_sensitive)
if tier in [ModelTier.LOCAL_SMALL, ModelTier.LOCAL_LARGE] and self.local_deployment:
return self.local_deployment.generate([prompt])[0]
elif tier == ModelTier.REMOTE_ADVANCED:
if requires_reasoning:
response = self.anthropic_client.messages.create(
model="claude-opus-4-20250514",
max_tokens=4000,
messages=[{"role": "user", "content": prompt}]
)
return response.content[0].text
else:
response = self.openai_client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
raise ValueError(f"No suitable model available for tier {tier}")
COMPLETE RUNNING EXAMPLE
from pathlib import Path
class CodeRepositoryAnalyzer:
"""Main orchestrator for complete code repository analysis."""
def __init__(self, repository_path: str, config: Dict):
"""Initialize the analyzer with all necessary components."""
self.repository_path = Path(repository_path)
self.config = config
self._initialize_infrastructure()
self._initialize_agents()
def _initialize_infrastructure(self):
"""Initialize knowledge graph, vector database, and embedding generator."""
print("Initializing infrastructure...")
self.knowledge_graph = KnowledgeGraphBuilder(
uri=self.config.get('neo4j_uri', 'bolt://localhost:7687'),
user=self.config.get('neo4j_user', 'neo4j'),
password=self.config.get('neo4j_password', 'password')
)
self.vector_db = VectorDatabaseManager(
host=self.config.get('qdrant_host', 'localhost'),
port=self.config.get('qdrant_port', 6333)
)
self.vector_db.initialize_collection()
self.embedding_generator = CodeEmbeddingGenerator(
api_key=self.config['openai_api_key']
)
print("Infrastructure initialized")
def _initialize_agents(self):
"""Initialize all specialized analysis agents."""
print("Initializing agents...")
self.architecture_agent = ArchitectureAnalysisAgent(
knowledge_graph=self.knowledge_graph,
vector_db=self.vector_db,
embedding_generator=self.embedding_generator,
llm_api_key=self.config['anthropic_api_key']
)
self.requirements_agent = RequirementsTraceabilityAgent(
knowledge_graph=self.knowledge_graph,
vector_db=self.vector_db,
embedding_generator=self.embedding_generator,
llm_api_key=self.config['openai_api_key']
)
self.report_generator = SWOTReportGenerator(
architecture_agent=self.architecture_agent,
requirements_agent=self.requirements_agent,
llm_api_key=self.config['openai_api_key']
)
print("Agents initialized")
def analyze(self, requirements: List[Requirement], business_context: str):
"""Run complete analysis and generate report."""
print("Starting complete repository analysis...")
# Parse repository
self.parse_repository()
# Generate comprehensive report
report = self.report_generator.generate_comprehensive_report(
requirements=requirements,
business_context=business_context
)
# Export report
self.report_generator.export_report(report, "analysis_report.json", format="json")
self.report_generator.export_report(report, "analysis_report.md", format="markdown")
print("Analysis complete! Reports generated.")
return report
def parse_repository(self):
"""Parse the entire repository."""
print(f"Parsing repository: {self.repository_path}")
python_parser = PythonCodeParser()
python_files = list(self.repository_path.rglob("*.py"))
print(f"Found {len(python_files)} Python files")
all_code_chunks = []
for py_file in python_files:
try:
module_info = python_parser.parse_file(str(py_file))
self.knowledge_graph.create_module_node(str(py_file), module_info)
chunks = self._create_code_chunks(py_file, module_info)
all_code_chunks.extend(chunks)
except Exception as e:
print(f"Error parsing {py_file}: {e}")
print(f"Generating embeddings for {len(all_code_chunks)} chunks...")
embeddings = self.embedding_generator.generate_batch_embeddings(all_code_chunks)
print("Storing in vector database...")
self.vector_db.store_batch(all_code_chunks, embeddings)
print("Repository parsing complete")
def _create_code_chunks(self, file_path: Path, module_info: Dict) -> List[CodeChunk]:
"""Create code chunks from parsed module information."""
chunks = []
parts = file_path.parts
subsystem = parts[-3] if len(parts) >= 3 else "unknown"
component = parts[-2] if len(parts) >= 2 else "unknown"
for class_info in module_info.get('classes', []):
chunk = CodeChunk(
content=self._extract_class_code(file_path, class_info),
file_path=str(file_path),
language="python",
chunk_type="class",
start_line=class_info.start_line,
end_line=class_info.end_line,
entity_name=class_info.name,
component=component,
subsystem=subsystem,
complexity=sum(m.complexity for m in class_info.methods)
)
chunks.append(chunk)
return chunks
def _extract_class_code(self, file_path: Path, class_info) -> str:
"""Extract source code for a class."""
with open(file_path, 'r') as f:
lines = f.readlines()
return ''.join(lines[class_info.start_line-1:class_info.end_line])
# Usage example
if __name__ == "__main__":
config = {
'neo4j_uri': 'bolt://localhost:7687',
'neo4j_user': 'neo4j',
'neo4j_password': 'your_password',
'qdrant_host': 'localhost',
'qdrant_port': 6333,
'openai_api_key': 'your_openai_key',
'anthropic_api_key': 'your_anthropic_key'
}
requirements = [
Requirement(
id="REQ-001",
title="User Authentication",
description="System shall provide secure user authentication",
type=RequirementType.FUNCTIONAL,
priority="High",
source="Business Requirements",
acceptance_criteria=["Users can log in with email/password", "Sessions expire after 24 hours"]
)
]
business_context = """
E-commerce platform serving 1M+ users.
Strategic goals: Improve conversion rate, reduce cart abandonment, expand to mobile.
"""
analyzer = CodeRepositoryAnalyzer(
repository_path="/path/to/repository",
config=config
)
report = analyzer.analyze(requirements, business_context)
print("Analysis complete!")
CONCLUSION AND FUTURE DIRECTIONS
This guide has presented a detailed approach to building LLM-based multi-agent AI systems for analyzing large polyglot code repositories. The system addresses fundamental limitations of current AI coding assistants through:
1. Hierarchical knowledge representation that overcomes context window limitations
2. Hybrid GraphRAG architecture combining vector databases with knowledge graphs
3. Rigorous fact-checking using retrieval-augmented generation
4. Specialized agent architecture with domain-expert agents
5. Polyglot code understanding using Tree-sitter
6. Comprehensive traceability linking business goals to implementation
7. SWOT analysis generation synthesizing technical findings into business insights
Future Directions:
* Autonomous Remediation Agents that automatically implement improvements
* Continuous Architectural Governance with real-time drift detection
* Predictive Analytics forecasting architectural challenges
* Multi-Repository Analysis across enterprise boundaries
* Domain-Specific Customization for healthcare, finance, etc.
* Interactive Exploration through conversational interfaces
* Development Workflow Integration with IDEs and CI/CD pipelines
The multi-agent approach represents a significant evolution in code analysis, combining LLM semantic understanding with structured knowledge representation to provide comprehensive, accurate, and actionable insights for enterprise software systems.
No comments:
Post a Comment