CHAPTER 1: THE IMPERATIVE FOR ADVANCED CODE INTELLIGENCE
Understanding the Modern Software Engineering Crisis
In contemporary software engineering organizations, codebases have evolved into sprawling ecosystems that defy simple comprehension. A typical enterprise repository might contain millions of lines of code written in a dozen programming languages, accumulated over years or decades of development by hundreds of contributors. The knowledge about why certain architectural decisions were made, how business requirements map to implementation details, and where technical debt lurks often exists only in fragmented form across scattered documentation, aging Architecture Decision Records, incomplete test suites, and most critically, in the minds of engineers who may have long since moved to other projects or organizations.
Senior management faces a persistent challenge when attempting to understand the true state of their software assets. When a Chief Technology Officer asks seemingly straightforward questions such as "What would it take to migrate our payment processing system to a new regulatory framework?" or "How much technical debt are we carrying in our customer-facing services?", the answers typically require weeks of manual investigation by senior architects and developers. This investigation involves reading through thousands of files, tracing dependencies across module boundaries, interviewing team members about undocumented design decisions, and synthesizing information from disparate sources. The resulting analysis is expensive, time-consuming, and often incomplete because human cognitive limitations prevent anyone from holding the entire system's complexity in their mind simultaneously.
Business managers struggle to connect strategic initiatives to technical implementation realities. When market conditions demand rapid feature development or system modernization, the disconnect between business timelines and technical feasibility creates friction. The question "Can we add multi-currency support to our e-commerce platform in time for the European expansion?" requires deep technical analysis that business stakeholders cannot perform themselves, yet they need accurate answers to make informed decisions. Traditional code analysis tools provide metrics and dependency graphs, but they cannot answer business-oriented questions or explain the implications of technical constraints in terms that non-technical stakeholders can understand.
Architects and senior developers carry enormous cognitive burdens. They must maintain mental models of system architecture, understand the rationale behind past design decisions, track technical debt accumulation, and evaluate the impact of proposed changes. When a new architect joins a project, the knowledge transfer process can take months or years. Much of the critical architectural knowledge exists as tacit understanding rather than explicit documentation. The architect who designed the original microservices decomposition might remember that Service A was split from Service B because of anticipated scaling requirements that never materialized, but this rationale rarely appears in any document. Without this context, future maintainers may perpetuate suboptimal patterns or make changes that violate unstated architectural principles.
The Current Landscape of AI-Assisted Development Tools
The emergence of AI-powered coding assistants has transformed certain aspects of software development, yet these tools reveal significant limitations when confronting the challenges of large-scale code comprehension and analysis. Tools such as GitHub Copilot, which leverages models from the GPT family, excel at generating code snippets based on immediate context. When a developer begins writing a function, Copilot can suggest completions that are often syntactically correct and contextually relevant within the narrow scope of the current file. This capability dramatically accelerates routine coding tasks and helps developers explore API usage patterns.
However, Copilot's strength in local code generation becomes a weakness when broader system understanding is required. The tool operates primarily within the context window of the current file or perhaps a few related files. It cannot reason about architectural patterns that span dozens of modules, understand the business logic distributed across multiple microservices, or recognize that a seemingly innocuous change in one component will violate an architectural constraint documented in an ADR from three years ago. When a developer asks Copilot to refactor a complex subsystem, the suggestions may be locally optimal but globally problematic because the tool lacks awareness of system-wide invariants, cross-cutting concerns, and the historical context that shaped the current design.
Cursor, which builds upon similar foundation models with enhanced IDE integration, provides more sophisticated code navigation and editing capabilities. It can analyze multiple files simultaneously and offer refactoring suggestions that maintain consistency across related components. Yet Cursor still operates within the fundamental constraints of transformer-based models, particularly the context window limitation. Even with extended context windows available in newer models, a context window of one hundred thousand tokens might accommodate only a fraction of a medium-sized repository. The tool cannot maintain a comprehensive understanding of a codebase containing millions of lines across hundreds of modules. When asked to explain how a specific business requirement is implemented across the entire system, Cursor can provide insights about individual components but struggles to synthesize a complete picture that traces the requirement from user interface through business logic to data persistence and back.
Anthropic's Claude, particularly the recent iterations like Claude 4.5 Sonnet, represents significant advances in reasoning capability and context handling. Claude demonstrates superior performance in understanding complex technical documentation and can engage in more nuanced discussions about code design trade-offs. The model's constitutional AI training makes it more reliable in acknowledging uncertainty and less prone to confidently presenting hallucinated information. Nevertheless, even Claude faces the same fundamental challenge that affects all current large language models when applied to comprehensive codebase analysis. The model processes information within a fixed context window, and while this window has grown substantially larger in recent generations, it remains insufficient for holding the complete state of an enterprise application along with all its associated documentation, test suites, architectural decisions, and historical context.
Windsurf and similar tools attempt to address these limitations through various strategies such as intelligent context selection, where the tool attempts to identify and load only the most relevant portions of the codebase for a given query. These approaches help but introduce new problems. The context selection mechanism itself must make decisions about relevance, and these decisions can exclude critical information that would change the analysis outcome. A query about performance optimization might focus the tool's attention on algorithmic complexity in business logic while missing the fact that the real bottleneck is an architectural decision to use synchronous communication where asynchronous messaging would be more appropriate, a decision documented in an ADR that the context selection algorithm deemed irrelevant to the immediate query.
All these tools share another critical limitation related to the nature of their training and operation. They are fundamentally pattern-matching systems that have learned statistical relationships between code patterns from their training data. When confronted with domain-specific business logic, proprietary architectural patterns, or organization-specific design conventions, these tools fall back on generic programming knowledge. A financial services application might use domain-specific terminology and implement complex business rules that reflect regulatory requirements. Generic coding assistants cannot understand these domain concepts without explicit context, and providing sufficient context often exceeds the available context window. The tool might suggest refactoring that makes the code more "clean" according to general programming principles while actually obscuring the business logic and making it harder for domain experts to verify correctness.
Why Comprehensive Multi-Agent Analysis Systems Are Essential
The limitations of current AI coding assistants point toward the need for a fundamentally different approach to AI-assisted code comprehension and analysis. Rather than attempting to compress all repository knowledge into a single model's context window, a multi-agent system distributes the analysis task across specialized agents, each focusing on a specific aspect of the software system. This architectural approach mirrors how human organizations structure software engineering work, with different specialists handling architecture, domain modeling, testing, operations, and project management.
A multi-agent system can maintain persistent knowledge about the codebase in structured forms that complement the statistical knowledge embedded in language models. By combining traditional knowledge representation techniques such as knowledge graphs and abstract syntax trees with modern vector databases and retrieval-augmented generation, the system creates a multi-layered understanding of the software that no single model could hold in its context window. When a stakeholder asks a question about the system, specialized agents collaborate to gather relevant information from different knowledge sources, synthesize this information, and provide an answer that reflects both broad system understanding and deep technical detail.
Consider a scenario where a product manager asks whether the system can support a new feature that requires processing customer data according to newly enacted privacy regulations. A human architect addressing this question would need to understand the current data flow architecture, identify all locations where customer data is processed or stored, review existing privacy controls, assess the gap between current capabilities and new requirements, estimate the effort required to implement necessary changes, and identify risks associated with the implementation. This analysis requires knowledge that spans business requirements, architectural design, implementation details, testing coverage, and operational considerations.
A multi-agent analysis system can orchestrate this investigation by deploying specialized agents in coordination. A requirements agent examines existing privacy-related requirements and compares them with the new regulatory framework. An architecture agent analyzes the system's data flow patterns and identifies all components that handle customer data. A code analysis agent examines the actual implementation to verify that architectural intentions are reflected in the code and identifies any discrepancies. A testing agent assesses whether existing tests cover privacy-related functionality and where test coverage gaps exist. A domain modeling agent ensures that the analysis uses consistent terminology and correctly interprets domain concepts. A knowledge graph ties these different perspectives together, creating a comprehensive view that individual agents can query and update.
The multi-agent approach also addresses the hallucination problem that plagues single-model systems. When a language model generates information beyond its training data or context, it may produce plausible-sounding but factually incorrect statements. In a multi-agent system, specialized agents can cross-validate information against multiple sources. If a code analysis agent claims that a particular module implements a specific business rule, a testing agent can verify whether tests exist that validate this behavior, a requirements agent can check whether this business rule appears in requirements documentation, and a knowledge graph query can confirm whether the claimed relationships between components actually exist in the parsed code structure. This multi-source validation dramatically reduces the risk of hallucinated information making its way into analysis reports.
The system must also address the challenge of information that exists only in human minds. Many critical architectural decisions, design rationales, and domain knowledge never get documented. A multi-agent system can incorporate a human-in-the-loop pattern where agents explicitly identify knowledge gaps and formulate specific questions for human experts. Rather than asking vague questions like "Can you explain the architecture?", the system can ask targeted questions such as "The code analysis shows that Service A makes synchronous calls to Service B even though the architecture documentation describes an event-driven pattern. Was this a temporary implementation decision, and if so, what were the reasons?" This focused questioning makes efficient use of human expert time and helps capture tacit knowledge in a structured form that becomes part of the system's knowledge base.
The Information Landscape of Software Repositories
Understanding what information exists within and around a software repository is fundamental to designing an effective analysis system. The most obvious information source is the code itself, but code is merely the tip of the information iceberg. Each line of code represents decisions made by developers, but the rationale for those decisions rarely appears in the code. Comments and documentation within code files provide some context, but they often become outdated as code evolves, and they typically focus on implementation details rather than strategic architectural reasoning.
Architecture Decision Records represent a valuable but often incomplete source of architectural knowledge. When teams maintain ADRs consistently, these documents capture the context, decision, and consequences of significant architectural choices. An ADR might explain that the team chose a microservices architecture to enable independent scaling of different system components, accepting the increased operational complexity as a necessary trade-off. However, many organizations adopt ADRs inconsistently, documenting some decisions while leaving others implicit. The multi-agent system must work with partial ADR coverage, using the available records to inform its understanding while recognizing that undocumented decisions also exist.
Test suites provide a unique form of documentation because tests encode behavioral expectations in executable form. A comprehensive test suite effectively documents how the system should behave under various conditions. Integration tests reveal how components interact, while unit tests document the expected behavior of individual functions and classes. However, test coverage is rarely complete, and tests themselves can contain bugs or reflect outdated understanding of requirements. A multi-agent system must analyze tests not just to understand intended behavior but also to identify gaps in test coverage that might indicate areas of the codebase where behavior is poorly understood or inadequately validated.
Requirements specifications exist in various forms and varying levels of detail across organizations. Some projects maintain formal requirements documents that trace business needs through functional specifications to acceptance criteria. Other projects capture requirements as user stories in issue tracking systems, with the detailed requirements emerging through conversation and implementation. The multi-agent system must be able to extract requirements information from diverse sources and formats, recognizing that requirements often exist in a state of partial documentation where some aspects are explicit while others remain implicit assumptions shared among team members.
Meeting notes, email threads, and chat conversations contain valuable context about decisions and discussions, but this information is typically unstructured and difficult to search. A discussion in a Slack channel might reveal that a particular implementation approach was chosen because of time pressure rather than technical merit, with the team acknowledging that refactoring would be needed later. This context is crucial for understanding technical debt, but it exists in a form that traditional code analysis tools cannot process. The multi-agent system needs natural language processing capabilities to extract relevant information from these unstructured sources.
Repository metadata provides another rich information source. The Git history shows how code has evolved over time, revealing patterns such as files that change frequently together, suggesting high coupling, or files that change very frequently, suggesting either active development or instability. Issue tracking systems contain bug reports that identify problematic areas of the codebase. Pull request discussions capture code review feedback and design debates. Deployment logs and operational metrics reveal how the system behaves in production. All this metadata contributes to a comprehensive understanding of the system's health and evolution.
Domain models represent perhaps the most critical yet often least explicit information source. The problem domain encompasses the business concepts, rules, and processes that the software must support. The solution domain describes the technical concepts, patterns, and structures used to implement the software. Effective software engineering requires maintaining clear mappings between these domains, but these mappings often exist only in developers' minds. A multi-agent system must reconstruct domain models from available evidence, identifying business concepts mentioned in requirements and code, recognizing domain-specific terminology, and building a structured representation of how the problem domain maps to the solution domain.
External dependencies represent another information dimension that affects system understanding. Modern applications rely on numerous open-source libraries, commercial components, and external services. Understanding these dependencies is crucial for assessing security vulnerabilities, license compliance, upgrade impacts, and architectural constraints. The multi-agent system must track not just direct dependencies but transitive dependencies, understanding how changes in external components might ripple through the system.
The Path Forward
Building a multi-agent system that can effectively analyze large, complex codebases requires integrating multiple technologies and approaches. The system must parse code in multiple programming languages to build abstract syntax trees and extract structural information. It must use natural language processing to extract information from documentation, requirements, and unstructured text. It must employ knowledge graphs to represent relationships between code entities, requirements, architectural elements, and domain concepts. It must use vector databases and retrieval-augmented generation to enable semantic search across large document collections. It must orchestrate multiple language models, potentially using different models for different tasks based on their specific strengths. It must implement reasoning strategies such as chain-of-thought and self-reflection to improve the quality of its analysis. It must provide mechanisms for human experts to contribute knowledge and validate findings.
The remainder of this article explores how to build such a system, providing both conceptual understanding and concrete implementation guidance. We will examine the architectural patterns that enable effective multi-agent collaboration, the techniques for extracting and representing knowledge from diverse sources, the strategies for managing LLM limitations, and the approaches for generating comprehensive analysis reports that serve the needs of all stakeholder groups from executives to developers.
CHAPTER 2: ARCHITECTURAL VISION AND DOMAIN MODELING FOUNDATIONS
Conceptualizing the Multi-Agent Architecture
The architecture of an effective code repository analysis system must reflect the multifaceted nature of software understanding itself. Just as a human expert team analyzing a complex system would divide responsibilities among specialists in architecture, domain modeling, testing, and operations, our multi-agent system distributes analytical tasks across purpose-built agents that collaborate through well-defined protocols. This distribution serves multiple critical purposes beyond simply dividing labor. It allows each agent to maintain focused context relevant to its specialty, avoiding the dilution of attention that occurs when a single model attempts to simultaneously reason about business strategy, low-level code implementation, test coverage, and operational concerns.
The central organizing principle of this architecture is the separation between knowledge acquisition, knowledge representation, and knowledge synthesis. Knowledge acquisition agents specialize in extracting information from specific sources. A code parsing agent understands how to transform source code in various programming languages into structured representations. A documentation analysis agent processes natural language text from architecture documents, requirements specifications, and design notes. A repository metadata agent extracts information from Git history, issue trackers, and pull request discussions. Each acquisition agent produces structured output in standardized formats that feed into the knowledge representation layer.
The knowledge representation layer maintains multiple complementary views of the system under analysis. A knowledge graph captures entities and relationships, representing elements such as software modules, classes, functions, requirements, architectural components, and domain concepts as nodes, with edges representing relationships like "implements," "depends on," "satisfies," "belongs to," and "communicates with." This graph structure enables traversal queries that answer questions like "which code components implement requirement R-42" or "what are all the dependencies of the payment processing module." The knowledge graph provides a factual foundation that grounds the system's understanding in verifiable structural relationships rather than statistical patterns learned during language model training.
Complementing the knowledge graph, vector databases store embeddings of code chunks, documentation sections, and other textual artifacts. These embeddings enable semantic similarity search, allowing the system to find relevant information even when exact keyword matches do not exist. When an analyst asks about "customer data privacy controls," the vector database can retrieve relevant code sections that implement privacy features even if they use different terminology like "personal information protection" or "data access restrictions." The combination of graph-based structural queries and vector-based semantic search provides both precision and recall in information retrieval.
A hierarchical summarization structure addresses the context window limitation that constrains language model reasoning. Individual functions and methods are summarized into class-level descriptions. Classes are summarized into module descriptions. Modules are summarized into subsystem descriptions. This hierarchy allows agents to reason at appropriate levels of abstraction. When analyzing high-level architectural questions, agents work with subsystem summaries rather than drowning in implementation details. When investigating specific implementation issues, agents can drill down through the hierarchy to access detailed code-level information. The hierarchical structure also enables efficient updates when code changes, as modifications to a single function require updating only the summaries in the path from that function to the root of the hierarchy, rather than reprocessing the entire codebase.
Knowledge synthesis agents consume information from the representation layer to answer questions and generate insights. These agents implement sophisticated reasoning patterns that go beyond simple information retrieval. A requirements traceability agent might trace a business requirement through architectural design decisions to implementation components and test cases, synthesizing information from multiple knowledge sources to provide a complete picture of how a requirement is realized in the system. An architectural quality agent might analyze the system against known architectural patterns and quality attributes, identifying where the implementation aligns with or deviates from stated architectural intentions.
The orchestration layer coordinates agent activities to accomplish complex analytical tasks. When a user poses a question or requests an analysis report, the orchestrator decomposes this high-level goal into subtasks, assigns subtasks to appropriate agents, manages dependencies between subtasks, and synthesizes results from multiple agents into coherent responses. The orchestrator also implements human-in-the-loop patterns, identifying when agent analysis encounters gaps in available information and formulating specific questions for human experts.
Understanding Domain Modeling in the Context of Code Analysis
Domain modeling forms the conceptual bridge between business understanding and technical implementation. The problem domain encompasses the real-world business concepts, processes, rules, and constraints that the software must address. In a financial trading system, the problem domain includes concepts like securities, orders, trades, portfolios, market data, and regulatory requirements. These concepts exist independently of any software implementation; they represent the business reality that the software must model and support.
The solution domain comprises the technical concepts, structures, and patterns used to implement the software. In the same trading system, the solution domain includes technical elements like microservices, message queues, databases, APIs, caching layers, and deployment containers. Solution domain concepts are chosen and structured to effectively represent and manipulate problem domain concepts while satisfying technical requirements like performance, scalability, and maintainability.
Effective software engineering requires maintaining clear mappings between these domains. A well-designed system exhibits clear correspondence between problem domain concepts and solution domain structures. The "Order" concept from the problem domain might map to an Order class in the code, an orders table in the database, an OrderService that implements order-related business logic, and a set of test cases that validate order processing behavior. When these mappings are clear and consistent, the system is easier to understand, maintain, and evolve because changes in business requirements have predictable impacts on technical implementation.
However, in real-world systems, especially those that have evolved over years with changing requirements and multiple development teams, these mappings often become obscured. A business concept might be implemented across multiple technical components with no clear organizing structure. The "Customer" concept might be represented differently in different parts of the system, with customer data scattered across multiple databases, customer-related logic distributed across numerous services, and no single authoritative source for customer information. Understanding these fragmented implementations requires reconstructing the implicit domain model from the code and other available artifacts.
A multi-agent analysis system must perform this domain model reconstruction as a foundational activity. A domain analysis agent examines requirements documents, architecture documentation, and code to identify problem domain concepts. Natural language processing techniques extract domain terminology from text. Code analysis identifies classes, modules, and data structures that appear to represent domain concepts based on naming conventions and structural patterns. The agent builds a problem domain model that catalogs identified concepts and their relationships.
Simultaneously, a solution domain agent analyzes the technical architecture, identifying the major structural elements like services, modules, layers, and components. This agent understands common architectural patterns such as layered architecture, microservices, event-driven architecture, and domain-driven design tactical patterns. By recognizing these patterns in the codebase, the agent can infer architectural intentions even when explicit documentation is incomplete.
The critical analytical step involves mapping between these domain models. A traceability agent examines how problem domain concepts are realized in solution domain structures. This mapping reveals important characteristics of the system's design quality. Strong, clear mappings indicate good domain alignment where the code structure reflects business concepts, making the system more maintainable. Weak or scattered mappings indicate potential architectural issues where business concepts are poorly represented in the technical implementation.
Consider a concrete example from an e-commerce system. The problem domain includes concepts like Product, Customer, ShoppingCart, Order, Payment, and Shipment. In a well-designed system, we would expect to find corresponding solution domain elements. There might be a Product service that manages product information, a Customer service that handles customer data and authentication, a ShoppingCart service that maintains cart state, an OrderProcessing service that coordinates order fulfillment, a Payment service that integrates with payment gateways, and a Shipping service that manages logistics.
The domain analysis agent would identify these problem domain concepts by analyzing requirements documents that discuss product catalogs, customer registration, shopping cart functionality, order placement, payment processing, and shipment tracking. The code analysis agent would parse the codebase and identify services, classes, and modules with names suggesting they implement these concepts. The knowledge graph would represent both problem domain concepts and solution domain elements as nodes, with edges representing the "implements" relationship between them.
However, the analysis might reveal complications. Perhaps the Payment concept is implemented across three different services because the system evolved from a monolithic architecture to microservices, and payment logic was never fully extracted into a single service. The ShoppingCart concept might be implemented partly in client-side JavaScript and partly in a server-side session management system, with no clear service ownership. These discoveries represent important architectural findings that should appear in the analysis report as potential weaknesses requiring attention.
The domain modeling capability also enables the system to use appropriate terminology when communicating with different stakeholders. When generating reports for business managers, the system can describe findings using problem domain terminology that business stakeholders understand. When communicating with developers, the system can use solution domain terminology that reflects the actual code structure. The ability to translate between these vocabularies makes the system valuable across the entire stakeholder spectrum.
Specialized Agent Roles and Responsibilities
Each agent in the multi-agent system carries specific responsibilities aligned with a particular aspect of software understanding. The code analysis agent serves as the foundation for technical understanding. This agent must parse source code in multiple programming languages, extracting structural information into abstract syntax trees. For Python code, the agent might use the built-in ast module to parse source files into ASTs that represent the syntactic structure of the code. For other languages, the agent employs appropriate parsing tools. The tree-sitter parsing library provides a unified interface for parsing many languages, generating consistent AST representations regardless of the source language.
Once code is parsed into ASTs, the code analysis agent extracts entities and relationships. It identifies classes, interfaces, functions, methods, and their relationships such as inheritance, composition, and method calls. This structural information populates the knowledge graph, creating nodes for each code entity and edges for each relationship. The agent also generates embeddings for code chunks, storing them in the vector database to enable semantic code search.
Beyond basic structural extraction, the code analysis agent performs deeper analysis to identify patterns and potential issues. It detects code smells like long methods, large classes, and duplicated code. It identifies architectural patterns like factory methods, observers, and decorators. It analyzes dependencies to detect circular dependencies, inappropriate coupling between modules, and violations of layering principles. All these findings feed into the knowledge base, contributing to the comprehensive understanding of the system's technical health.
The architecture analysis agent operates at a higher level of abstraction, focusing on system-wide structural concerns rather than individual code elements. This agent examines how modules and components are organized, how they communicate, and how they align with stated architectural principles. If architecture documentation describes a layered architecture with presentation, business logic, and data access layers, the architecture agent verifies whether the actual code respects these layer boundaries or whether there are violations where presentation code directly accesses data access components.
The architecture agent also analyzes the system against quality attributes. If requirements specify that the system must be scalable, the architecture agent examines architectural patterns to assess scalability characteristics. Does the system use stateless services that can be horizontally scaled? Are there bottlenecks like shared mutable state or single-point-of-failure components? The agent generates quality attribute scenarios that describe how the system behaves under various conditions, comparing these scenarios against requirements to identify gaps.
A specialized quality attribute agent focuses specifically on non-functional requirements like performance, security, reliability, and maintainability. This agent analyzes code for security vulnerabilities, examining patterns like SQL query construction to detect potential SQL injection risks, or reviewing authentication and authorization logic to identify security weaknesses. It assesses performance characteristics by identifying potentially expensive operations like nested loops with high complexity or database queries executed within loops. It evaluates maintainability by analyzing code metrics like cyclomatic complexity, coupling, and cohesion.
The requirements analysis agent processes requirements documentation in various forms. It extracts functional requirements, constraints, and quality attributes from requirements documents, user stories, and acceptance criteria. Using natural language processing, it identifies key concepts, actors, actions, and conditions mentioned in requirements. It builds a requirements model that represents requirements as structured entities in the knowledge graph, enabling traceability analysis that links requirements to design elements, code components, and test cases.
This agent also performs requirements quality analysis, identifying ambiguous, incomplete, or conflicting requirements. If two requirements specify contradictory behavior, the agent flags this conflict for human review. If a requirement uses vague terms like "fast" or "user-friendly" without quantifiable criteria, the agent identifies this ambiguity as a potential issue.
The domain analysis agent, as discussed earlier, focuses on identifying and modeling problem domain concepts. This agent employs named entity recognition techniques to extract domain terminology from various sources. It builds a domain vocabulary that catalogs terms and their definitions, helping ensure consistent terminology usage across the analysis. When different documents use different terms for the same concept, the agent identifies these synonyms and establishes canonical terminology.
The test and quality agent analyzes test suites to understand what behavior is validated and where test coverage gaps exist. It examines unit tests to understand the expected behavior of individual functions and classes. It analyzes integration tests to understand how components interact. It reviews system tests and acceptance tests to understand end-to-end behavior scenarios. By mapping tests to code coverage, the agent identifies untested or under-tested portions of the codebase, which represent risk areas where behavior is not validated.
This agent also analyzes test quality, looking for test smells like tests that are too tightly coupled to implementation details, tests that test multiple concerns simultaneously, or tests with inadequate assertions that do not actually validate meaningful behavior. High-quality tests serve as valuable documentation of intended behavior, while poor-quality tests provide false confidence without actually validating correctness.
The repository and project management agent analyzes metadata from version control, issue tracking, and project management systems. It examines Git history to identify hotspots where code changes frequently, suggesting either active development or instability. It analyzes commit patterns to understand which files tend to change together, revealing coupling that might not be apparent from static code analysis. It processes issue tracker data to identify which components have the most bug reports, indicating quality problems. It examines pull request discussions to extract design rationale and technical decisions that might not appear in formal documentation.
The operations and deployment agent focuses on how the system runs in production environments. It analyzes deployment configurations, infrastructure-as-code definitions, and operational runbooks. It examines monitoring and logging configurations to understand what operational metrics are tracked. It reviews incident reports and post-mortem documents to understand what operational problems have occurred and how they were resolved. This operational perspective is crucial for understanding the complete system lifecycle beyond just the code.
Technology-specific agents provide deep expertise in particular technologies that are critical to the system under analysis. If the system uses a complex technology like Apache Kafka for event streaming, a Kafka-specialized agent can analyze how Kafka is configured and used, identifying best practices and potential issues specific to Kafka. If the system uses a particular database technology, a database-specialized agent can analyze schema design, query patterns, and database-specific optimization opportunities.
The knowledge graph agent maintains the graph database that represents entities and relationships. This agent provides query capabilities that other agents use to traverse the knowledge graph, finding related entities and analyzing relationship patterns. It also implements graph algorithms like centrality analysis to identify critical components that have many dependencies, or community detection to identify clusters of highly related components that might represent subsystems.
The summary agent generates hierarchical summaries of code and documentation. Starting with detailed information at the lowest level, it progressively abstracts information into higher-level summaries. A method summary describes what the method does and its key parameters and return value. A class summary describes the class's purpose and its key methods. A module summary describes the module's role and its key classes. This hierarchical summarization enables efficient reasoning at multiple levels of abstraction.
The report generation agent synthesizes information from all other agents to produce comprehensive analysis reports. This agent understands the structure and content requirements for different report types. When generating a SWOT analysis, it gathers findings from various specialized agents, categorizes them into strengths, weaknesses, opportunities, and threats, evaluates their criticality, and organizes them into a coherent report structure. It generates executive summaries that highlight key findings for senior management, and detailed technical sections that provide the depth developers and architects need.
Orchestrating Agent Collaboration Through Reasoning Patterns
The power of the multi-agent system emerges not just from individual agent capabilities but from their coordinated collaboration. The orchestration layer implements sophisticated reasoning patterns that enable agents to work together effectively. Chain-of-thought reasoning breaks complex analytical tasks into sequences of simpler steps. When asked to evaluate whether the system can support a new feature, the orchestrator might chain together a sequence of agent invocations. First, the requirements agent analyzes the feature description to extract specific functional and non-functional requirements. Then, the architecture agent examines whether the current architecture can accommodate these requirements. Next, the code analysis agent identifies which components would need modification. The test agent assesses what new tests would be needed. Finally, the synthesis agent combines these perspectives into an integrated feasibility assessment.
Tree-of-thought reasoning explores multiple analytical paths in parallel, evaluating different hypotheses or approaches. When investigating a performance problem, the system might explore multiple potential causes simultaneously. One branch investigates algorithmic complexity in business logic. Another examines database query efficiency. A third looks at network communication patterns. Each branch pursues its hypothesis, gathering evidence. The orchestrator evaluates which branches yield the most promising findings and may prune branches that lead to dead ends, focusing computational resources on the most fruitful investigations.
Self-reflection enables agents to evaluate and improve their own reasoning. After generating an analysis, an agent can reflect on the quality and completeness of its work. Did it consider all relevant information sources? Are there gaps in its reasoning? Are its conclusions well-supported by evidence? This self-reflection might identify the need for additional analysis or highlight uncertainties that require human expert input.
The human-in-the-loop pattern recognizes that no automated system can fully replace human expertise, especially for tacit knowledge that was never documented. When agents encounter knowledge gaps, they formulate specific questions for human experts rather than making unsupported assumptions. The orchestrator maintains a question queue, prioritizing questions by their importance to the overall analysis. It presents questions to appropriate human experts based on their roles and expertise. When experts provide answers, this new knowledge is incorporated into the knowledge base, improving future analyses.
Validation through multiple sources reduces hallucination risk. When an agent makes a claim, the orchestrator seeks corroborating evidence from independent sources. If the code analysis agent claims that a particular class implements a specific business rule, the test agent checks whether tests validate this behavior, and the requirements agent verifies whether this business rule appears in requirements documentation. Only claims supported by multiple independent sources are presented as high-confidence findings. Claims supported by single sources are flagged as needing verification.
Addressing the Context Window Challenge Through Intelligent Information Management
The context window limitation represents perhaps the most fundamental constraint when working with large language models. Even the most advanced models available today, including GPT-5.1 with its extended context capabilities, Gemini 3 Pro with its multi-modal processing, and Claude 4.5 Sonnet with its enhanced reasoning, cannot hold an entire enterprise codebase in context simultaneously. A million-line codebase might contain tens of millions of tokens, far exceeding any current or near-future context window.
The multi-agent architecture addresses this limitation through several complementary strategies. The hierarchical summarization structure, as mentioned earlier, allows reasoning at appropriate abstraction levels. When analyzing high-level architectural questions, agents work with subsystem and module summaries that fit comfortably within context windows. Only when drilling into specific implementation details do agents load detailed code-level information.
The knowledge graph provides a navigation structure that guides selective information loading. When an agent needs to understand a particular component, it queries the knowledge graph to identify related entities. It loads summaries of directly related components rather than attempting to load the entire codebase. If deeper understanding is needed, it can traverse the graph further, loading additional related entities. This graph-guided exploration ensures that loaded information is relevant to the current analytical task.
Vector database retrieval enables semantic search that finds relevant information without requiring exhaustive scanning. When an agent needs information about a particular topic, it generates a query embedding and retrieves the most semantically similar chunks from the vector database. This retrieval is highly efficient, returning relevant information from a corpus of millions of chunks in milliseconds. The retrieved chunks provide focused context for the language model to reason about.
Specialized lightweight prompts replace monolithic complex prompts. Rather than constructing a single enormous prompt that attempts to provide all context and ask all questions simultaneously, the system uses sequences of focused prompts. Each prompt addresses a specific analytical subtask with just the context needed for that subtask. The results from multiple lightweight prompts are combined to answer complex questions. This approach not only fits within context windows but also improves reasoning quality because the model can focus on one aspect at a time without the cognitive load of simultaneously considering multiple concerns.
Incremental processing handles codebase updates efficiently. When code changes, the system does not reprocess the entire repository. Instead, it identifies which files changed, updates the AST representations for those files, updates affected entries in the knowledge graph, regenerates embeddings for modified code chunks, and propagates updates through the summarization hierarchy. This incremental approach makes the system practical for continuous use in active development environments where code changes frequently.
CHAPTER 3: TECHNICAL IMPLEMENTATION FOUNDATIONS
Building the Code Parsing Infrastructure
The foundation of any code analysis system rests on its ability to transform source code from text into structured representations that enable programmatic analysis. For Python code, the language's built-in ast module provides robust parsing capabilities that convert source code into abstract syntax trees. An abstract syntax tree represents the hierarchical structure of code, with nodes representing language constructs like classes, functions, statements, and expressions, and edges representing the containment relationships between these constructs.
When the code analysis agent encounters a Python file, it reads the source code as text and invokes the ast module to parse it. The parsing process validates that the code is syntactically correct Python and produces a tree structure that the agent can traverse programmatically. Consider a simple Python class that represents a customer in an e-commerce system. The source code might define a Customer class with methods for updating contact information and calculating loyalty points. When parsed, this code becomes a tree where the root represents the module, a child node represents the class definition, and grandchildren represent individual method definitions.
The agent traverses this tree to extract meaningful information. It identifies the class name, which becomes an entity in the knowledge graph. It examines the class's base classes to identify inheritance relationships, creating edges in the knowledge graph that connect the class to its parents. It iterates through method definitions, creating entities for each method and edges that connect methods to their containing class. For each method, it extracts the parameter list, return type annotations if present, and docstring documentation. It analyzes the method body to identify what other classes and functions the method calls, creating dependency edges in the knowledge graph.
The actual implementation of this parsing logic requires careful handling of Python's AST node types. The ast module defines specific node classes for different language constructs. A ClassDef node represents a class definition and contains attributes for the class name, base classes, decorators, and body. The body is a list of nodes representing the class contents, which might include FunctionDef nodes for methods, Assign nodes for class variables, and other constructs. The agent implements visitor methods that handle each node type appropriately.
Here is a concrete code example showing how the parsing agent processes a Python class definition:
import ast
from typing import Dict, List, Set, Any
class PythonCodeAnalyzer:
def __init__(self):
self.entities = {}
self.relationships = []
self.current_class = None
self.current_function = None
def analyze_file(self, filepath: str) -> Dict[str, Any]:
with open(filepath, 'r', encoding='utf-8') as f:
source_code = f.read()
try:
tree = ast.parse(source_code, filename=filepath)
except SyntaxError as e:
return {
'success': False,
'error': f"Syntax error in {filepath}: {str(e)}",
'entities': {},
'relationships': []
}
self.visit_node(tree, filepath)
return {
'success': True,
'entities': self.entities,
'relationships': self.relationships
}
def visit_node(self, node: ast.AST, filepath: str, parent_id: str = None):
if isinstance(node, ast.Module):
module_id = f"module:{filepath}"
self.entities[module_id] = {
'type': 'module',
'name': filepath,
'filepath': filepath,
'docstring': ast.get_docstring(node)
}
for child in node.body:
self.visit_node(child, filepath, module_id)
elif isinstance(node, ast.ClassDef):
class_id = f"class:{filepath}:{node.name}"
self.entities[class_id] = {
'type': 'class',
'name': node.name,
'filepath': filepath,
'line_number': node.lineno,
'docstring': ast.get_docstring(node),
'decorators': [self.get_decorator_name(d) for d in node.decorator_list],
'base_classes': [self.get_base_name(b) for b in node.bases]
}
if parent_id:
self.relationships.append({
'source': parent_id,
'target': class_id,
'type': 'contains'
})
for base in node.bases:
base_name = self.get_base_name(base)
if base_name:
self.relationships.append({
'source': class_id,
'target': f"class:{base_name}",
'type': 'inherits_from'
})
previous_class = self.current_class
self.current_class = class_id
for child in node.body:
self.visit_node(child, filepath, class_id)
self.current_class = previous_class
elif isinstance(node, ast.FunctionDef):
if self.current_class:
function_id = f"method:{filepath}:{self.current_class}:{node.name}"
function_type = 'method'
else:
function_id = f"function:{filepath}:{node.name}"
function_type = 'function'
parameters = []
for arg in node.args.args:
param_info = {'name': arg.arg}
if arg.annotation:
param_info['type'] = ast.unparse(arg.annotation)
parameters.append(param_info)
return_type = None
if node.returns:
return_type = ast.unparse(node.returns)
self.entities[function_id] = {
'type': function_type,
'name': node.name,
'filepath': filepath,
'line_number': node.lineno,
'docstring': ast.get_docstring(node),
'parameters': parameters,
'return_type': return_type,
'decorators': [self.get_decorator_name(d) for d in node.decorator_list],
'is_async': isinstance(node, ast.AsyncFunctionDef)
}
if parent_id:
self.relationships.append({
'source': parent_id,
'target': function_id,
'type': 'contains'
})
previous_function = self.current_function
self.current_function = function_id
for child in ast.walk(node):
if isinstance(child, ast.Call):
called_name = self.get_call_name(child)
if called_name:
self.relationships.append({
'source': function_id,
'target': f"function:{called_name}",
'type': 'calls'
})
self.current_function = previous_function
def get_decorator_name(self, decorator: ast.AST) -> str:
if isinstance(decorator, ast.Name):
return decorator.id
elif isinstance(decorator, ast.Call):
return self.get_call_name(decorator)
else:
return ast.unparse(decorator)
def get_base_name(self, base: ast.AST) -> str:
if isinstance(base, ast.Name):
return base.id
elif isinstance(base, ast.Attribute):
return ast.unparse(base)
else:
return ast.unparse(base)
def get_call_name(self, call: ast.Call) -> str:
if isinstance(call.func, ast.Name):
return call.func.id
elif isinstance(call.func, ast.Attribute):
return ast.unparse(call.func)
else:
return ast.unparse(call.func)
This code analyzer demonstrates the fundamental pattern for extracting structural information from Python code. The analyzer traverses the AST, creating entity records for modules, classes, and functions, and relationship records for containment, inheritance, and function calls. The extracted information provides the raw material for populating the knowledge graph.
The analyzer handles several important details that make it robust for real-world code. It extracts docstrings using the ast.get_docstring function, which correctly retrieves documentation strings according to Python conventions. It captures line numbers so that findings can reference specific locations in source files. It handles both synchronous and asynchronous function definitions. It extracts type annotations when present, providing additional semantic information about parameters and return values. It identifies decorators, which often carry important semantic meaning in Python frameworks.
For languages other than Python, the multi-agent system employs different parsing strategies. The tree-sitter library provides a unified parsing framework that supports dozens of programming languages including JavaScript, TypeScript, Java, C, C++, Go, Rust, and many others. Tree-sitter generates concrete syntax trees that represent the exact syntactic structure of code, including details like whitespace and comments that abstract syntax trees typically omit. While concrete syntax trees contain more detail than necessary for most analysis tasks, tree-sitter provides query capabilities that allow extracting just the relevant structural information.
A tree-sitter based parser for JavaScript would load the JavaScript grammar, parse source files into syntax trees, and execute queries to extract entities and relationships. The query language allows specifying patterns to match. A query might specify that we want to find all function declarations, capturing the function name, parameters, and body. Another query might find all class declarations. Yet another might identify import statements to understand module dependencies.
The implementation of a tree-sitter based analyzer follows a similar pattern to the Python analyzer but uses tree-sitter's API instead of Python's ast module. The analyzer loads the appropriate language grammar, parses each source file, executes queries to find relevant syntax nodes, and extracts entity and relationship information from the matched nodes. The extracted information follows the same schema regardless of source language, ensuring that the knowledge graph maintains a consistent structure even when analyzing polyglot repositories.
Here is an example showing how tree-sitter can be used to parse JavaScript code:
from tree_sitter import Language, Parser
import tree_sitter_javascript
from typing import Dict, List, Any
class JavaScriptCodeAnalyzer:
def __init__(self):
self.entities = {}
self.relationships = []
JS_LANGUAGE = Language(tree_sitter_javascript.language())
self.parser = Parser(JS_LANGUAGE)
def analyze_file(self, filepath: str) -> Dict[str, Any]:
with open(filepath, 'rb') as f:
source_code = f.read()
tree = self.parser.parse(source_code)
root_node = tree.root_node
module_id = f"module:{filepath}"
self.entities[module_id] = {
'type': 'module',
'name': filepath,
'filepath': filepath
}
self.extract_classes(root_node, filepath, module_id)
self.extract_functions(root_node, filepath, module_id)
self.extract_imports(root_node, filepath, module_id)
return {
'success': True,
'entities': self.entities,
'relationships': self.relationships
}
def extract_classes(self, node, filepath: str, parent_id: str):
if node.type == 'class_declaration':
class_name_node = node.child_by_field_name('name')
if class_name_node:
class_name = self.get_node_text(class_name_node)
class_id = f"class:{filepath}:{class_name}"
self.entities[class_id] = {
'type': 'class',
'name': class_name,
'filepath': filepath,
'line_number': node.start_point[0] + 1
}
self.relationships.append({
'source': parent_id,
'target': class_id,
'type': 'contains'
})
heritage_node = node.child_by_field_name('heritage')
if heritage_node:
for child in heritage_node.children:
if child.type == 'extends_clause':
for subchild in child.children:
if subchild.type == 'identifier':
base_class = self.get_node_text(subchild)
self.relationships.append({
'source': class_id,
'target': f"class:{base_class}",
'type': 'inherits_from'
})
body_node = node.child_by_field_name('body')
if body_node:
self.extract_methods(body_node, filepath, class_id)
for child in node.children:
self.extract_classes(child, filepath, parent_id)
def extract_methods(self, class_body_node, filepath: str, class_id: str):
for child in class_body_node.children:
if child.type == 'method_definition':
name_node = child.child_by_field_name('name')
if name_node:
method_name = self.get_node_text(name_node)
method_id = f"method:{filepath}:{class_id}:{method_name}"
parameters = []
params_node = child.child_by_field_name('parameters')
if params_node:
for param_child in params_node.children:
if param_child.type == 'identifier':
parameters.append({
'name': self.get_node_text(param_child)
})
self.entities[method_id] = {
'type': 'method',
'name': method_name,
'filepath': filepath,
'line_number': child.start_point[0] + 1,
'parameters': parameters
}
self.relationships.append({
'source': class_id,
'target': method_id,
'type': 'contains'
})
def extract_functions(self, node, filepath: str, parent_id: str):
if node.type == 'function_declaration':
name_node = node.child_by_field_name('name')
if name_node:
function_name = self.get_node_text(name_node)
function_id = f"function:{filepath}:{function_name}"
parameters = []
params_node = node.child_by_field_name('parameters')
if params_node:
for param_child in params_node.children:
if param_child.type == 'identifier':
parameters.append({
'name': self.get_node_text(param_child)
})
self.entities[function_id] = {
'type': 'function',
'name': function_name,
'filepath': filepath,
'line_number': node.start_point[0] + 1,
'parameters': parameters
}
self.relationships.append({
'source': parent_id,
'target': function_id,
'type': 'contains'
})
for child in node.children:
self.extract_functions(child, filepath, parent_id)
def extract_imports(self, node, filepath: str, module_id: str):
if node.type == 'import_statement':
source_node = node.child_by_field_name('source')
if source_node:
imported_module = self.get_node_text(source_node).strip('"').strip("'")
self.relationships.append({
'source': module_id,
'target': f"module:{imported_module}",
'type': 'imports'
})
for child in node.children:
self.extract_imports(child, filepath, module_id)
def get_node_text(self, node) -> str:
return node.text.decode('utf-8')
This JavaScript analyzer demonstrates how tree-sitter enables parsing non-Python languages using similar patterns. The analyzer navigates the syntax tree by examining node types and extracting text from relevant nodes. It handles JavaScript-specific constructs like class heritage clauses for inheritance and method definitions within class bodies. The extracted information follows the same entity and relationship schema used for Python code, ensuring consistency across languages.
The multi-language parsing capability is essential for analyzing real-world polyglot repositories. A modern web application might include Python backend services, JavaScript frontend code, Java microservices, and SQL database schemas. The multi-agent system must parse all these languages to build a complete understanding of the system. Each language requires its own parser, but all parsers produce output in a common format that feeds into the unified knowledge graph.
Constructing and Querying the Knowledge Graph
The knowledge graph serves as the central repository for structural information about the analyzed system. While various graph database technologies exist, including Neo4j, Amazon Neptune, and ArangoDB, the implementation can also use lighter-weight solutions for smaller deployments. For this discussion, we will use Neo4j as the reference implementation because it provides a mature, feature-rich graph database with excellent query capabilities through the Cypher query language.
The knowledge graph stores entities as nodes and relationships as edges. Each node has a label indicating its type, such as Module, Class, Function, Method, Requirement, or ArchitecturalComponent. Nodes also have properties that store attributes specific to that entity type. A Class node might have properties for name, filepath, line_number, docstring, and programming_language. A Requirement node might have properties for requirement_id, description, priority, and status.
Relationships connect nodes to represent various associations. A Contains relationship connects a module to the classes it contains, or a class to its methods. An InheritsFrom relationship connects a class to its parent class. A Calls relationship connects a function to other functions it invokes. An Implements relationship connects a code component to the requirement it satisfies. A DependsOn relationship connects components that have dependencies.
The code analysis agents populate the knowledge graph by creating nodes and relationships based on the information extracted from parsing. After parsing a Python file and extracting entities and relationships, the agent connects to the graph database and executes queries to create or update nodes and edges. The implementation must handle the fact that multiple files might reference the same entity. If class A in module X inherits from class B in module Y, the parser for module X creates a relationship to class B even though class B is defined elsewhere. When the parser later processes module Y and creates the actual node for class B, the graph database merges these references into a single coherent representation.
Here is code demonstrating how to populate a Neo4j knowledge graph with parsed code information:
from neo4j import GraphDatabase
from typing import Dict, List, Any
import hashlib
class KnowledgeGraphManager:
def __init__(self, uri: str, user: str, password: str):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def close(self):
self.driver.close()
def add_code_entities(self, analysis_result: Dict[str, Any]):
with self.driver.session() as session:
for entity_id, entity_data in analysis_result['entities'].items():
session.execute_write(self._create_entity, entity_id, entity_data)
for relationship in analysis_result['relationships']:
session.execute_write(self._create_relationship, relationship)
@staticmethod
def _create_entity(tx, entity_id: str, entity_data: Dict[str, Any]):
entity_type = entity_data['type']
if entity_type == 'module':
query = """
MERGE (m:Module {id: $entity_id})
SET m.name = $name,
m.filepath = $filepath,
m.docstring = $docstring,
m.last_updated = datetime()
"""
tx.run(query,
entity_id=entity_id,
name=entity_data['name'],
filepath=entity_data['filepath'],
docstring=entity_data.get('docstring'))
elif entity_type == 'class':
query = """
MERGE (c:Class {id: $entity_id})
SET c.name = $name,
c.filepath = $filepath,
c.line_number = $line_number,
c.docstring = $docstring,
c.decorators = $decorators,
c.last_updated = datetime()
"""
tx.run(query,
entity_id=entity_id,
name=entity_data['name'],
filepath=entity_data['filepath'],
line_number=entity_data.get('line_number'),
docstring=entity_data.get('docstring'),
decorators=entity_data.get('decorators', []))
elif entity_type in ['function', 'method']:
label = 'Method' if entity_type == 'method' else 'Function'
query = f"""
MERGE (f:{label} {{id: $entity_id}})
SET f.name = $name,
f.filepath = $filepath,
f.line_number = $line_number,
f.docstring = $docstring,
f.parameters = $parameters,
f.return_type = $return_type,
f.decorators = $decorators,
f.is_async = $is_async,
f.last_updated = datetime()
"""
tx.run(query,
entity_id=entity_id,
name=entity_data['name'],
filepath=entity_data['filepath'],
line_number=entity_data.get('line_number'),
docstring=entity_data.get('docstring'),
parameters=[str(p) for p in entity_data.get('parameters', [])],
return_type=entity_data.get('return_type'),
decorators=entity_data.get('decorators', []),
is_async=entity_data.get('is_async', False))
@staticmethod
def _create_relationship(tx, relationship: Dict[str, Any]):
source_id = relationship['source']
target_id = relationship['target']
rel_type = relationship['type']
query = f"""
MATCH (source {{id: $source_id}})
MATCH (target {{id: $target_id}})
MERGE (source)-[r:{rel_type}]->(target)
SET r.last_updated = datetime()
"""
tx.run(query, source_id=source_id, target_id=target_id)
def find_class_dependencies(self, class_name: str) -> List[Dict[str, Any]]:
with self.driver.session() as session:
result = session.execute_read(self._query_class_dependencies, class_name)
return result
@staticmethod
def _query_class_dependencies(tx, class_name: str):
query = """
MATCH (c:Class {name: $class_name})-[:CONTAINS]->(m:Method)-[:CALLS]->(f)
MATCH (f)<-[:CONTAINS]-(dep:Class)
WHERE dep.name <> $class_name
RETURN DISTINCT dep.name as dependency, dep.filepath as filepath
"""
result = tx.run(query, class_name=class_name)
return [{'name': record['dependency'], 'filepath': record['filepath']}
for record in result]
def find_requirement_implementation(self, requirement_id: str) -> List[Dict[str, Any]]:
with self.driver.session() as session:
result = session.execute_read(self._query_requirement_implementation, requirement_id)
return result
@staticmethod
def _query_requirement_implementation(tx, requirement_id: str):
query = """
MATCH (r:Requirement {id: $requirement_id})-[:IMPLEMENTED_BY]->(component)
RETURN component.name as component_name,
component.type as component_type,
component.filepath as filepath
"""
result = tx.run(query, requirement_id=requirement_id)
return [{'name': record['component_name'],
'type': record['component_type'],
'filepath': record['filepath']}
for record in result]
def find_architectural_hotspots(self, min_dependencies: int = 10) -> List[Dict[str, Any]]:
with self.driver.session() as session:
result = session.execute_read(self._query_hotspots, min_dependencies)
return result
@staticmethod
def _query_hotspots(tx, min_dependencies: int):
query = """
MATCH (c:Class)
OPTIONAL MATCH (c)-[:DEPENDS_ON]->(dep)
WITH c, count(dep) as dependency_count
WHERE dependency_count >= $min_dependencies
RETURN c.name as class_name,
c.filepath as filepath,
dependency_count
ORDER BY dependency_count DESC
"""
result = tx.run(query, min_dependencies=min_dependencies)
return [{'name': record['class_name'],
'filepath': record['filepath'],
'dependency_count': record['dependency_count']}
for record in result]
def find_circular_dependencies(self) -> List[List[str]]:
with self.driver.session() as session:
result = session.execute_read(self._query_circular_dependencies)
return result
@staticmethod
def _query_circular_dependencies(tx):
query = """
MATCH path = (c1:Class)-[:DEPENDS_ON*2..10]->(c1)
WHERE ALL(node in nodes(path) WHERE single(x in nodes(path) WHERE x = node))
RETURN [node in nodes(path) | node.name] as cycle
LIMIT 100
"""
result = tx.run(query)
return [record['cycle'] for record in result]
def calculate_component_centrality(self) -> List[Dict[str, Any]]:
with self.driver.session() as session:
result = session.execute_read(self._query_centrality)
return result
@staticmethod
def _query_centrality(tx):
query = """
CALL gds.pageRank.stream({
nodeProjection: 'Class',
relationshipProjection: 'DEPENDS_ON'
})
YIELD nodeId, score
RETURN gds.util.asNode(nodeId).name as class_name,
gds.util.asNode(nodeId).filepath as filepath,
score
ORDER BY score DESC
LIMIT 50
"""
result = tx.run(query)
return [{'name': record['class_name'],
'filepath': record['filepath'],
'centrality_score': record['score']}
for record in result]
This knowledge graph manager demonstrates the essential operations for populating and querying the graph. The add_code_entities method takes the output from code analyzers and creates nodes and relationships in Neo4j. It uses MERGE operations rather than CREATE to ensure idempotency, meaning that running the same analysis multiple times does not create duplicate nodes. The method sets a last_updated timestamp on each entity, enabling the system to track when information was last refreshed.
The query methods demonstrate different types of analysis enabled by the graph structure. The find_class_dependencies method identifies which other classes a given class depends on by traversing from the class through its methods to the functions they call and then to the classes containing those functions. This traversal would be complex and inefficient with traditional relational database queries but is natural and efficient in a graph database.
The find_requirement_implementation method traces requirements to their implementing components, enabling requirements traceability analysis. The find_architectural_hotspots method identifies components with many dependencies, which often represent architectural risk points. Components with excessive dependencies are difficult to change because modifications ripple through many dependent components. The find_circular_dependencies method detects circular dependency cycles, which represent architectural violations that should be eliminated.
The calculate_component_centrality method uses Neo4j's Graph Data Science library to compute PageRank centrality scores for classes. Centrality measures identify the most important nodes in the graph based on their connectivity patterns. High-centrality components are architecturally significant because many other components depend on them directly or indirectly. Changes to high-centrality components have broad impact across the system.
Implementing Retrieval Augmented Generation with Vector Databases
While the knowledge graph excels at representing structured relationships, much valuable information exists in unstructured or semi-structured text such as documentation, comments, requirements descriptions, and meeting notes. Vector databases enable semantic search over this textual information through embedding-based retrieval. The system generates vector embeddings for text chunks using language models, stores these embeddings in a vector database, and retrieves relevant chunks by finding embeddings similar to a query embedding.
The process begins with chunking documents into manageable pieces. A large architecture document might be split into sections, with each section becoming a separate chunk. Source code files might be chunked at the class or function level. The chunking strategy balances competing concerns. Smaller chunks provide more precise retrieval, returning only the most relevant information. Larger chunks provide more context, ensuring that retrieved information includes surrounding details necessary for understanding. A typical approach uses overlapping chunks where consecutive chunks share some content, ensuring that information spanning chunk boundaries is not lost.
After chunking, the system generates embeddings for each chunk using an embedding model. Modern embedding models like OpenAI's text-embedding-3 or open-source alternatives like sentence-transformers produce high-dimensional vectors that capture semantic meaning. Semantically similar texts produce similar embeddings even when they use different words. A chunk discussing "customer data privacy" and another discussing "personal information protection" would have similar embeddings because they address related concepts.
The embeddings are stored in a vector database along with metadata about each chunk. Popular vector database options include Pinecone, Weaviate, Qdrant, and Chroma. For local deployment, Chroma provides an excellent lightweight option that can run embedded within the application or as a standalone server. The vector database indexes embeddings to enable efficient similarity search across millions of vectors.
When a user or agent poses a question, the system generates an embedding for the question using the same embedding model used for chunks. It then queries the vector database to find the chunks with embeddings most similar to the question embedding. The similarity metric is typically cosine similarity, which measures the angle between vectors in the high-dimensional embedding space. The database returns the top-k most similar chunks, where k might be five or ten depending on how much context the system wants to provide to the language model.
The retrieved chunks provide context for the language model to generate an answer. Rather than relying solely on the model's training data, which might be outdated or might not include information specific to this codebase, the model generates its response based on the retrieved chunks. This retrieval-augmented generation approach grounds the model's responses in factual information from the codebase and documentation, dramatically reducing hallucination.
Here is an implementation of a vector database manager using Chroma:
import chromadb
from chromadb.config import Settings
from typing import List, Dict, Any, Optional
import hashlib
from sentence_transformers import SentenceTransformer
class VectorDatabaseManager:
def __init__(self, persist_directory: str, embedding_model_name: str = "all-MiniLM-L6-v2"):
self.client = chromadb.Client(Settings(
persist_directory=persist_directory,
anonymized_telemetry=False
))
self.embedding_model = SentenceTransformer(embedding_model_name)
self.collections = {}
def get_or_create_collection(self, collection_name: str):
if collection_name not in self.collections:
self.collections[collection_name] = self.client.get_or_create_collection(
name=collection_name,
metadata={"description": f"Collection for {collection_name}"}
)
return self.collections[collection_name]
def add_code_chunks(self, chunks: List[Dict[str, Any]], collection_name: str = "code"):
collection = self.get_or_create_collection(collection_name)
documents = []
metadatas = []
ids = []
for chunk in chunks:
chunk_text = chunk['text']
chunk_metadata = {
'filepath': chunk.get('filepath', ''),
'start_line': chunk.get('start_line', 0),
'end_line': chunk.get('end_line', 0),
'entity_type': chunk.get('entity_type', 'unknown'),
'entity_name': chunk.get('entity_name', ''),
'language': chunk.get('language', 'unknown')
}
chunk_id = self._generate_chunk_id(chunk_text, chunk_metadata)
documents.append(chunk_text)
metadatas.append(chunk_metadata)
ids.append(chunk_id)
if documents:
embeddings = self.embedding_model.encode(documents).tolist()
collection.add(
documents=documents,
metadatas=metadatas,
ids=ids,
embeddings=embeddings
)
def add_documentation_chunks(self, chunks: List[Dict[str, Any]], collection_name: str = "documentation"):
collection = self.get_or_create_collection(collection_name)
documents = []
metadatas = []
ids = []
for chunk in chunks:
chunk_text = chunk['text']
chunk_metadata = {
'document_type': chunk.get('document_type', 'unknown'),
'document_path': chunk.get('document_path', ''),
'section': chunk.get('section', ''),
'subsection': chunk.get('subsection', ''),
'page_number': chunk.get('page_number', 0)
}
chunk_id = self._generate_chunk_id(chunk_text, chunk_metadata)
documents.append(chunk_text)
metadatas.append(chunk_metadata)
ids.append(chunk_id)
if documents:
embeddings = self.embedding_model.encode(documents).tolist()
collection.add(
documents=documents,
metadatas=metadatas,
ids=ids,
embeddings=embeddings
)
def search_code(self, query: str, n_results: int = 5,
filter_metadata: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
collection = self.get_or_create_collection("code")
query_embedding = self.embedding_model.encode([query]).tolist()
results = collection.query(
query_embeddings=query_embedding,
n_results=n_results,
where=filter_metadata
)
return self._format_results(results)
def search_documentation(self, query: str, n_results: int = 5,
filter_metadata: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
collection = self.get_or_create_collection("documentation")
query_embedding = self.embedding_model.encode([query]).tolist()
results = collection.query(
query_embeddings=query_embedding,
n_results=n_results,
where=filter_metadata
)
return self._format_results(results)
def search_all(self, query: str, n_results_per_collection: int = 3) -> Dict[str, List[Dict[str, Any]]]:
all_results = {}
for collection_name in ['code', 'documentation', 'requirements', 'architecture']:
try:
collection = self.get_or_create_collection(collection_name)
query_embedding = self.embedding_model.encode([query]).tolist()
results = collection.query(
query_embeddings=query_embedding,
n_results=n_results_per_collection
)
all_results[collection_name] = self._format_results(results)
except Exception as e:
all_results[collection_name] = []
return all_results
def _format_results(self, results: Dict[str, Any]) -> List[Dict[str, Any]]:
formatted = []
if not results['ids'] or not results['ids'][0]:
return formatted
for i in range(len(results['ids'][0])):
formatted.append({
'id': results['ids'][0][i],
'text': results['documents'][0][i],
'metadata': results['metadatas'][0][i],
'distance': results['distances'][0][i] if 'distances' in results else None
})
return formatted
def _generate_chunk_id(self, text: str, metadata: Dict[str, Any]) -> str:
content = text + str(sorted(metadata.items()))
return hashlib.sha256(content.encode()).hexdigest()
def update_chunk(self, chunk_id: str, new_text: str, new_metadata: Dict[str, Any],
collection_name: str):
collection = self.get_or_create_collection(collection_name)
embedding = self.embedding_model.encode([new_text]).tolist()
collection.update(
ids=[chunk_id],
documents=[new_text],
metadatas=[new_metadata],
embeddings=embedding
)
def delete_chunks_by_filepath(self, filepath: str, collection_name: str):
collection = self.get_or_create_collection(collection_name)
collection.delete(where={"filepath": filepath})
This vector database manager provides the core functionality for storing and retrieving code and documentation chunks. The add_code_chunks method accepts chunks with associated metadata and stores them in the code collection. Each chunk receives a unique identifier generated by hashing its content and metadata, ensuring that identical chunks are not stored multiple times. The method generates embeddings using the sentence-transformers library, which provides high-quality embeddings that can run locally without requiring API calls to external services.
The search methods enable querying the vector database with natural language questions. The search_code method finds code chunks semantically similar to the query. The search_documentation method searches documentation. The search_all method searches across multiple collections simultaneously, providing a comprehensive view of all relevant information regardless of source. The methods support metadata filtering, allowing queries to be restricted to specific files, languages, or document types.
The vector database integrates with the knowledge graph to provide complementary search capabilities. When an agent needs to understand a particular component, it might first query the knowledge graph to identify the component and its immediate relationships, then query the vector database to retrieve detailed documentation and code for that component and related components. The combination of graph-based structural queries and vector-based semantic search provides powerful information retrieval capabilities.
CHAPTER 4: SPECIALIZED AGENT IMPLEMENTATIONS AND REASONING STRATEGIES
Designing the Agent Orchestration Framework
The orchestration framework coordinates the activities of specialized agents, managing their interactions and synthesizing their outputs into coherent analyses. The orchestrator operates as a state machine that tracks the progress of complex analytical tasks, maintaining context about what information has been gathered, what questions remain unanswered, and what agents need to be invoked next. This stateful coordination enables the system to pursue multi-step reasoning chains where the output of one agent informs the inputs to subsequent agents.
The orchestrator maintains a task queue that represents pending analytical work. When a user submits a high-level request such as "analyze the security posture of the authentication subsystem," the orchestrator decomposes this request into concrete subtasks. It might create subtasks for identifying all components involved in authentication, analyzing those components for common security vulnerabilities, checking whether security requirements are documented, verifying that security tests exist, and examining whether security best practices are followed. Each subtask is assigned to an appropriate specialized agent.
The orchestrator also manages dependencies between subtasks. Some subtasks cannot begin until others complete because they require information produced by earlier tasks. The task of analyzing components for vulnerabilities depends on first identifying which components are involved in authentication. The orchestrator constructs a dependency graph of subtasks and schedules them in an order that respects these dependencies. When multiple subtasks have no dependencies between them, the orchestrator can execute them concurrently, improving overall system throughput.
Here is an implementation of the core orchestration framework:
from typing import Dict, List, Any, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
import asyncio
from datetime import datetime
import uuid
class TaskStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
BLOCKED = "blocked"
@dataclass
class Task:
task_id: str
task_type: str
description: str
agent_type: str
parameters: Dict[str, Any]
dependencies: List[str] = field(default_factory=list)
status: TaskStatus = TaskStatus.PENDING
result: Optional[Any] = None
error: Optional[str] = None
created_at: datetime = field(default_factory=datetime.now)
started_at: Optional[datetime] = None
completed_at: Optional[datetime] = None
priority: int = 5
class AgentOrchestrator:
def __init__(self, knowledge_graph_manager, vector_db_manager, llm_manager):
self.knowledge_graph = knowledge_graph_manager
self.vector_db = vector_db_manager
self.llm_manager = llm_manager
self.tasks: Dict[str, Task] = {}
self.agents: Dict[str, Any] = {}
self.task_results: Dict[str, Any] = {}
self.human_questions: List[Dict[str, Any]] = []
def register_agent(self, agent_type: str, agent_instance):
self.agents[agent_type] = agent_instance
async def execute_analysis(self, analysis_request: Dict[str, Any]) -> Dict[str, Any]:
root_task_id = self._create_task_tree(analysis_request)
await self._execute_task_tree(root_task_id)
return self._synthesize_results(root_task_id)
def _create_task_tree(self, analysis_request: Dict[str, Any]) -> str:
request_type = analysis_request.get('type', 'general_analysis')
if request_type == 'security_analysis':
return self._create_security_analysis_tasks(analysis_request)
elif request_type == 'architecture_analysis':
return self._create_architecture_analysis_tasks(analysis_request)
elif request_type == 'requirement_traceability':
return self._create_traceability_tasks(analysis_request)
elif request_type == 'full_system_analysis':
return self._create_full_analysis_tasks(analysis_request)
else:
return self._create_general_analysis_tasks(analysis_request)
def _create_security_analysis_tasks(self, request: Dict[str, Any]) -> str:
subsystem_name = request.get('subsystem', 'authentication')
root_task = Task(
task_id=str(uuid.uuid4()),
task_type='security_analysis',
description=f'Security analysis of {subsystem_name} subsystem',
agent_type='synthesis',
parameters={'subsystem': subsystem_name},
priority=1
)
self.tasks[root_task.task_id] = root_task
identify_task = Task(
task_id=str(uuid.uuid4()),
task_type='identify_components',
description=f'Identify components in {subsystem_name}',
agent_type='architecture',
parameters={'subsystem': subsystem_name},
priority=2
)
self.tasks[identify_task.task_id] = identify_task
vulnerability_task = Task(
task_id=str(uuid.uuid4()),
task_type='analyze_vulnerabilities',
description='Analyze security vulnerabilities',
agent_type='security',
parameters={'subsystem': subsystem_name},
dependencies=[identify_task.task_id],
priority=3
)
self.tasks[vulnerability_task.task_id] = vulnerability_task
requirements_task = Task(
task_id=str(uuid.uuid4()),
task_type='check_security_requirements',
description='Check security requirements',
agent_type='requirements',
parameters={'subsystem': subsystem_name},
priority=3
)
self.tasks[requirements_task.task_id] = requirements_task
testing_task = Task(
task_id=str(uuid.uuid4()),
task_type='analyze_security_tests',
description='Analyze security test coverage',
agent_type='testing',
parameters={'subsystem': subsystem_name},
dependencies=[identify_task.task_id],
priority=3
)
self.tasks[testing_task.task_id] = testing_task
root_task.dependencies = [
vulnerability_task.task_id,
requirements_task.task_id,
testing_task.task_id
]
return root_task.task_id
def _create_architecture_analysis_tasks(self, request: Dict[str, Any]) -> str:
root_task = Task(
task_id=str(uuid.uuid4()),
task_type='architecture_analysis',
description='Comprehensive architecture analysis',
agent_type='synthesis',
parameters=request.get('parameters', {}),
priority=1
)
self.tasks[root_task.task_id] = root_task
structure_task = Task(
task_id=str(uuid.uuid4()),
task_type='analyze_structure',
description='Analyze architectural structure',
agent_type='architecture',
parameters={},
priority=2
)
self.tasks[structure_task.task_id] = structure_task
patterns_task = Task(
task_id=str(uuid.uuid4()),
task_type='identify_patterns',
description='Identify architectural patterns',
agent_type='architecture',
parameters={},
priority=2
)
self.tasks[patterns_task.task_id] = patterns_task
quality_task = Task(
task_id=str(uuid.uuid4()),
task_type='assess_quality_attributes',
description='Assess quality attributes',
agent_type='quality',
parameters={},
dependencies=[structure_task.task_id],
priority=3
)
self.tasks[quality_task.task_id] = quality_task
dependencies_task = Task(
task_id=str(uuid.uuid4()),
task_type='analyze_dependencies',
description='Analyze component dependencies',
agent_type='architecture',
parameters={},
priority=2
)
self.tasks[dependencies_task.task_id] = dependencies_task
hotspots_task = Task(
task_id=str(uuid.uuid4()),
task_type='identify_hotspots',
description='Identify architectural hotspots',
agent_type='repository',
parameters={},
priority=3
)
self.tasks[hotspots_task.task_id] = hotspots_task
root_task.dependencies = [
structure_task.task_id,
patterns_task.task_id,
quality_task.task_id,
dependencies_task.task_id,
hotspots_task.task_id
]
return root_task.task_id
def _create_traceability_tasks(self, request: Dict[str, Any]) -> str:
requirement_id = request.get('requirement_id')
root_task = Task(
task_id=str(uuid.uuid4()),
task_type='requirement_traceability',
description=f'Trace requirement {requirement_id}',
agent_type='synthesis',
parameters={'requirement_id': requirement_id},
priority=1
)
self.tasks[root_task.task_id] = root_task
requirement_task = Task(
task_id=str(uuid.uuid4()),
task_type='analyze_requirement',
description=f'Analyze requirement {requirement_id}',
agent_type='requirements',
parameters={'requirement_id': requirement_id},
priority=2
)
self.tasks[requirement_task.task_id] = requirement_task
design_task = Task(
task_id=str(uuid.uuid4()),
task_type='find_design_elements',
description='Find design elements implementing requirement',
agent_type='architecture',
parameters={'requirement_id': requirement_id},
dependencies=[requirement_task.task_id],
priority=3
)
self.tasks[design_task.task_id] = design_task
code_task = Task(
task_id=str(uuid.uuid4()),
task_type='find_code_implementation',
description='Find code implementing requirement',
agent_type='code',
parameters={'requirement_id': requirement_id},
dependencies=[design_task.task_id],
priority=4
)
self.tasks[code_task.task_id] = code_task
test_task = Task(
task_id=str(uuid.uuid4()),
task_type='find_tests',
description='Find tests validating requirement',
agent_type='testing',
parameters={'requirement_id': requirement_id},
dependencies=[code_task.task_id],
priority=5
)
self.tasks[test_task.task_id] = test_task
root_task.dependencies = [
requirement_task.task_id,
design_task.task_id,
code_task.task_id,
test_task.task_id
]
return root_task.task_id
def _create_full_analysis_tasks(self, request: Dict[str, Any]) -> str:
root_task = Task(
task_id=str(uuid.uuid4()),
task_type='full_system_analysis',
description='Complete system analysis with SWOT report',
agent_type='report',
parameters=request.get('parameters', {}),
priority=1
)
self.tasks[root_task.task_id] = root_task
business_task = Task(
task_id=str(uuid.uuid4()),
task_type='analyze_business_context',
description='Analyze business goals and strategy',
agent_type='business',
parameters={},
priority=2
)
self.tasks[business_task.task_id] = business_task
requirements_task = Task(
task_id=str(uuid.uuid4()),
task_type='analyze_requirements',
description='Analyze requirements completeness and quality',
agent_type='requirements',
parameters={},
priority=2
)
self.tasks[requirements_task.task_id] = requirements_task
architecture_subtask_id = self._create_architecture_analysis_tasks({
'type': 'architecture_analysis'
})
domain_task = Task(
task_id=str(uuid.uuid4()),
task_type='analyze_domain_model',
description='Analyze problem and solution domain models',
agent_type='domain',
parameters={},
priority=2
)
self.tasks[domain_task.task_id] = domain_task
code_quality_task = Task(
task_id=str(uuid.uuid4()),
task_type='analyze_code_quality',
description='Analyze code quality metrics',
agent_type='code',
parameters={},
priority=3
)
self.tasks[code_quality_task.task_id] = code_quality_task
testing_task = Task(
task_id=str(uuid.uuid4()),
task_type='analyze_testing',
description='Analyze test coverage and quality',
agent_type='testing',
parameters={},
priority=3
)
self.tasks[testing_task.task_id] = testing_task
operations_task = Task(
task_id=str(uuid.uuid4()),
task_type='analyze_operations',
description='Analyze operational aspects',
agent_type='operations',
parameters={},
priority=3
)
self.tasks[operations_task.task_id] = operations_task
technical_debt_task = Task(
task_id=str(uuid.uuid4()),
task_type='analyze_technical_debt',
description='Identify and categorize technical debt',
agent_type='code',
parameters={},
priority=3
)
self.tasks[technical_debt_task.task_id] = technical_debt_task
root_task.dependencies = [
business_task.task_id,
requirements_task.task_id,
architecture_subtask_id,
domain_task.task_id,
code_quality_task.task_id,
testing_task.task_id,
operations_task.task_id,
technical_debt_task.task_id
]
return root_task.task_id
def _create_general_analysis_tasks(self, request: Dict[str, Any]) -> str:
root_task = Task(
task_id=str(uuid.uuid4()),
task_type='general_analysis',
description=request.get('description', 'General analysis'),
agent_type='synthesis',
parameters=request.get('parameters', {}),
priority=1
)
self.tasks[root_task.task_id] = root_task
return root_task.task_id
async def _execute_task_tree(self, root_task_id: str):
ready_tasks = self._get_ready_tasks()
while ready_tasks or self._has_in_progress_tasks():
if ready_tasks:
task_batch = ready_tasks[:5]
await asyncio.gather(*[self._execute_task(task.task_id) for task in task_batch])
else:
await asyncio.sleep(0.1)
ready_tasks = self._get_ready_tasks()
if not self._all_tasks_completed():
failed_tasks = [t for t in self.tasks.values() if t.status == TaskStatus.FAILED]
blocked_tasks = [t for t in self.tasks.values() if t.status == TaskStatus.BLOCKED]
raise Exception(f"Task execution incomplete. Failed: {len(failed_tasks)}, Blocked: {len(blocked_tasks)}")
def _get_ready_tasks(self) -> List[Task]:
ready = []
for task in self.tasks.values():
if task.status == TaskStatus.PENDING:
if all(self.tasks[dep_id].status == TaskStatus.COMPLETED
for dep_id in task.dependencies):
ready.append(task)
ready.sort(key=lambda t: t.priority, reverse=True)
return ready
def _has_in_progress_tasks(self) -> bool:
return any(t.status == TaskStatus.IN_PROGRESS for t in self.tasks.values())
def _all_tasks_completed(self) -> bool:
return all(t.status in [TaskStatus.COMPLETED, TaskStatus.FAILED]
for t in self.tasks.values())
async def _execute_task(self, task_id: str):
task = self.tasks[task_id]
task.status = TaskStatus.IN_PROGRESS
task.started_at = datetime.now()
try:
agent = self.agents.get(task.agent_type)
if not agent:
raise Exception(f"No agent registered for type {task.agent_type}")
dependency_results = {
dep_id: self.tasks[dep_id].result
for dep_id in task.dependencies
}
task.parameters['dependency_results'] = dependency_results
task.parameters['knowledge_graph'] = self.knowledge_graph
task.parameters['vector_db'] = self.vector_db
task.parameters['llm_manager'] = self.llm_manager
result = await agent.execute(task)
task.result = result
task.status = TaskStatus.COMPLETED
task.completed_at = datetime.now()
self.task_results[task_id] = result
except Exception as e:
task.status = TaskStatus.FAILED
task.error = str(e)
task.completed_at = datetime.now()
def _synthesize_results(self, root_task_id: str) -> Dict[str, Any]:
root_task = self.tasks[root_task_id]
if root_task.status != TaskStatus.COMPLETED:
return {
'success': False,
'error': f"Root task failed: {root_task.error}",
'partial_results': self.task_results
}
return {
'success': True,
'result': root_task.result,
'task_tree': self._build_task_tree_summary(root_task_id),
'human_questions': self.human_questions
}
def _build_task_tree_summary(self, task_id: str) -> Dict[str, Any]:
task = self.tasks[task_id]
return {
'task_id': task.task_id,
'task_type': task.task_type,
'description': task.description,
'status': task.status.value,
'duration_seconds': (
(task.completed_at - task.started_at).total_seconds()
if task.started_at and task.completed_at else None
),
'dependencies': [
self._build_task_tree_summary(dep_id)
for dep_id in task.dependencies
]
}
def add_human_question(self, question: str, context: Dict[str, Any], priority: int = 5):
self.human_questions.append({
'question': question,
'context': context,
'priority': priority,
'timestamp': datetime.now()
})
This orchestrator implementation demonstrates the core patterns for managing complex analytical workflows. The execute_analysis method serves as the entry point, accepting high-level analysis requests and coordinating their execution. The orchestrator decomposes requests into task trees where each task represents a discrete analytical step performed by a specialized agent. The task tree structure explicitly represents dependencies, ensuring that tasks execute in the correct order and that each task receives the results from its prerequisite tasks.
The orchestrator supports concurrent execution of independent tasks through asynchronous programming with asyncio. When multiple tasks have no dependencies between them, they execute in parallel, significantly reducing the total time required for complex analyses. The implementation limits concurrent execution to batches of five tasks to prevent overwhelming system resources, but this limit can be adjusted based on available computational capacity.
The task creation methods demonstrate how different types of analyses decompose into specific task sequences. A security analysis identifies components, analyzes vulnerabilities, checks requirements, and examines test coverage. An architecture analysis examines structure, identifies patterns, assesses quality attributes, analyzes dependencies, and identifies hotspots. A full system analysis orchestrates multiple sub-analyses covering business context, requirements, architecture, domain modeling, code quality, testing, and operations. This decomposition strategy allows the orchestrator to reuse specialized agents across different types of analyses.
Implementing Specialized Analysis Agents
Each specialized agent focuses on a particular analytical domain, implementing deep expertise in that area. The architecture analysis agent understands architectural patterns, quality attributes, and structural principles. It examines the system's organization, identifying how components are grouped into modules and subsystems, how these elements communicate, and whether the organization follows recognized architectural patterns.
The architecture agent begins by querying the knowledge graph to understand the system's high-level structure. It identifies major components and their dependencies, constructing a mental model of the system's architecture. It looks for patterns that indicate specific architectural styles. If it finds components organized into presentation, business logic, and data access layers with dependencies flowing in one direction, it recognizes a layered architecture. If it finds independent services communicating through message passing, it recognizes a microservices architecture. If it finds a central data store with multiple components reading and writing shared data, it recognizes a shared database pattern.
The agent evaluates whether the actual implementation aligns with documented architectural intentions. If architecture documentation describes a microservices architecture but the code analysis reveals tight coupling between services with synchronous calls and shared databases, the agent identifies this misalignment as an architectural weakness. It generates findings that describe the discrepancy, explain why it matters, and suggest remediation approaches.
Here is an implementation of an architecture analysis agent:
from typing import Dict, List, Any, Optional
import asyncio
class ArchitectureAnalysisAgent:
def __init__(self):
self.patterns = {
'layered': self._detect_layered_architecture,
'microservices': self._detect_microservices,
'event_driven': self._detect_event_driven,
'hexagonal': self._detect_hexagonal,
'monolithic': self._detect_monolithic
}
async def execute(self, task: Any) -> Dict[str, Any]:
task_type = task.task_type
parameters = task.parameters
kg = parameters['knowledge_graph']
vector_db = parameters['vector_db']
llm = parameters['llm_manager']
if task_type == 'analyze_structure':
return await self._analyze_structure(kg, vector_db, llm)
elif task_type == 'identify_patterns':
return await self._identify_patterns(kg, vector_db, llm)
elif task_type == 'analyze_dependencies':
return await self._analyze_dependencies(kg, llm)
elif task_type == 'identify_components':
subsystem = parameters.get('subsystem')
return await self._identify_components(subsystem, kg, vector_db, llm)
elif task_type == 'find_design_elements':
requirement_id = parameters.get('requirement_id')
return await self._find_design_elements(requirement_id, kg, vector_db, llm)
else:
return {'error': f'Unknown task type: {task_type}'}
async def _analyze_structure(self, kg, vector_db, llm) -> Dict[str, Any]:
modules = await self._get_all_modules(kg)
module_analysis = []
for module in modules:
classes = await self._get_module_classes(kg, module['name'])
dependencies = await self._get_module_dependencies(kg, module['name'])
module_analysis.append({
'module': module['name'],
'filepath': module['filepath'],
'class_count': len(classes),
'dependency_count': len(dependencies),
'classes': [c['name'] for c in classes],
'dependencies': [d['name'] for d in dependencies]
})
subsystems = await self._identify_subsystems(module_analysis, llm)
layering_analysis = await self._analyze_layering(kg, subsystems, llm)
return {
'module_count': len(modules),
'modules': module_analysis,
'subsystems': subsystems,
'layering': layering_analysis
}
async def _identify_patterns(self, kg, vector_db, llm) -> Dict[str, Any]:
detected_patterns = {}
for pattern_name, detector_func in self.patterns.items():
detection_result = await detector_func(kg, vector_db, llm)
if detection_result['detected']:
detected_patterns[pattern_name] = detection_result
architecture_docs = vector_db.search_documentation(
"architecture pattern style design",
n_results=5
)
documented_patterns = await self._extract_documented_patterns(
architecture_docs, llm
)
alignment_analysis = await self._analyze_pattern_alignment(
detected_patterns, documented_patterns, llm
)
return {
'detected_patterns': detected_patterns,
'documented_patterns': documented_patterns,
'alignment': alignment_analysis
}
async def _detect_layered_architecture(self, kg, vector_db, llm) -> Dict[str, Any]:
modules = await self._get_all_modules(kg)
layer_indicators = {
'presentation': ['ui', 'view', 'controller', 'api', 'rest', 'graphql', 'web'],
'business': ['service', 'business', 'logic', 'domain', 'use_case', 'application'],
'data': ['repository', 'dao', 'database', 'persistence', 'storage', 'data_access']
}
layer_modules = {'presentation': [], 'business': [], 'data': []}
for module in modules:
module_name_lower = module['name'].lower()
module_path_lower = module['filepath'].lower()
for layer, indicators in layer_indicators.items():
if any(ind in module_name_lower or ind in module_path_lower
for ind in indicators):
layer_modules[layer].append(module['name'])
break
if all(len(modules) > 0 for modules in layer_modules.values()):
violations = await self._check_layer_violations(kg, layer_modules)
return {
'detected': True,
'confidence': 0.8 if len(violations) == 0 else 0.5,
'layers': layer_modules,
'violations': violations,
'description': 'System appears to follow a layered architecture pattern'
}
return {
'detected': False,
'confidence': 0.0,
'description': 'No clear layered architecture detected'
}
async def _detect_microservices(self, kg, vector_db, llm) -> Dict[str, Any]:
services = await self._identify_services(kg, vector_db)
if len(services) < 2:
return {
'detected': False,
'confidence': 0.0,
'description': 'Insufficient number of independent services'
}
independence_score = await self._assess_service_independence(kg, services)
communication_analysis = await self._analyze_service_communication(
kg, vector_db, services
)
data_analysis = await self._analyze_service_data_isolation(kg, services)
confidence = (independence_score +
communication_analysis['async_ratio'] +
data_analysis['isolation_score']) / 3
return {
'detected': confidence > 0.5,
'confidence': confidence,
'service_count': len(services),
'services': services,
'independence_score': independence_score,
'communication': communication_analysis,
'data_isolation': data_analysis,
'description': f'System contains {len(services)} services with {confidence:.2f} microservices characteristics'
}
async def _detect_event_driven(self, kg, vector_db, llm) -> Dict[str, Any]:
event_indicators = vector_db.search_code(
"event publish subscribe emit listener handler message queue",
n_results=20
)
if len(event_indicators) < 5:
return {
'detected': False,
'confidence': 0.0,
'description': 'Insufficient event-driven patterns found'
}
event_classes = await self._identify_event_classes(kg, event_indicators)
publishers = await self._identify_publishers(kg, event_classes)
subscribers = await self._identify_subscribers(kg, event_classes)
if len(event_classes) > 0 and len(publishers) > 0 and len(subscribers) > 0:
return {
'detected': True,
'confidence': 0.7,
'event_count': len(event_classes),
'publisher_count': len(publishers),
'subscriber_count': len(subscribers),
'events': event_classes,
'description': 'System uses event-driven architecture patterns'
}
return {
'detected': False,
'confidence': 0.2,
'description': 'Limited event-driven patterns detected'
}
async def _detect_hexagonal(self, kg, vector_db, llm) -> Dict[str, Any]:
port_indicators = vector_db.search_code(
"port adapter interface repository gateway",
n_results=15
)
ports = await self._identify_ports(kg, port_indicators)
adapters = await self._identify_adapters(kg, port_indicators)
domain_core = await self._identify_domain_core(kg, vector_db)
if len(ports) > 0 and len(adapters) > 0 and domain_core:
isolation_score = await self._assess_domain_isolation(
kg, domain_core, adapters
)
return {
'detected': True,
'confidence': isolation_score,
'port_count': len(ports),
'adapter_count': len(adapters),
'ports': ports,
'adapters': adapters,
'domain_core': domain_core,
'description': 'System shows hexagonal architecture characteristics'
}
return {
'detected': False,
'confidence': 0.0,
'description': 'No hexagonal architecture pattern detected'
}
async def _detect_monolithic(self, kg, vector_db, llm) -> Dict[str, Any]:
modules = await self._get_all_modules(kg)
coupling_analysis = await self._analyze_coupling(kg, modules)
if coupling_analysis['average_coupling'] > 0.6:
return {
'detected': True,
'confidence': 0.8,
'module_count': len(modules),
'average_coupling': coupling_analysis['average_coupling'],
'description': 'System exhibits monolithic architecture characteristics with high coupling'
}
return {
'detected': False,
'confidence': 0.3,
'description': 'System does not appear to be monolithic'
}
async def _analyze_dependencies(self, kg, llm) -> Dict[str, Any]:
circular_deps = kg.find_circular_dependencies()
hotspots = kg.find_architectural_hotspots(min_dependencies=10)
all_modules = await self._get_all_modules(kg)
coupling_metrics = await self._calculate_coupling_metrics(kg, all_modules)
dependency_violations = await self._identify_dependency_violations(
kg, llm
)
return {
'circular_dependency_count': len(circular_deps),
'circular_dependencies': circular_deps[:10],
'hotspot_count': len(hotspots),
'hotspots': hotspots[:20],
'coupling_metrics': coupling_metrics,
'violations': dependency_violations
}
async def _identify_components(self, subsystem: str, kg, vector_db, llm) -> Dict[str, Any]:
subsystem_docs = vector_db.search_documentation(
f"{subsystem} subsystem components modules",
n_results=5
)
subsystem_code = vector_db.search_code(
f"{subsystem}",
n_results=20
)
component_names = await self._extract_component_names(
subsystem_docs, subsystem_code, llm
)
components = []
for name in component_names:
component_details = await self._get_component_details(kg, name)
if component_details:
components.append(component_details)
return {
'subsystem': subsystem,
'component_count': len(components),
'components': components
}
async def _find_design_elements(self, requirement_id: str, kg, vector_db, llm) -> Dict[str, Any]:
requirement_impl = kg.find_requirement_implementation(requirement_id)
if not requirement_impl:
requirement_docs = vector_db.search_documentation(
f"requirement {requirement_id}",
n_results=3
)
if requirement_docs:
potential_components = await self._infer_components_from_docs(
requirement_docs, kg, llm
)
return {
'requirement_id': requirement_id,
'design_elements': potential_components,
'source': 'inferred_from_documentation'
}
return {
'requirement_id': requirement_id,
'design_elements': requirement_impl,
'source': 'knowledge_graph'
}
async def _get_all_modules(self, kg) -> List[Dict[str, Any]]:
return []
async def _get_module_classes(self, kg, module_name: str) -> List[Dict[str, Any]]:
return []
async def _get_module_dependencies(self, kg, module_name: str) -> List[Dict[str, Any]]:
return []
async def _identify_subsystems(self, module_analysis: List[Dict], llm) -> List[Dict[str, Any]]:
return []
async def _analyze_layering(self, kg, subsystems: List[Dict], llm) -> Dict[str, Any]:
return {}
async def _extract_documented_patterns(self, docs: List[Dict], llm) -> List[str]:
return []
async def _analyze_pattern_alignment(self, detected: Dict, documented: List[str], llm) -> Dict[str, Any]:
return {}
async def _check_layer_violations(self, kg, layer_modules: Dict) -> List[Dict[str, Any]]:
return []
async def _identify_services(self, kg, vector_db) -> List[Dict[str, Any]]:
return []
async def _assess_service_independence(self, kg, services: List[Dict]) -> float:
return 0.0
async def _analyze_service_communication(self, kg, vector_db, services: List[Dict]) -> Dict[str, Any]:
return {'async_ratio': 0.0}
async def _analyze_service_data_isolation(self, kg, services: List[Dict]) -> Dict[str, Any]:
return {'isolation_score': 0.0}
async def _identify_event_classes(self, kg, indicators: List[Dict]) -> List[str]:
return []
async def _identify_publishers(self, kg, event_classes: List[str]) -> List[str]:
return []
async def _identify_subscribers(self, kg, event_classes: List[str]) -> List[str]:
return []
async def _identify_ports(self, kg, indicators: List[Dict]) -> List[str]:
return []
async def _identify_adapters(self, kg, indicators: List[Dict]) -> List[str]:
return []
async def _identify_domain_core(self, kg, vector_db) -> Optional[Dict[str, Any]]:
return None
async def _assess_domain_isolation(self, kg, domain_core: Dict, adapters: List[str]) -> float:
return 0.0
async def _analyze_coupling(self, kg, modules: List[Dict]) -> Dict[str, Any]:
return {'average_coupling': 0.0}
async def _calculate_coupling_metrics(self, kg, modules: List[Dict]) -> Dict[str, Any]:
return {}
async def _identify_dependency_violations(self, kg, llm) -> List[Dict[str, Any]]:
return []
async def _extract_component_names(self, docs: List[Dict], code: List[Dict], llm) -> List[str]:
return []
async def _get_component_details(self, kg, name: str) -> Optional[Dict[str, Any]]:
return None
async def _infer_components_from_docs(self, docs: List[Dict], kg, llm) -> List[Dict[str, Any]]:
return []
This architecture agent demonstrates the sophisticated analysis patterns required for understanding system architecture. The agent implements multiple pattern detection algorithms, each looking for specific architectural characteristics. The layered architecture detector examines module names and paths for layer indicators, then checks whether dependencies respect layer boundaries. The microservices detector identifies independent services and assesses their isolation and communication patterns. The event-driven detector looks for event publishing and subscription patterns.
The agent combines information from multiple sources to form conclusions. It queries the knowledge graph for structural information, searches the vector database for relevant documentation and code, and uses language models to interpret ambiguous information. When the agent detects discrepancies between documented architectural intentions and actual implementation, it generates findings that highlight these issues.
Implementing Domain Analysis with Named Entity Recognition
The domain analysis agent focuses on identifying and modeling the business concepts that the software represents. This agent employs named entity recognition techniques to extract domain terminology from various sources including requirements documents, architecture documentation, code comments, and the code itself. Understanding the domain model is crucial for assessing whether the software accurately represents the business reality it is meant to support.
The agent begins by processing requirements documents to identify entities, which are typically nouns that represent important business concepts. In an e-commerce system, entities might include Customer, Product, Order, Payment, and Shipment. The agent also identifies relationships between entities, which are typically expressed through verbs. A Customer places an Order. An Order contains Products. A Payment processes an Order.
After extracting entities and relationships from requirements, the agent examines the code to identify how these domain concepts are implemented. It looks for classes, database tables, and API endpoints whose names correspond to domain entities. When it finds a Customer class, a customers database table, and a customer API endpoint, it recognizes that the Customer domain concept is represented across multiple technical layers.
The agent also identifies cases where domain concepts are poorly represented in the code. If requirements discuss a complex business concept like "subscription renewal with promotional pricing," but the code contains no clear representation of this concept, the agent flags this as a potential domain modeling weakness. The business logic for subscription renewal might be scattered across multiple classes with no clear organizing structure, making it difficult to understand and maintain.
Here is an implementation sketch of a domain analysis agent:
import spacy
from typing import Dict, List, Any, Set, Tuple
import re
class DomainAnalysisAgent:
def __init__(self):
self.nlp = spacy.load("en_core_web_sm")
self.domain_entities = {}
self.domain_relationships = []
self.problem_domain = {'entities': {}, 'relationships': []}
self.solution_domain = {'entities': {}, 'relationships': []}
async def execute(self, task: Any) -> Dict[str, Any]:
task_type = task.task_type
parameters = task.parameters
kg = parameters['knowledge_graph']
vector_db = parameters['vector_db']
llm = parameters['llm_manager']
if task_type == 'analyze_domain_model':
return await self._analyze_domain_model(kg, vector_db, llm)
else:
return {'error': f'Unknown task type: {task_type}'}
async def _analyze_domain_model(self, kg, vector_db, llm) -> Dict[str, Any]:
requirements_docs = vector_db.search_documentation(
"requirements functional business domain",
n_results=20,
filter_metadata={'document_type': 'requirements'}
)
architecture_docs = vector_db.search_documentation(
"domain model entity aggregate",
n_results=10
)
await self._extract_problem_domain(requirements_docs, architecture_docs, llm)
await self._extract_solution_domain(kg, vector_db)
mapping_analysis = await self._analyze_domain_mapping(kg, llm)
consistency_analysis = await self._analyze_terminology_consistency(
vector_db, llm
)
completeness_analysis = await self._analyze_domain_completeness(llm)
return {
'problem_domain': self.problem_domain,
'solution_domain': self.solution_domain,
'mapping': mapping_analysis,
'terminology_consistency': consistency_analysis,
'completeness': completeness_analysis
}
async def _extract_problem_domain(self, requirements_docs: List[Dict],
architecture_docs: List[Dict], llm) -> None:
all_docs = requirements_docs + architecture_docs
for doc in all_docs:
doc_entities = self._extract_entities_from_text(doc['text'])
doc_relationships = self._extract_relationships_from_text(doc['text'])
for entity in doc_entities:
if entity['text'] not in self.problem_domain['entities']:
self.problem_domain['entities'][entity['text']] = {
'name': entity['text'],
'type': entity['label'],
'occurrences': [],
'attributes': set(),
'descriptions': []
}
self.problem_domain['entities'][entity['text']]['occurrences'].append({
'document': doc['metadata'].get('document_path', 'unknown'),
'context': entity['context']
})
for rel in doc_relationships:
self.problem_domain['relationships'].append(rel)
await self._enrich_entities_with_llm(self.problem_domain['entities'], llm)
def _extract_entities_from_text(self, text: str) -> List[Dict[str, Any]]:
doc = self.nlp(text)
entities = []
for ent in doc.ents:
if ent.label_ in ['PERSON', 'ORG', 'PRODUCT', 'EVENT']:
continue
context_start = max(0, ent.start_char - 50)
context_end = min(len(text), ent.end_char + 50)
context = text[context_start:context_end]
entities.append({
'text': ent.text,
'label': ent.label_,
'context': context
})
noun_chunks = [chunk for chunk in doc.noun_chunks
if chunk.root.pos_ == 'NOUN' and len(chunk.text.split()) <= 3]
for chunk in noun_chunks:
if chunk.text not in [e['text'] for e in entities]:
context_start = max(0, chunk.start_char - 50)
context_end = min(len(text), chunk.end_char + 50)
context = text[context_start:context_end]
entities.append({
'text': chunk.text,
'label': 'CONCEPT',
'context': context
})
return entities
def _extract_relationships_from_text(self, text: str) -> List[Dict[str, Any]]:
doc = self.nlp(text)
relationships = []
for sent in doc.sents:
subject = None
verb = None
obj = None
for token in sent:
if token.dep_ in ['nsubj', 'nsubjpass'] and not subject:
subject = token.text
elif token.pos_ == 'VERB' and not verb:
verb = token.lemma_
elif token.dep_ in ['dobj', 'pobj'] and not obj:
obj = token.text
if subject and verb and obj:
relationships.append({
'subject': subject,
'predicate': verb,
'object': obj,
'sentence': sent.text
})
return relationships
async def _extract_solution_domain(self, kg, vector_db) -> None:
code_entities = vector_db.search_code(
"class interface service repository entity model",
n_results=50
)
for code_chunk in code_entities:
entity_name = code_chunk['metadata'].get('entity_name', '')
entity_type = code_chunk['metadata'].get('entity_type', '')
if entity_name and entity_type in ['class', 'interface']:
if entity_name not in self.solution_domain['entities']:
self.solution_domain['entities'][entity_name] = {
'name': entity_name,
'type': entity_type,
'filepath': code_chunk['metadata'].get('filepath', ''),
'methods': [],
'attributes': [],
'dependencies': []
}
async def _analyze_domain_mapping(self, kg, llm) -> Dict[str, Any]:
mappings = []
unmapped_problem_entities = []
unmapped_solution_entities = []
for problem_entity_name, problem_entity in self.problem_domain['entities'].items():
matching_solution = self._find_matching_solution_entity(problem_entity_name)
if matching_solution:
mappings.append({
'problem_entity': problem_entity_name,
'solution_entity': matching_solution,
'mapping_quality': self._assess_mapping_quality(
problem_entity, self.solution_domain['entities'][matching_solution]
)
})
else:
unmapped_problem_entities.append(problem_entity_name)
for solution_entity_name in self.solution_domain['entities'].keys():
if not any(m['solution_entity'] == solution_entity_name for m in mappings):
unmapped_solution_entities.append(solution_entity_name)
return {
'mapped_count': len(mappings),
'mappings': mappings,
'unmapped_problem_entities': unmapped_problem_entities,
'unmapped_solution_entities': unmapped_solution_entities,
'mapping_coverage': len(mappings) / max(len(self.problem_domain['entities']), 1)
}
def _find_matching_solution_entity(self, problem_entity_name: str) -> Optional[str]:
problem_name_lower = problem_entity_name.lower()
problem_name_normalized = re.sub(r'[^a-z0-9]', '', problem_name_lower)
for solution_name in self.solution_domain['entities'].keys():
solution_name_lower = solution_name.lower()
solution_name_normalized = re.sub(r'[^a-z0-9]', '', solution_name_lower)
if problem_name_normalized == solution_name_normalized:
return solution_name
if problem_name_normalized in solution_name_normalized or solution_name_normalized in problem_name_normalized:
return solution_name
return None
def _assess_mapping_quality(self, problem_entity: Dict, solution_entity: Dict) -> str:
return 'good'
async def _analyze_terminology_consistency(self, vector_db, llm) -> Dict[str, Any]:
return {}
async def _analyze_domain_completeness(self, llm) -> Dict[str, Any]:
return {}
async def _enrich_entities_with_llm(self, entities: Dict[str, Any], llm) -> None:
pass
This domain analysis agent demonstrates how natural language processing techniques extract domain knowledge from textual sources. The agent uses spaCy for named entity recognition and dependency parsing. It identifies entities by finding noun phrases and named entities in requirements documents. It extracts relationships by analyzing sentence structure to find subject-verb-object patterns.
The agent builds separate models for the problem domain and solution domain, then analyzes the mapping between them. Strong mappings where problem domain concepts clearly correspond to solution domain structures indicate good domain alignment. Weak mappings or unmapped entities indicate potential issues where the software does not accurately model the business domain.
CHAPTER 5: LLM INTEGRATION, REASONING PATTERNS, AND MULTI-PLATFORM SUPPORT
Designing a Flexible LLM Manager for Multiple Providers
The multi-agent system must support various language models from different providers, both cloud-based and local. Different models have different strengths, and the system should leverage these strengths by routing tasks to the most appropriate model. Cloud-based models like GPT-5.1, Gemini 3 Pro, and Claude 4.5 Sonnet offer cutting-edge capabilities and large context windows but require API calls and incur usage costs. Local models running on consumer hardware provide privacy, cost predictability, and independence from external services but typically offer less sophisticated reasoning capabilities.
The LLM manager abstracts the differences between providers, presenting a unified interface to agents while handling provider-specific details internally. An agent requests text generation by providing a prompt and parameters like temperature and maximum tokens. The LLM manager routes this request to an appropriate model based on configuration, task requirements, and current system load. It handles authentication, rate limiting, error recovery, and response parsing, shielding agents from these complexities.
The manager supports both synchronous and asynchronous invocation patterns. For tasks where the agent must wait for the LLM response before proceeding, synchronous calls provide a straightforward programming model. For tasks where multiple LLM calls can execute concurrently, asynchronous calls improve throughput by allowing the system to process multiple requests simultaneously without blocking.
The implementation must handle the reality that different providers use different API structures and authentication mechanisms. OpenAI models use API keys passed in headers. Anthropic models use a similar pattern but with different endpoint URLs and request formats. Google's Gemini models use yet another API structure. Local models running through frameworks like llama.cpp or vLLM expose HTTP endpoints with their own request formats. The LLM manager encapsulates these differences, translating the unified internal request format into provider-specific API calls.
Here is an implementation of a comprehensive LLM manager:
import asyncio
import aiohttp
from typing import Dict, List, Any, Optional, Union
from dataclasses import dataclass
from enum import Enum
import json
import time
from abc import ABC, abstractmethod
class ModelProvider(Enum):
OPENAI = "openai"
ANTHROPIC = "anthropic"
GOOGLE = "google"
LOCAL_LLAMA = "local_llama"
LOCAL_VLLM = "local_vllm"
OLLAMA = "ollama"
@dataclass
class LLMRequest:
prompt: str
system_prompt: Optional[str] = None
temperature: float = 0.7
max_tokens: int = 2000
top_p: float = 0.95
stop_sequences: Optional[List[str]] = None
model_preference: Optional[str] = None
task_type: Optional[str] = None
@dataclass
class LLMResponse:
text: str
model_used: str
provider: str
tokens_used: int
latency_ms: float
finish_reason: str
metadata: Dict[str, Any]
class LLMProvider(ABC):
@abstractmethod
async def generate(self, request: LLMRequest) -> LLMResponse:
pass
@abstractmethod
def is_available(self) -> bool:
pass
@abstractmethod
def get_cost_per_token(self) -> float:
pass
class OpenAIProvider(LLMProvider):
def __init__(self, api_key: str, model: str = "gpt-4", base_url: str = "https://api.openai.com/v1"):
self.api_key = api_key
self.model = model
self.base_url = base_url
self.session = None
async def generate(self, request: LLMRequest) -> LLMResponse:
if not self.session:
self.session = aiohttp.ClientSession()
messages = []
if request.system_prompt:
messages.append({"role": "system", "content": request.system_prompt})
messages.append({"role": "user", "content": request.prompt})
payload = {
"model": self.model,
"messages": messages,
"temperature": request.temperature,
"max_tokens": request.max_tokens,
"top_p": request.top_p
}
if request.stop_sequences:
payload["stop"] = request.stop_sequences
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
start_time = time.time()
async with self.session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers
) as response:
if response.status != 200:
error_text = await response.text()
raise Exception(f"OpenAI API error: {response.status} - {error_text}")
result = await response.json()
latency_ms = (time.time() - start_time) * 1000
choice = result["choices"][0]
usage = result.get("usage", {})
return LLMResponse(
text=choice["message"]["content"],
model_used=self.model,
provider="openai",
tokens_used=usage.get("total_tokens", 0),
latency_ms=latency_ms,
finish_reason=choice.get("finish_reason", "unknown"),
metadata={"usage": usage}
)
def is_available(self) -> bool:
return bool(self.api_key)
def get_cost_per_token(self) -> float:
cost_map = {
"gpt-4": 0.00003,
"gpt-4-turbo": 0.00001,
"gpt-3.5-turbo": 0.000002
}
return cost_map.get(self.model, 0.00001)
async def close(self):
if self.session:
await self.session.close()
class AnthropicProvider(LLMProvider):
def __init__(self, api_key: str, model: str = "claude-3-5-sonnet-20241022"):
self.api_key = api_key
self.model = model
self.base_url = "https://api.anthropic.com/v1"
self.session = None
async def generate(self, request: LLMRequest) -> LLMResponse:
if not self.session:
self.session = aiohttp.ClientSession()
payload = {
"model": self.model,
"max_tokens": request.max_tokens,
"temperature": request.temperature,
"top_p": request.top_p,
"messages": [{"role": "user", "content": request.prompt}]
}
if request.system_prompt:
payload["system"] = request.system_prompt
if request.stop_sequences:
payload["stop_sequences"] = request.stop_sequences
headers = {
"x-api-key": self.api_key,
"anthropic-version": "2023-06-01",
"Content-Type": "application/json"
}
start_time = time.time()
async with self.session.post(
f"{self.base_url}/messages",
json=payload,
headers=headers
) as response:
if response.status != 200:
error_text = await response.text()
raise Exception(f"Anthropic API error: {response.status} - {error_text}")
result = await response.json()
latency_ms = (time.time() - start_time) * 1000
content = result["content"][0]["text"]
usage = result.get("usage", {})
return LLMResponse(
text=content,
model_used=self.model,
provider="anthropic",
tokens_used=usage.get("input_tokens", 0) + usage.get("output_tokens", 0),
latency_ms=latency_ms,
finish_reason=result.get("stop_reason", "unknown"),
metadata={"usage": usage}
)
def is_available(self) -> bool:
return bool(self.api_key)
def get_cost_per_token(self) -> float:
return 0.000015
async def close(self):
if self.session:
await self.session.close()
class GoogleGeminiProvider(LLMProvider):
def __init__(self, api_key: str, model: str = "gemini-pro"):
self.api_key = api_key
self.model = model
self.base_url = "https://generativelanguage.googleapis.com/v1beta"
self.session = None
async def generate(self, request: LLMRequest) -> LLMResponse:
if not self.session:
self.session = aiohttp.ClientSession()
contents = []
if request.system_prompt:
contents.append({"role": "user", "parts": [{"text": request.system_prompt}]})
contents.append({"role": "model", "parts": [{"text": "Understood. I will follow these instructions."}]})
contents.append({"role": "user", "parts": [{"text": request.prompt}]})
payload = {
"contents": contents,
"generationConfig": {
"temperature": request.temperature,
"maxOutputTokens": request.max_tokens,
"topP": request.top_p
}
}
if request.stop_sequences:
payload["generationConfig"]["stopSequences"] = request.stop_sequences
start_time = time.time()
async with self.session.post(
f"{self.base_url}/models/{self.model}:generateContent?key={self.api_key}",
json=payload,
headers={"Content-Type": "application/json"}
) as response:
if response.status != 200:
error_text = await response.text()
raise Exception(f"Google Gemini API error: {response.status} - {error_text}")
result = await response.json()
latency_ms = (time.time() - start_time) * 1000
if "candidates" not in result or len(result["candidates"]) == 0:
raise Exception("No candidates in Gemini response")
candidate = result["candidates"][0]
text = candidate["content"]["parts"][0]["text"]
usage_metadata = result.get("usageMetadata", {})
total_tokens = usage_metadata.get("totalTokenCount", 0)
return LLMResponse(
text=text,
model_used=self.model,
provider="google",
tokens_used=total_tokens,
latency_ms=latency_ms,
finish_reason=candidate.get("finishReason", "unknown"),
metadata={"usage": usage_metadata}
)
def is_available(self) -> bool:
return bool(self.api_key)
def get_cost_per_token(self) -> float:
return 0.0000005
async def close(self):
if self.session:
await self.session.close()
class LocalLlamaProvider(LLMProvider):
def __init__(self, base_url: str = "http://localhost:8080", model: str = "local-llama"):
self.base_url = base_url
self.model = model
self.session = None
async def generate(self, request: LLMRequest) -> LLMResponse:
if not self.session:
self.session = aiohttp.ClientSession()
full_prompt = ""
if request.system_prompt:
full_prompt = f"<|system|>\n{request.system_prompt}\n<|user|>\n{request.prompt}\n<|assistant|>\n"
else:
full_prompt = f"<|user|>\n{request.prompt}\n<|assistant|>\n"
payload = {
"prompt": full_prompt,
"temperature": request.temperature,
"max_tokens": request.max_tokens,
"top_p": request.top_p,
"stop": request.stop_sequences or ["<|user|>", "<|system|>"]
}
start_time = time.time()
try:
async with self.session.post(
f"{self.base_url}/completion",
json=payload,
timeout=aiohttp.ClientTimeout(total=120)
) as response:
if response.status != 200:
error_text = await response.text()
raise Exception(f"Local Llama error: {response.status} - {error_text}")
result = await response.json()
except aiohttp.ClientError as e:
raise Exception(f"Failed to connect to local Llama server: {str(e)}")
latency_ms = (time.time() - start_time) * 1000
return LLMResponse(
text=result.get("content", ""),
model_used=self.model,
provider="local_llama",
tokens_used=result.get("tokens_evaluated", 0) + result.get("tokens_predicted", 0),
latency_ms=latency_ms,
finish_reason=result.get("stop_reason", "unknown"),
metadata=result
)
def is_available(self) -> bool:
return True
def get_cost_per_token(self) -> float:
return 0.0
async def close(self):
if self.session:
await self.session.close()
class OllamaProvider(LLMProvider):
def __init__(self, base_url: str = "http://localhost:11434", model: str = "llama2"):
self.base_url = base_url
self.model = model
self.session = None
async def generate(self, request: LLMRequest) -> LLMResponse:
if not self.session:
self.session = aiohttp.ClientSession()
payload = {
"model": self.model,
"prompt": request.prompt,
"system": request.system_prompt,
"options": {
"temperature": request.temperature,
"num_predict": request.max_tokens,
"top_p": request.top_p
},
"stream": False
}
if request.stop_sequences:
payload["options"]["stop"] = request.stop_sequences
start_time = time.time()
try:
async with self.session.post(
f"{self.base_url}/api/generate",
json=payload,
timeout=aiohttp.ClientTimeout(total=120)
) as response:
if response.status != 200:
error_text = await response.text()
raise Exception(f"Ollama error: {response.status} - {error_text}")
result = await response.json()
except aiohttp.ClientError as e:
raise Exception(f"Failed to connect to Ollama server: {str(e)}")
latency_ms = (time.time() - start_time) * 1000
return LLMResponse(
text=result.get("response", ""),
model_used=self.model,
provider="ollama",
tokens_used=result.get("prompt_eval_count", 0) + result.get("eval_count", 0),
latency_ms=latency_ms,
finish_reason="stop" if result.get("done", False) else "unknown",
metadata=result
)
def is_available(self) -> bool:
return True
def get_cost_per_token(self) -> float:
return 0.0
async def close(self):
if self.session:
await self.session.close()
class LLMManager:
def __init__(self, config: Dict[str, Any]):
self.providers: Dict[str, LLMProvider] = {}
self.default_provider = None
self.task_routing = {}
self.request_history = []
self.total_tokens_used = 0
self.total_cost = 0.0
self._initialize_providers(config)
def _initialize_providers(self, config: Dict[str, Any]):
if "openai" in config and config["openai"].get("enabled", False):
self.providers["openai"] = OpenAIProvider(
api_key=config["openai"]["api_key"],
model=config["openai"].get("model", "gpt-4")
)
if config["openai"].get("default", False):
self.default_provider = "openai"
if "anthropic" in config and config["anthropic"].get("enabled", False):
self.providers["anthropic"] = AnthropicProvider(
api_key=config["anthropic"]["api_key"],
model=config["anthropic"].get("model", "claude-3-5-sonnet-20241022")
)
if config["anthropic"].get("default", False):
self.default_provider = "anthropic"
if "google" in config and config["google"].get("enabled", False):
self.providers["google"] = GoogleGeminiProvider(
api_key=config["google"]["api_key"],
model=config["google"].get("model", "gemini-pro")
)
if config["google"].get("default", False):
self.default_provider = "google"
if "local_llama" in config and config["local_llama"].get("enabled", False):
self.providers["local_llama"] = LocalLlamaProvider(
base_url=config["local_llama"].get("base_url", "http://localhost:8080"),
model=config["local_llama"].get("model", "local-llama")
)
if config["local_llama"].get("default", False):
self.default_provider = "local_llama"
if "ollama" in config and config["ollama"].get("enabled", False):
self.providers["ollama"] = OllamaProvider(
base_url=config["ollama"].get("base_url", "http://localhost:11434"),
model=config["ollama"].get("model", "llama2")
)
if config["ollama"].get("default", False):
self.default_provider = "ollama"
self.task_routing = config.get("task_routing", {})
if not self.default_provider and self.providers:
self.default_provider = list(self.providers.keys())[0]
async def generate(self, request: LLMRequest) -> LLMResponse:
provider_name = self._select_provider(request)
if provider_name not in self.providers:
raise Exception(f"Provider {provider_name} not configured")
provider = self.providers[provider_name]
if not provider.is_available():
fallback_provider = self._get_fallback_provider(provider_name)
if fallback_provider:
provider_name = fallback_provider
provider = self.providers[provider_name]
else:
raise Exception(f"Provider {provider_name} not available and no fallback configured")
try:
response = await provider.generate(request)
self.total_tokens_used += response.tokens_used
cost = response.tokens_used * provider.get_cost_per_token()
self.total_cost += cost
self.request_history.append({
'timestamp': time.time(),
'provider': provider_name,
'model': response.model_used,
'tokens': response.tokens_used,
'cost': cost,
'latency_ms': response.latency_ms,
'task_type': request.task_type
})
return response
except Exception as e:
fallback_provider = self._get_fallback_provider(provider_name)
if fallback_provider:
return await self.generate(request)
else:
raise e
def _select_provider(self, request: LLMRequest) -> str:
if request.model_preference and request.model_preference in self.providers:
return request.model_preference
if request.task_type and request.task_type in self.task_routing:
return self.task_routing[request.task_type]
return self.default_provider
def _get_fallback_provider(self, failed_provider: str) -> Optional[str]:
available_providers = [
name for name, provider in self.providers.items()
if name != failed_provider and provider.is_available()
]
if available_providers:
return available_providers[0]
return None
async def generate_with_retry(self, request: LLMRequest, max_retries: int = 3) -> LLMResponse:
last_exception = None
for attempt in range(max_retries):
try:
return await self.generate(request)
except Exception as e:
last_exception = e
if attempt < max_retries - 1:
await asyncio.sleep(2 ** attempt)
raise last_exception
async def batch_generate(self, requests: List[LLMRequest], max_concurrent: int = 5) -> List[LLMResponse]:
semaphore = asyncio.Semaphore(max_concurrent)
async def generate_with_semaphore(req):
async with semaphore:
return await self.generate(req)
tasks = [generate_with_semaphore(req) for req in requests]
return await asyncio.gather(*tasks)
def get_usage_statistics(self) -> Dict[str, Any]:
provider_stats = {}
for entry in self.request_history:
provider = entry['provider']
if provider not in provider_stats:
provider_stats[provider] = {
'request_count': 0,
'total_tokens': 0,
'total_cost': 0.0,
'avg_latency_ms': 0.0,
'latencies': []
}
stats = provider_stats[provider]
stats['request_count'] += 1
stats['total_tokens'] += entry['tokens']
stats['total_cost'] += entry['cost']
stats['latencies'].append(entry['latency_ms'])
for provider, stats in provider_stats.items():
if stats['latencies']:
stats['avg_latency_ms'] = sum(stats['latencies']) / len(stats['latencies'])
del stats['latencies']
return {
'total_requests': len(self.request_history),
'total_tokens': self.total_tokens_used,
'total_cost': self.total_cost,
'by_provider': provider_stats
}
async def close_all(self):
for provider in self.providers.values():
if hasattr(provider, 'close'):
await provider.close()
This LLM manager provides comprehensive support for multiple providers with a unified interface. The manager handles provider-specific API details, allowing agents to work with a consistent request and response format regardless of which model actually processes the request. The implementation supports both cloud providers like OpenAI, Anthropic, and Google, as well as local providers like llama.cpp and Ollama.
The manager implements intelligent provider selection based on task type, allowing the system to route different kinds of tasks to models best suited for them. Code analysis tasks might route to a model fine-tuned for code understanding, while natural language processing tasks might route to a model optimized for general text. The fallback mechanism ensures robustness by automatically retrying failed requests with alternative providers.
The usage tracking functionality provides visibility into system costs and performance. By recording every request with its token usage, cost, and latency, the manager enables analysis of which tasks consume the most resources and which providers offer the best performance for different task types. This information guides optimization efforts and helps manage operational costs.
Implementing Advanced Reasoning Patterns
The multi-agent system employs sophisticated reasoning patterns that go beyond simple prompt-response interactions. Chain-of-thought reasoning breaks complex problems into sequences of simpler steps, with each step building on previous results. Self-reflection enables the system to evaluate its own reasoning and identify potential errors or gaps. Tree-of-thought reasoning explores multiple solution paths in parallel, evaluating alternatives before committing to a final answer.
Chain-of-thought reasoning is particularly valuable for complex analytical tasks that require multiple steps of logical reasoning. When analyzing whether a system can support a new feature, the chain-of-thought approach might first identify the components involved, then analyze their current capabilities, then identify gaps between current capabilities and requirements, then estimate the effort to close those gaps, and finally synthesize this information into a feasibility assessment. Each step produces intermediate results that inform subsequent steps.
The implementation of chain-of-thought reasoning involves constructing prompts that explicitly request step-by-step reasoning. Rather than asking "Can the system support feature X?", the prompt asks "Let's analyze whether the system can support feature X. First, identify which components would be involved. Then, for each component, assess its current capabilities. Next, identify gaps between current capabilities and what feature X requires. Finally, estimate the effort to address these gaps." This structured prompt guides the language model to produce reasoning that follows the desired logical sequence.
Self-reflection adds a critical evaluation layer where the system examines its own reasoning for potential flaws. After generating an analysis, a self-reflection prompt asks the model to critique its work. The prompt might ask "Review the analysis above. Are there any unsupported claims? Are there alternative interpretations of the evidence? Are there important factors that were not considered?" This self-critique often identifies weaknesses in the initial analysis, prompting additional investigation or qualifying statements that acknowledge uncertainty.
Here is an implementation of reasoning pattern helpers:
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
@dataclass
class ReasoningStep:
step_number: int
description: str
prompt: str
result: Optional[str] = None
confidence: Optional[float] = None
supporting_evidence: List[str] = None
class ReasoningPatterns:
def __init__(self, llm_manager):
self.llm_manager = llm_manager
async def chain_of_thought(self,
initial_question: str,
steps: List[str],
context: Dict[str, Any]) -> Dict[str, Any]:
reasoning_chain = []
accumulated_context = context.copy()
for i, step_description in enumerate(steps):
step = ReasoningStep(
step_number=i + 1,
description=step_description,
prompt=self._construct_step_prompt(
step_description,
accumulated_context,
reasoning_chain
)
)
response = await self.llm_manager.generate(LLMRequest(
prompt=step.prompt,
system_prompt="You are an expert code analyst. Provide detailed, factual analysis based on the evidence provided.",
temperature=0.3,
max_tokens=1500,
task_type="reasoning"
))
step.result = response.text
step.confidence = self._assess_confidence(step.result)
step.supporting_evidence = self._extract_evidence_references(step.result)
reasoning_chain.append(step)
accumulated_context[f"step_{i+1}_result"] = step.result
final_synthesis = await self._synthesize_chain(
initial_question,
reasoning_chain,
accumulated_context
)
return {
'question': initial_question,
'reasoning_steps': [
{
'step': s.step_number,
'description': s.description,
'result': s.result,
'confidence': s.confidence
}
for s in reasoning_chain
],
'synthesis': final_synthesis,
'overall_confidence': self._calculate_overall_confidence(reasoning_chain)
}
def _construct_step_prompt(self,
step_description: str,
context: Dict[str, Any],
previous_steps: List[ReasoningStep]) -> str:
prompt_parts = []
if previous_steps:
prompt_parts.append("Previous analysis steps:")
for prev_step in previous_steps:
prompt_parts.append(f"\nStep {prev_step.step_number}: {prev_step.description}")
prompt_parts.append(f"Result: {prev_step.result}\n")
if context:
prompt_parts.append("\nRelevant context:")
for key, value in context.items():
if not key.startswith('step_'):
if isinstance(value, str) and len(value) < 500:
prompt_parts.append(f"{key}: {value}")
elif isinstance(value, (list, dict)):
prompt_parts.append(f"{key}: {str(value)[:500]}")
prompt_parts.append(f"\nCurrent step: {step_description}")
prompt_parts.append("\nProvide a detailed analysis for this step. Support your conclusions with specific evidence from the context. If you cannot determine something with certainty, acknowledge the uncertainty.")
return "\n".join(prompt_parts)
async def _synthesize_chain(self,
question: str,
reasoning_chain: List[ReasoningStep],
context: Dict[str, Any]) -> str:
synthesis_prompt_parts = [
f"Original question: {question}",
"\nReasoning chain:"
]
for step in reasoning_chain:
synthesis_prompt_parts.append(f"\n{step.description}")
synthesis_prompt_parts.append(f"Finding: {step.result}")
synthesis_prompt_parts.append("\nBased on this chain of reasoning, provide a comprehensive answer to the original question. Integrate the findings from all steps into a coherent conclusion.")
response = await self.llm_manager.generate(LLMRequest(
prompt="\n".join(synthesis_prompt_parts),
system_prompt="You are an expert analyst. Synthesize the reasoning chain into a clear, comprehensive conclusion.",
temperature=0.3,
max_tokens=2000,
task_type="synthesis"
))
return response.text
async def self_reflection(self,
analysis: str,
original_question: str,
evidence: List[str]) -> Dict[str, Any]:
reflection_prompt = f"""
Original question: {original_question}
Analysis produced:
{analysis}
Available evidence:
{chr(10).join(f"- {e}" for e in evidence[:10])}
Please critically evaluate this analysis:
1. Are all claims supported by the available evidence?
2. Are there alternative interpretations of the evidence that were not considered?
3. Are there important factors or perspectives that were overlooked?
4. What is the confidence level for each major conclusion?
5. What additional information would strengthen this analysis?
Provide a structured critique that identifies strengths and weaknesses.
"""
response = await self.llm_manager.generate(LLMRequest(
prompt=reflection_prompt,
system_prompt="You are a critical reviewer. Evaluate the analysis objectively, identifying both strengths and potential weaknesses.",
temperature=0.4,
max_tokens=2000,
task_type="reflection"
))
critique = response.text
issues = await self._extract_identified_issues(critique)
if issues:
refinement = await self._refine_analysis(
analysis,
critique,
issues,
evidence
)
return {
'original_analysis': analysis,
'critique': critique,
'identified_issues': issues,
'refined_analysis': refinement,
'improvement_made': True
}
return {
'original_analysis': analysis,
'critique': critique,
'identified_issues': [],
'refined_analysis': analysis,
'improvement_made': False
}
async def tree_of_thought(self,
question: str,
hypotheses: List[str],
context: Dict[str, Any],
max_depth: int = 3) -> Dict[str, Any]:
exploration_results = []
for hypothesis in hypotheses:
result = await self._explore_hypothesis(
question,
hypothesis,
context,
depth=0,
max_depth=max_depth
)
exploration_results.append(result)
evaluation = await self._evaluate_hypotheses(
question,
exploration_results
)
best_hypothesis = max(
exploration_results,
key=lambda x: x['final_score']
)
return {
'question': question,
'hypotheses_explored': len(hypotheses),
'exploration_results': exploration_results,
'evaluation': evaluation,
'best_hypothesis': best_hypothesis,
'conclusion': best_hypothesis['conclusion']
}
async def _explore_hypothesis(self,
question: str,
hypothesis: str,
context: Dict[str, Any],
depth: int,
max_depth: int) -> Dict[str, Any]:
exploration_prompt = f"""
Question: {question}
Hypothesis to explore: {hypothesis}
Context:
{self._format_context(context)}
Explore this hypothesis:
1. What evidence supports this hypothesis?
2. What evidence contradicts this hypothesis?
3. What are the implications if this hypothesis is correct?
4. How confident can we be in this hypothesis?
Provide a structured analysis.
"""
response = await self.llm_manager.generate(LLMRequest(
prompt=exploration_prompt,
system_prompt="You are an analytical expert. Objectively evaluate the hypothesis based on available evidence.",
temperature=0.4,
max_tokens=1500,
task_type="hypothesis_exploration"
))
analysis = response.text
score = await self._score_hypothesis(analysis, context)
sub_hypotheses = []
if depth < max_depth:
sub_hypotheses = await self._generate_sub_hypotheses(
hypothesis,
analysis
)
return {
'hypothesis': hypothesis,
'depth': depth,
'analysis': analysis,
'score': score,
'sub_hypotheses': sub_hypotheses,
'final_score': score,
'conclusion': analysis
}
async def _evaluate_hypotheses(self,
question: str,
results: List[Dict[str, Any]]) -> str:
evaluation_prompt = f"""
Question: {question}
Hypotheses explored:
{self._format_hypothesis_results(results)}
Compare these hypotheses:
1. Which hypothesis is best supported by evidence?
2. Which hypothesis has the strongest logical foundation?
3. Are there ways to combine insights from multiple hypotheses?
4. What is the most reasonable conclusion?
Provide a comparative evaluation.
"""
response = await self.llm_manager.generate(LLMRequest(
prompt=evaluation_prompt,
system_prompt="You are an expert evaluator. Compare the hypotheses objectively and identify the strongest conclusion.",
temperature=0.3,
max_tokens=2000,
task_type="evaluation"
))
return response.text
def _assess_confidence(self, text: str) -> float:
confidence_indicators = {
'certain': 0.95,
'very confident': 0.9,
'confident': 0.8,
'likely': 0.7,
'probably': 0.65,
'possibly': 0.5,
'uncertain': 0.3,
'unclear': 0.2
}
text_lower = text.lower()
for indicator, score in confidence_indicators.items():
if indicator in text_lower:
return score
return 0.6
def _extract_evidence_references(self, text: str) -> List[str]:
return []
def _calculate_overall_confidence(self, steps: List[ReasoningStep]) -> float:
if not steps:
return 0.0
confidences = [s.confidence for s in steps if s.confidence is not None]
if not confidences:
return 0.5
return sum(confidences) / len(confidences)
async def _extract_identified_issues(self, critique: str) -> List[str]:
return []
async def _refine_analysis(self,
original: str,
critique: str,
issues: List[str],
evidence: List[str]) -> str:
return original
def _format_context(self, context: Dict[str, Any]) -> str:
return str(context)
async def _score_hypothesis(self, analysis: str, context: Dict[str, Any]) -> float:
return 0.5
async def _generate_sub_hypotheses(self, hypothesis: str, analysis: str) -> List[str]:
return []
def _format_hypothesis_results(self, results: List[Dict[str, Any]]) -> str:
formatted = []
for i, result in enumerate(results):
formatted.append(f"\nHypothesis {i+1}: {result['hypothesis']}")
formatted.append(f"Score: {result['score']:.2f}")
formatted.append(f"Analysis: {result['analysis'][:200]}...")
return "\n".join(formatted)
This reasoning patterns implementation provides reusable components for sophisticated analytical workflows. The chain-of-thought method orchestrates multi-step reasoning, maintaining context across steps and synthesizing results into a coherent conclusion. The self-reflection method implements critical evaluation of generated analyses, identifying potential weaknesses and refining conclusions. The tree-of-thought method explores multiple hypotheses in parallel, evaluating each and selecting the most promising.
These reasoning patterns significantly improve the quality of analysis compared to simple single-shot prompting. By breaking complex questions into steps, the system can tackle problems that would overwhelm a single prompt. By reflecting on its own reasoning, the system catches errors and acknowledges uncertainty appropriately. By exploring multiple hypotheses, the system avoids premature commitment to potentially incorrect conclusions.
Supporting Multiple GPU Architectures for Local LLM Inference
Running language models locally requires GPU acceleration for acceptable performance. Different hardware platforms use different GPU architectures and software stacks. NVIDIA GPUs use CUDA, AMD GPUs use ROCm, Intel GPUs use oneAPI, and Apple Silicon uses Metal Performance Shaders. The system must support all these platforms to enable deployment across diverse hardware environments.
The abstraction layer for GPU support detects the available hardware and configures the inference engine accordingly. When initializing a local LLM, the system queries the hardware to determine what GPUs are available. On a system with NVIDIA GPUs, it configures the inference engine to use CUDA. On a system with AMD GPUs, it uses ROCm. On an Apple Silicon Mac, it uses Metal. This automatic detection ensures that the system utilizes available hardware acceleration without requiring manual configuration.
Different inference engines have different strengths and platform support. llama.cpp provides broad platform support and runs efficiently on CPUs and GPUs across NVIDIA, AMD, Intel, and Apple platforms. vLLM offers optimized inference for NVIDIA GPUs with features like paged attention that dramatically improve throughput. Ollama provides a user-friendly interface for running models locally with automatic model management. The system can use different engines based on available hardware and performance requirements.
Here is an implementation of GPU detection and inference engine configuration:
import platform
import subprocess
import os
from typing import Dict, List, Any, Optional
from enum import Enum
class GPUArchitecture(Enum):
NVIDIA_CUDA = "nvidia_cuda"
AMD_ROCM = "amd_rocm"
INTEL_ONEAPI = "intel_oneapi"
APPLE_MPS = "apple_mps"
CPU_ONLY = "cpu_only"
class InferenceEngine(Enum):
LLAMA_CPP = "llama_cpp"
VLLM = "vllm"
OLLAMA = "ollama"
TRANSFORMERS = "transformers"
class GPUDetector:
@staticmethod
def detect_gpu_architecture() -> GPUArchitecture:
system = platform.system()
if system == "Darwin":
machine = platform.machine()
if machine == "arm64":
return GPUArchitecture.APPLE_MPS
if GPUDetector._has_nvidia_gpu():
return GPUArchitecture.NVIDIA_CUDA
if GPUDetector._has_amd_gpu():
return GPUArchitecture.AMD_ROCM
if GPUDetector._has_intel_gpu():
return GPUArchitecture.INTEL_ONEAPI
return GPUArchitecture.CPU_ONLY
@staticmethod
def _has_nvidia_gpu() -> bool:
try:
result = subprocess.run(
['nvidia-smi'],
capture_output=True,
text=True,
timeout=5
)
return result.returncode == 0
except (FileNotFoundError, subprocess.TimeoutExpired):
return False
@staticmethod
def _has_amd_gpu() -> bool:
try:
result = subprocess.run(
['rocm-smi'],
capture_output=True,
text=True,
timeout=5
)
return result.returncode == 0
except (FileNotFoundError, subprocess.TimeoutExpired):
return False
@staticmethod
def _has_intel_gpu() -> bool:
if platform.system() == "Linux":
try:
with open('/proc/cpuinfo', 'r') as f:
cpuinfo = f.read()
if 'Intel' in cpuinfo:
result = subprocess.run(
['lspci'],
capture_output=True,
text=True,
timeout=5
)
return 'Intel' in result.stdout and 'VGA' in result.stdout
except:
pass
return False
@staticmethod
def get_gpu_info() -> Dict[str, Any]:
arch = GPUDetector.detect_gpu_architecture()
info = {
'architecture': arch.value,
'devices': [],
'total_memory_gb': 0
}
if arch == GPUArchitecture.NVIDIA_CUDA:
info.update(GPUDetector._get_nvidia_info())
elif arch == GPUArchitecture.AMD_ROCM:
info.update(GPUDetector._get_amd_info())
elif arch == GPUArchitecture.APPLE_MPS:
info.update(GPUDetector._get_apple_info())
return info
@staticmethod
def _get_nvidia_info() -> Dict[str, Any]:
try:
result = subprocess.run(
['nvidia-smi', '--query-gpu=name,memory.total', '--format=csv,noheader'],
capture_output=True,
text=True,
timeout=5
)
devices = []
total_memory = 0
for line in result.stdout.strip().split('\n'):
if line:
parts = line.split(',')
name = parts[0].strip()
memory_str = parts[1].strip().replace(' MiB', '')
memory_gb = int(memory_str) / 1024
devices.append({
'name': name,
'memory_gb': memory_gb
})
total_memory += memory_gb
return {
'devices': devices,
'total_memory_gb': total_memory
}
except:
return {'devices': [], 'total_memory_gb': 0}
@staticmethod
def _get_amd_info() -> Dict[str, Any]:
return {'devices': [], 'total_memory_gb': 0}
@staticmethod
def _get_apple_info() -> Dict[str, Any]:
try:
result = subprocess.run(
['sysctl', 'hw.memsize'],
capture_output=True,
text=True,
timeout=5
)
memory_bytes = int(result.stdout.split(':')[1].strip())
memory_gb = memory_bytes / (1024 ** 3)
return {
'devices': [{'name': 'Apple Silicon Unified Memory', 'memory_gb': memory_gb}],
'total_memory_gb': memory_gb
}
except:
return {'devices': [], 'total_memory_gb': 0}
class LocalInferenceManager:
def __init__(self, model_path: str, config: Optional[Dict[str, Any]] = None):
self.model_path = model_path
self.config = config or {}
self.gpu_arch = GPUDetector.detect_gpu_architecture()
self.gpu_info = GPUDetector.get_gpu_info()
self.engine = None
self.engine_type = None
self._initialize_engine()
def _initialize_engine(self):
preferred_engine = self.config.get('preferred_engine', 'auto')
if preferred_engine == 'auto':
self.engine_type = self._select_best_engine()
else:
self.engine_type = InferenceEngine(preferred_engine)
if self.engine_type == InferenceEngine.LLAMA_CPP:
self._initialize_llama_cpp()
elif self.engine_type == InferenceEngine.VLLM:
self._initialize_vllm()
elif self.engine_type == InferenceEngine.OLLAMA:
self._initialize_ollama()
elif self.engine_type == InferenceEngine.TRANSFORMERS:
self._initialize_transformers()
def _select_best_engine(self) -> InferenceEngine:
if self.gpu_arch == GPUArchitecture.NVIDIA_CUDA:
if self.gpu_info['total_memory_gb'] >= 24:
return InferenceEngine.VLLM
else:
return InferenceEngine.LLAMA_CPP
elif self.gpu_arch == GPUArchitecture.APPLE_MPS:
return InferenceEngine.LLAMA_CPP
elif self.gpu_arch == GPUArchitecture.AMD_ROCM:
return InferenceEngine.LLAMA_CPP
else:
return InferenceEngine.LLAMA_CPP
def _initialize_llama_cpp(self):
try:
from llama_cpp import Llama
n_gpu_layers = 0
if self.gpu_arch != GPUArchitecture.CPU_ONLY:
n_gpu_layers = self.config.get('n_gpu_layers', -1)
self.engine = Llama(
model_path=self.model_path,
n_gpu_layers=n_gpu_layers,
n_ctx=self.config.get('context_length', 4096),
n_batch=self.config.get('batch_size', 512),
n_threads=self.config.get('n_threads', 8),
verbose=self.config.get('verbose', False)
)
except ImportError:
raise Exception("llama-cpp-python not installed. Install with: pip install llama-cpp-python")
def _initialize_vllm(self):
try:
from vllm import LLM, SamplingParams
if self.gpu_arch != GPUArchitecture.NVIDIA_CUDA:
raise Exception("vLLM requires NVIDIA CUDA GPUs")
self.engine = LLM(
model=self.model_path,
tensor_parallel_size=self.config.get('tensor_parallel_size', 1),
gpu_memory_utilization=self.config.get('gpu_memory_utilization', 0.9),
max_model_len=self.config.get('max_model_len', 4096)
)
except ImportError:
raise Exception("vLLM not installed. Install with: pip install vllm")
def _initialize_ollama(self):
pass
def _initialize_transformers(self):
try:
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
device = "cpu"
if self.gpu_arch == GPUArchitecture.NVIDIA_CUDA:
device = "cuda"
elif self.gpu_arch == GPUArchitecture.APPLE_MPS:
device = "mps"
self.tokenizer = AutoTokenizer.from_pretrained(self.model_path)
self.engine = AutoModelForCausalLM.from_pretrained(
self.model_path,
torch_dtype=torch.float16 if device != "cpu" else torch.float32,
device_map="auto" if device != "cpu" else None
)
if device == "cpu":
self.engine = self.engine.to(device)
except ImportError:
raise Exception("transformers not installed. Install with: pip install transformers torch")
def generate(self, prompt: str, max_tokens: int = 512, temperature: float = 0.7) -> str:
if self.engine_type == InferenceEngine.LLAMA_CPP:
return self._generate_llama_cpp(prompt, max_tokens, temperature)
elif self.engine_type == InferenceEngine.VLLM:
return self._generate_vllm(prompt, max_tokens, temperature)
elif self.engine_type == InferenceEngine.TRANSFORMERS:
return self._generate_transformers(prompt, max_tokens, temperature)
else:
raise Exception(f"Generation not implemented for {self.engine_type}")
def _generate_llama_cpp(self, prompt: str, max_tokens: int, temperature: float) -> str:
output = self.engine(
prompt,
max_tokens=max_tokens,
temperature=temperature,
top_p=0.95,
echo=False
)
return output['choices'][0]['text']
def _generate_vllm(self, prompt: str, max_tokens: int, temperature: float) -> str:
from vllm import SamplingParams
sampling_params = SamplingParams(
temperature=temperature,
max_tokens=max_tokens,
top_p=0.95
)
outputs = self.engine.generate([prompt], sampling_params)
return outputs[0].outputs[0].text
def _generate_transformers(self, prompt: str, max_tokens: int, temperature: float) -> str:
import torch
inputs = self.tokenizer(prompt, return_tensors="pt")
device = next(self.engine.parameters()).device
inputs = {k: v.to(device) for k, v in inputs.items()}
with torch.no_grad():
outputs = self.engine.generate(
**inputs,
max_new_tokens=max_tokens,
temperature=temperature,
top_p=0.95,
do_sample=True
)
generated_text = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
return generated_text[len(prompt):]
def get_system_info(self) -> Dict[str, Any]:
return {
'gpu_architecture': self.gpu_arch.value,
'gpu_info': self.gpu_info,
'inference_engine': self.engine_type.value if self.engine_type else None,
'model_path': self.model_path
}
This GPU detection and inference management implementation provides automatic hardware detection and optimal engine selection. The system detects whether NVIDIA, AMD, Intel, or Apple GPUs are available and configures the inference engine accordingly. For NVIDIA GPUs with sufficient memory, it uses vLLM for maximum throughput. For other platforms, it uses llama.cpp which provides broad compatibility. The abstraction allows the same high-level code to run efficiently across diverse hardware platforms.
CHAPTER 6: REPORT GENERATION, SWOT ANALYSIS, AND COMPLETE SYSTEM INTEGRATION
Building the Comprehensive Report Generation System
The report generation system synthesizes findings from all specialized agents into coherent documents tailored for different stakeholder audiences. A comprehensive analysis report must serve multiple purposes simultaneously. Executives need high-level summaries that highlight strategic implications and business impacts. Architects need detailed technical findings with specific recommendations. Developers need actionable items with clear priorities. Project managers need effort estimates and risk assessments. The report generation system creates a multi-layered document structure that addresses all these needs.
The report generator operates as a specialized agent that consumes outputs from other agents and orchestrates their synthesis into structured reports. It begins by gathering all analysis results from the task tree executed by the orchestrator. These results include architectural findings, code quality metrics, security assessments, requirements traceability data, domain model analysis, and operational considerations. The generator organizes this information into a hierarchical structure that flows from high-level executive summary through detailed technical sections to appendices with supporting data.
The executive summary distills the entire analysis into a concise narrative that non-technical stakeholders can understand. It highlights the most critical findings, expresses technical concepts in business terms, and provides clear recommendations with justifications. When the analysis identifies architectural weaknesses, the executive summary explains the business implications rather than dwelling on technical details. A finding about tight coupling between microservices becomes a statement about reduced agility and increased risk when making changes to support new business requirements.
The SWOT analysis framework provides a structured way to present findings that resonates with business stakeholders. Strengths represent positive aspects of the current system that should be preserved and leveraged. Weaknesses identify areas requiring improvement or remediation. Opportunities highlight possibilities for enhancement or strategic advantage. Threats describe risks that could impact the system's ability to meet business objectives. By categorizing findings into this framework, the report makes complex technical analysis accessible to business decision-makers.
Here is an implementation of the report generation system:
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from datetime import datetime
import json
@dataclass
class Finding:
category: str
severity: str
title: str
description: str
evidence: List[str]
recommendations: List[str]
business_impact: str
technical_details: Dict[str, Any]
affected_components: List[str]
@dataclass
class SWOTItem:
category: str # strength, weakness, opportunity, threat
title: str
description: str
impact_level: str # high, medium, low
related_findings: List[str]
stakeholder_relevance: List[str]
class ReportGenerator:
def __init__(self, llm_manager):
self.llm_manager = llm_manager
self.findings: List[Finding] = []
self.swot_items: List[SWOTItem] = []
async def generate_comprehensive_report(self,
analysis_results: Dict[str, Any],
repository_info: Dict[str, Any]) -> Dict[str, Any]:
await self._extract_findings(analysis_results)
await self._generate_swot_analysis()
executive_summary = await self._generate_executive_summary(repository_info)
technical_analysis = await self._generate_technical_analysis()
architecture_section = await self._generate_architecture_section(
analysis_results.get('architecture_analysis', {})
)
code_quality_section = await self._generate_code_quality_section(
analysis_results.get('code_quality_analysis', {})
)
security_section = await self._generate_security_section(
analysis_results.get('security_analysis', {})
)
requirements_section = await self._generate_requirements_section(
analysis_results.get('requirements_analysis', {})
)
recommendations = await self._generate_recommendations()
visualizations = await self._generate_visualizations(analysis_results)
report = {
'metadata': {
'generated_at': datetime.now().isoformat(),
'repository': repository_info.get('name', 'Unknown'),
'analysis_scope': repository_info.get('scope', 'Full system'),
'report_version': '1.0'
},
'executive_summary': executive_summary,
'swot_analysis': {
'strengths': [s for s in self.swot_items if s.category == 'strength'],
'weaknesses': [w for w in self.swot_items if w.category == 'weakness'],
'opportunities': [o for o in self.swot_items if o.category == 'opportunity'],
'threats': [t for t in self.swot_items if t.category == 'threat']
},
'technical_analysis': technical_analysis,
'detailed_sections': {
'architecture': architecture_section,
'code_quality': code_quality_section,
'security': security_section,
'requirements': requirements_section
},
'recommendations': recommendations,
'visualizations': visualizations,
'findings_summary': {
'total_findings': len(self.findings),
'by_severity': self._count_by_severity(),
'by_category': self._count_by_category()
}
}
formatted_report = await self._format_report_for_stakeholders(report)
return formatted_report
async def _extract_findings(self, analysis_results: Dict[str, Any]):
for task_type, task_result in analysis_results.items():
if isinstance(task_result, dict):
await self._extract_findings_from_task(task_type, task_result)
async def _extract_findings_from_task(self, task_type: str, task_result: Dict[str, Any]):
if 'violations' in task_result:
for violation in task_result['violations']:
finding = Finding(
category='architecture',
severity=violation.get('severity', 'medium'),
title=violation.get('title', 'Architectural violation'),
description=violation.get('description', ''),
evidence=[violation.get('location', '')],
recommendations=violation.get('recommendations', []),
business_impact=await self._infer_business_impact(violation),
technical_details=violation,
affected_components=violation.get('affected_components', [])
)
self.findings.append(finding)
if 'hotspots' in task_result:
for hotspot in task_result['hotspots'][:10]:
finding = Finding(
category='maintainability',
severity='medium',
title=f"High complexity hotspot: {hotspot.get('name', 'Unknown')}",
description=f"Component has {hotspot.get('dependency_count', 0)} dependencies, indicating high coupling",
evidence=[hotspot.get('filepath', '')],
recommendations=[
"Review dependencies and identify opportunities to reduce coupling",
"Consider refactoring to improve modularity",
"Add comprehensive tests before making changes"
],
business_impact="High coupling increases the risk and cost of making changes to this component",
technical_details=hotspot,
affected_components=[hotspot.get('name', '')]
)
self.findings.append(finding)
if 'security_vulnerabilities' in task_result:
for vuln in task_result['security_vulnerabilities']:
finding = Finding(
category='security',
severity=vuln.get('severity', 'high'),
title=vuln.get('title', 'Security vulnerability'),
description=vuln.get('description', ''),
evidence=vuln.get('evidence', []),
recommendations=vuln.get('recommendations', []),
business_impact="Security vulnerabilities pose risk to data confidentiality, integrity, and system availability",
technical_details=vuln,
affected_components=vuln.get('affected_components', [])
)
self.findings.append(finding)
if 'test_coverage_gaps' in task_result:
gaps = task_result['test_coverage_gaps']
if gaps:
finding = Finding(
category='quality',
severity='medium',
title="Test coverage gaps identified",
description=f"Found {len(gaps)} components with insufficient test coverage",
evidence=[g.get('component', '') for g in gaps[:5]],
recommendations=[
"Prioritize adding tests for critical business logic",
"Implement test coverage tracking in CI/CD pipeline",
"Set minimum coverage thresholds for new code"
],
business_impact="Insufficient testing increases the risk of defects reaching production",
technical_details={'gaps': gaps},
affected_components=[g.get('component', '') for g in gaps]
)
self.findings.append(finding)
async def _infer_business_impact(self, technical_finding: Dict[str, Any]) -> str:
impact_prompt = f"""
Technical finding: {json.dumps(technical_finding, indent=2)}
Translate this technical finding into a concise business impact statement that a non-technical executive would understand. Focus on:
- Risk to business operations
- Impact on agility and ability to respond to market changes
- Cost implications
- Customer impact
Provide a 1-2 sentence business impact statement.
"""
response = await self.llm_manager.generate(LLMRequest(
prompt=impact_prompt,
system_prompt="You are a technical translator who explains technical concepts to business stakeholders.",
temperature=0.3,
max_tokens=200,
task_type="business_translation"
))
return response.text.strip()
async def _generate_swot_analysis(self):
strengths = await self._identify_strengths()
weaknesses = await self._identify_weaknesses()
opportunities = await self._identify_opportunities()
threats = await self._identify_threats()
self.swot_items.extend(strengths)
self.swot_items.extend(weaknesses)
self.swot_items.extend(opportunities)
self.swot_items.extend(threats)
async def _identify_strengths(self) -> List[SWOTItem]:
strengths = []
positive_findings = [f for f in self.findings if f.severity == 'low' or 'good' in f.description.lower()]
if len([f for f in self.findings if f.category == 'security' and f.severity == 'high']) == 0:
strengths.append(SWOTItem(
category='strength',
title='Strong Security Posture',
description='No critical security vulnerabilities identified in the analysis',
impact_level='high',
related_findings=[],
stakeholder_relevance=['executives', 'security_team', 'compliance']
))
architecture_findings = [f for f in self.findings if f.category == 'architecture']
if len(architecture_findings) < 5:
strengths.append(SWOTItem(
category='strength',
title='Clean Architecture',
description='The system demonstrates good architectural practices with minimal violations',
impact_level='medium',
related_findings=[f.title for f in architecture_findings],
stakeholder_relevance=['architects', 'developers', 'executives']
))
return strengths
async def _identify_weaknesses(self) -> List[SWOTItem]:
weaknesses = []
high_severity_findings = [f for f in self.findings if f.severity == 'high']
if high_severity_findings:
weaknesses.append(SWOTItem(
category='weakness',
title=f'{len(high_severity_findings)} Critical Issues Require Immediate Attention',
description=f'Found {len(high_severity_findings)} high-severity issues that pose significant risk',
impact_level='high',
related_findings=[f.title for f in high_severity_findings[:5]],
stakeholder_relevance=['executives', 'architects', 'project_managers']
))
coupling_findings = [f for f in self.findings if 'coupling' in f.description.lower() or 'hotspot' in f.title.lower()]
if len(coupling_findings) > 5:
weaknesses.append(SWOTItem(
category='weakness',
title='High Coupling Reduces Agility',
description=f'Identified {len(coupling_findings)} areas of high coupling that make changes risky and expensive',
impact_level='high',
related_findings=[f.title for f in coupling_findings[:3]],
stakeholder_relevance=['executives', 'architects', 'developers']
))
test_findings = [f for f in self.findings if f.category == 'quality' and 'test' in f.description.lower()]
if test_findings:
weaknesses.append(SWOTItem(
category='weakness',
title='Test Coverage Gaps',
description='Insufficient test coverage increases the risk of defects and makes refactoring dangerous',
impact_level='medium',
related_findings=[f.title for f in test_findings],
stakeholder_relevance=['quality_assurance', 'developers', 'project_managers']
))
return weaknesses
async def _identify_opportunities(self) -> List[SWOTItem]:
opportunities = []
refactoring_candidates = [f for f in self.findings if 'refactor' in ' '.join(f.recommendations).lower()]
if refactoring_candidates:
opportunities.append(SWOTItem(
category='opportunity',
title='Strategic Refactoring Can Improve Maintainability',
description=f'Identified {len(refactoring_candidates)} components that would benefit from refactoring to improve long-term maintainability',
impact_level='medium',
related_findings=[f.title for f in refactoring_candidates[:3]],
stakeholder_relevance=['architects', 'technical_leads', 'project_managers']
))
opportunities.append(SWOTItem(
category='opportunity',
title='Automation Can Reduce Manual Analysis Burden',
description='Implementing automated code analysis and monitoring can catch issues earlier and reduce manual review effort',
impact_level='medium',
related_findings=[],
stakeholder_relevance=['developers', 'architects', 'project_managers']
))
return opportunities
async def _identify_threats(self) -> List[SWOTItem]:
threats = []
security_findings = [f for f in self.findings if f.category == 'security' and f.severity in ['high', 'critical']]
if security_findings:
threats.append(SWOTItem(
category='threat',
title='Security Vulnerabilities Pose Risk',
description=f'Identified {len(security_findings)} security vulnerabilities that could be exploited',
impact_level='high',
related_findings=[f.title for f in security_findings],
stakeholder_relevance=['executives', 'security_team', 'compliance']
))
technical_debt_findings = [f for f in self.findings if 'debt' in f.description.lower() or f.category == 'maintainability']
if len(technical_debt_findings) > 10:
threats.append(SWOTItem(
category='threat',
title='Accumulating Technical Debt Threatens Agility',
description='High levels of technical debt will increasingly slow development and increase costs if not addressed',
impact_level='high',
related_findings=[f.title for f in technical_debt_findings[:3]],
stakeholder_relevance=['executives', 'architects', 'project_managers']
))
return threats
async def _generate_executive_summary(self, repository_info: Dict[str, Any]) -> str:
summary_prompt = f"""
Generate an executive summary for a comprehensive code repository analysis.
Repository: {repository_info.get('name', 'Unknown')}
Total Findings: {len(self.findings)}
High Severity Issues: {len([f for f in self.findings if f.severity == 'high'])}
Key SWOT Points:
Strengths: {len([s for s in self.swot_items if s.category == 'strength'])}
Weaknesses: {len([w for w in self.swot_items if w.category == 'weakness'])}
Opportunities: {len([o for o in self.swot_items if o.category == 'opportunity'])}
Threats: {len([t for t in self.swot_items if t.category == 'threat'])}
Top Findings:
{self._format_top_findings()}
Create a concise executive summary (300-400 words) that:
1. Provides an overall assessment of the system's health
2. Highlights the most critical findings and their business implications
3. Summarizes key recommendations
4. Uses business language that non-technical executives can understand
5. Provides a clear call to action
The summary should be professional, balanced, and actionable.
"""
response = await self.llm_manager.generate(LLMRequest(
prompt=summary_prompt,
system_prompt="You are an experienced technology consultant writing for C-level executives. Focus on business impact and strategic implications.",
temperature=0.4,
max_tokens=800,
task_type="executive_summary"
))
return response.text
def _format_top_findings(self) -> str:
sorted_findings = sorted(
self.findings,
key=lambda f: {'critical': 0, 'high': 1, 'medium': 2, 'low': 3}.get(f.severity, 4)
)
top_findings = sorted_findings[:5]
formatted = []
for i, finding in enumerate(top_findings, 1):
formatted.append(f"{i}. [{finding.severity.upper()}] {finding.title}")
formatted.append(f" Impact: {finding.business_impact}")
return '\n'.join(formatted)
async def _generate_technical_analysis(self) -> Dict[str, Any]:
return {
'overview': await self._generate_technical_overview(),
'metrics': self._calculate_technical_metrics(),
'trends': await self._identify_trends(),
'risk_assessment': await self._assess_technical_risks()
}
async def _generate_technical_overview(self) -> str:
overview_prompt = f"""
Generate a technical overview section for a code analysis report.
Total Findings: {len(self.findings)}
Categories: {', '.join(set(f.category for f in self.findings))}
Findings by Category:
{self._format_findings_by_category()}
Create a comprehensive technical overview (400-500 words) that:
1. Describes the overall technical state of the system
2. Discusses architectural patterns and their effectiveness
3. Analyzes code quality and maintainability
4. Addresses security and reliability concerns
5. Provides technical context for the recommendations
Use technical language appropriate for architects and senior developers.
"""
response = await self.llm_manager.generate(LLMRequest(
prompt=overview_prompt,
system_prompt="You are a senior software architect providing technical analysis for other technical professionals.",
temperature=0.4,
max_tokens=1000,
task_type="technical_overview"
))
return response.text
def _format_findings_by_category(self) -> str:
category_counts = {}
for finding in self.findings:
category_counts[finding.category] = category_counts.get(finding.category, 0) + 1
formatted = []
for category, count in sorted(category_counts.items(), key=lambda x: x[1], reverse=True):
formatted.append(f"- {category}: {count} findings")
return '\n'.join(formatted)
def _calculate_technical_metrics(self) -> Dict[str, Any]:
return {
'total_findings': len(self.findings),
'findings_by_severity': self._count_by_severity(),
'findings_by_category': self._count_by_category(),
'affected_components': len(set(
comp for f in self.findings for comp in f.affected_components
)),
'critical_path_components': self._identify_critical_components()
}
def _count_by_severity(self) -> Dict[str, int]:
counts = {}
for finding in self.findings:
counts[finding.severity] = counts.get(finding.severity, 0) + 1
return counts
def _count_by_category(self) -> Dict[str, int]:
counts = {}
for finding in self.findings:
counts[finding.category] = counts.get(finding.category, 0) + 1
return counts
def _identify_critical_components(self) -> List[str]:
component_mentions = {}
for finding in self.findings:
for comp in finding.affected_components:
if comp:
component_mentions[comp] = component_mentions.get(comp, 0) + 1
critical = sorted(
component_mentions.items(),
key=lambda x: x[1],
reverse=True
)[:10]
return [comp for comp, count in critical]
async def _identify_trends(self) -> List[Dict[str, Any]]:
return []
async def _assess_technical_risks(self) -> Dict[str, Any]:
high_risk_findings = [f for f in self.findings if f.severity in ['high', 'critical']]
return {
'overall_risk_level': 'high' if len(high_risk_findings) > 5 else 'medium' if len(high_risk_findings) > 0 else 'low',
'high_risk_areas': list(set(f.category for f in high_risk_findings)),
'risk_factors': [
{
'factor': f.title,
'severity': f.severity,
'mitigation': f.recommendations[0] if f.recommendations else 'No recommendation provided'
}
for f in high_risk_findings[:5]
]
}
async def _generate_architecture_section(self, architecture_data: Dict[str, Any]) -> Dict[str, Any]:
return {
'summary': 'Architecture analysis section',
'patterns_detected': architecture_data.get('detected_patterns', {}),
'quality_attributes': architecture_data.get('quality_attributes', {}),
'recommendations': []
}
async def _generate_code_quality_section(self, code_quality_data: Dict[str, Any]) -> Dict[str, Any]:
return {
'summary': 'Code quality analysis section',
'metrics': code_quality_data.get('metrics', {}),
'issues': code_quality_data.get('issues', []),
'recommendations': []
}
async def _generate_security_section(self, security_data: Dict[str, Any]) -> Dict[str, Any]:
return {
'summary': 'Security analysis section',
'vulnerabilities': security_data.get('vulnerabilities', []),
'compliance': security_data.get('compliance', {}),
'recommendations': []
}
async def _generate_requirements_section(self, requirements_data: Dict[str, Any]) -> Dict[str, Any]:
return {
'summary': 'Requirements analysis section',
'traceability': requirements_data.get('traceability', {}),
'coverage': requirements_data.get('coverage', {}),
'recommendations': []
}
async def _generate_recommendations(self) -> List[Dict[str, Any]]:
all_recommendations = []
for finding in self.findings:
for rec in finding.recommendations:
all_recommendations.append({
'recommendation': rec,
'priority': finding.severity,
'related_finding': finding.title,
'affected_components': finding.affected_components
})
prioritized = sorted(
all_recommendations,
key=lambda x: {'critical': 0, 'high': 1, 'medium': 2, 'low': 3}.get(x['priority'], 4)
)
return prioritized[:20]
async def _generate_visualizations(self, analysis_results: Dict[str, Any]) -> Dict[str, Any]:
return {
'dependency_graph': await self._generate_dependency_graph(analysis_results),
'findings_distribution': self._generate_findings_chart(),
'component_risk_heatmap': self._generate_risk_heatmap(),
'trend_analysis': self._generate_trend_chart()
}
async def _generate_dependency_graph(self, analysis_results: Dict[str, Any]) -> Dict[str, Any]:
architecture_data = analysis_results.get('architecture_analysis', {})
dependencies_data = architecture_data.get('dependencies', {})
nodes = []
edges = []
if 'modules' in architecture_data:
for module in architecture_data['modules'][:20]:
nodes.append({
'id': module.get('module', ''),
'label': module.get('module', ''),
'size': module.get('class_count', 1) * 2,
'color': self._get_node_color(module)
})
for dep in module.get('dependencies', [])[:5]:
edges.append({
'source': module.get('module', ''),
'target': dep.get('name', ''),
'weight': 1
})
return {
'type': 'network_graph',
'nodes': nodes,
'edges': edges,
'layout': 'force_directed'
}
def _get_node_color(self, module: Dict[str, Any]) -> str:
dependency_count = module.get('dependency_count', 0)
if dependency_count > 15:
return '#ff4444'
elif dependency_count > 8:
return '#ffaa44'
else:
return '#44ff44'
def _generate_findings_chart(self) -> Dict[str, Any]:
severity_counts = self._count_by_severity()
return {
'type': 'bar_chart',
'title': 'Findings by Severity',
'data': [
{'label': severity, 'value': count}
for severity, count in severity_counts.items()
],
'colors': {
'critical': '#ff0000',
'high': '#ff6600',
'medium': '#ffaa00',
'low': '#00aa00'
}
}
def _generate_risk_heatmap(self) -> Dict[str, Any]:
component_risks = {}
for finding in self.findings:
severity_score = {'critical': 4, 'high': 3, 'medium': 2, 'low': 1}.get(finding.severity, 0)
for comp in finding.affected_components:
if comp:
component_risks[comp] = component_risks.get(comp, 0) + severity_score
top_risks = sorted(
component_risks.items(),
key=lambda x: x[1],
reverse=True
)[:15]
return {
'type': 'heatmap',
'title': 'Component Risk Heatmap',
'data': [
{'component': comp, 'risk_score': score}
for comp, score in top_risks
]
}
def _generate_trend_chart(self) -> Dict[str, Any]:
return {
'type': 'line_chart',
'title': 'Analysis Trends',
'data': [],
'note': 'Trend data requires historical analysis runs'
}
async def _format_report_for_stakeholders(self, report: Dict[str, Any]) -> Dict[str, Any]:
stakeholder_views = {
'executive': await self._create_executive_view(report),
'architect': await self._create_architect_view(report),
'developer': await self._create_developer_view(report),
'project_manager': await self._create_pm_view(report)
}
report['stakeholder_views'] = stakeholder_views
return report
async def _create_executive_view(self, report: Dict[str, Any]) -> Dict[str, Any]:
return {
'summary': report['executive_summary'],
'swot': report['swot_analysis'],
'key_metrics': {
'total_findings': report['findings_summary']['total_findings'],
'critical_issues': report['findings_summary']['by_severity'].get('critical', 0),
'high_priority_issues': report['findings_summary']['by_severity'].get('high', 0)
},
'top_recommendations': report['recommendations'][:5],
'visualizations': {
'findings_distribution': report['visualizations']['findings_distribution']
}
}
async def _create_architect_view(self, report: Dict[str, Any]) -> Dict[str, Any]:
return {
'technical_analysis': report['technical_analysis'],
'architecture_section': report['detailed_sections']['architecture'],
'all_findings': self.findings,
'recommendations': report['recommendations'],
'visualizations': report['visualizations']
}
async def _create_developer_view(self, report: Dict[str, Any]) -> Dict[str, Any]:
actionable_findings = [
f for f in self.findings
if f.severity in ['high', 'medium'] and f.affected_components
]
return {
'actionable_items': actionable_findings,
'code_quality': report['detailed_sections']['code_quality'],
'security_issues': report['detailed_sections']['security'],
'recommendations_by_component': self._group_recommendations_by_component()
}
async def _create_pm_view(self, report: Dict[str, Any]) -> Dict[str, Any]:
return {
'summary': report['executive_summary'],
'risk_assessment': report['technical_analysis']['risk_assessment'],
'prioritized_recommendations': report['recommendations'],
'effort_estimates': await self._estimate_remediation_effort(),
'timeline_suggestions': await self._suggest_remediation_timeline()
}
def _group_recommendations_by_component(self) -> Dict[str, List[str]]:
grouped = {}
for finding in self.findings:
for comp in finding.affected_components:
if comp:
if comp not in grouped:
grouped[comp] = []
grouped[comp].extend(finding.recommendations)
return grouped
async def _estimate_remediation_effort(self) -> Dict[str, Any]:
return {
'critical_issues': '2-4 weeks',
'high_priority_issues': '4-8 weeks',
'medium_priority_issues': '8-16 weeks',
'note': 'Estimates are approximate and depend on team size and expertise'
}
async def _suggest_remediation_timeline(self) -> List[Dict[str, Any]]:
return [
{
'phase': 'Immediate (Weeks 1-2)',
'focus': 'Address critical security vulnerabilities',
'items': [f.title for f in self.findings if f.severity == 'critical'][:3]
},
{
'phase': 'Short-term (Weeks 3-8)',
'focus': 'Resolve high-priority architectural issues',
'items': [f.title for f in self.findings if f.severity == 'high'][:5]
},
{
'phase': 'Medium-term (Weeks 9-16)',
'focus': 'Improve code quality and test coverage',
'items': [f.title for f in self.findings if f.severity == 'medium'][:5]
}
]
This comprehensive report generation system demonstrates how to synthesize complex technical findings into actionable reports for diverse stakeholders. The system extracts findings from all analysis tasks, categorizes them by severity and type, generates SWOT analysis items, and creates tailored views for executives, architects, developers, and project managers. The LLM integration enables generating natural language summaries that translate technical findings into business language.
Creating the Complete Running Example: E-Commerce Platform Analysis
To demonstrate the entire multi-agent system in action, we now present a complete running example that analyzes a hypothetical e-commerce platform. This example brings together all the components we have discussed, showing how they work together to produce comprehensive analysis.
Here is the complete integration example:
import asyncio
from typing import Dict, Any
import os
class ECommerceAnalysisExample:
def __init__(self):
self.config = self._load_configuration()
self.system = None
def _load_configuration(self) -> Dict[str, Any]:
return {
'repository': {
'name': 'ECommerce Platform',
'path': '/path/to/ecommerce/repo',
'languages': ['python', 'javascript', 'java'],
'scope': 'Full system analysis'
},
'llm_providers': {
'openai': {
'enabled': True,
'api_key': os.getenv('OPENAI_API_KEY', ''),
'model': 'gpt-4',
'default': False
},
'anthropic': {
'enabled': True,
'api_key': os.getenv('ANTHROPIC_API_KEY', ''),
'model': 'claude-3-5-sonnet-20241022',
'default': True
},
'local_llama': {
'enabled': True,
'base_url': 'http://localhost:8080',
'model': 'codellama-13b',
'default': False
},
'task_routing': {
'code_analysis': 'local_llama',
'reasoning': 'anthropic',
'synthesis': 'anthropic',
'executive_summary': 'openai'
}
},
'knowledge_graph': {
'uri': 'bolt://localhost:7687',
'user': 'neo4j',
'password': os.getenv('NEO4J_PASSWORD', 'password')
},
'vector_database': {
'persist_directory': './chroma_db',
'embedding_model': 'all-MiniLM-L6-v2'
},
'analysis_scope': {
'include_architecture': True,
'include_security': True,
'include_code_quality': True,
'include_requirements': True,
'include_domain_model': True,
'include_testing': True,
'include_operations': True
}
}
async def run_complete_analysis(self) -> Dict[str, Any]:
print("Initializing Multi-Agent Analysis System...")
await self._initialize_system()
print("\nPhase 1: Repository Ingestion and Parsing...")
await self._ingest_repository()
print("\nPhase 2: Knowledge Graph Construction...")
await self._build_knowledge_graph()
print("\nPhase 3: Vector Database Population...")
await self._populate_vector_database()
print("\nPhase 4: Multi-Agent Analysis Execution...")
analysis_results = await self._execute_analysis()
print("\nPhase 5: Report Generation...")
final_report = await self._generate_report(analysis_results)
print("\nPhase 6: Visualization Generation...")
await self._generate_visualizations(final_report)
print("\nAnalysis Complete!")
return final_report
async def _initialize_system(self):
from neo4j import GraphDatabase
llm_manager = LLMManager(self.config['llm_providers'])
kg_manager = KnowledgeGraphManager(
uri=self.config['knowledge_graph']['uri'],
user=self.config['knowledge_graph']['user'],
password=self.config['knowledge_graph']['password']
)
vector_db_manager = VectorDatabaseManager(
persist_directory=self.config['vector_database']['persist_directory'],
embedding_model_name=self.config['vector_database']['embedding_model']
)
orchestrator = AgentOrchestrator(kg_manager, vector_db_manager, llm_manager)
orchestrator.register_agent('architecture', ArchitectureAnalysisAgent())
orchestrator.register_agent('domain', DomainAnalysisAgent())
orchestrator.register_agent('code', CodeAnalysisAgent())
orchestrator.register_agent('security', SecurityAnalysisAgent())
orchestrator.register_agent('testing', TestingAnalysisAgent())
orchestrator.register_agent('requirements', RequirementsAnalysisAgent())
orchestrator.register_agent('quality', QualityAnalysisAgent())
orchestrator.register_agent('operations', OperationsAnalysisAgent())
orchestrator.register_agent('repository', RepositoryAnalysisAgent())
orchestrator.register_agent('business', BusinessAnalysisAgent())
orchestrator.register_agent('synthesis', SynthesisAgent())
orchestrator.register_agent('report', ReportGenerator(llm_manager))
self.system = {
'llm_manager': llm_manager,
'kg_manager': kg_manager,
'vector_db_manager': vector_db_manager,
'orchestrator': orchestrator
}
async def _ingest_repository(self):
repo_path = self.config['repository']['path']
python_files = self._find_files_by_extension(repo_path, '.py')
javascript_files = self._find_files_by_extension(repo_path, '.js')
java_files = self._find_files_by_extension(repo_path, '.java')
print(f"Found {len(python_files)} Python files")
print(f"Found {len(javascript_files)} JavaScript files")
print(f"Found {len(java_files)} Java files")
python_analyzer = PythonCodeAnalyzer()
js_analyzer = JavaScriptCodeAnalyzer()
self.parsed_results = {
'python': [],
'javascript': [],
'java': []
}
for py_file in python_files[:50]:
try:
result = python_analyzer.analyze_file(py_file)
if result['success']:
self.parsed_results['python'].append(result)
except Exception as e:
print(f"Error parsing {py_file}: {e}")
for js_file in javascript_files[:50]:
try:
result = js_analyzer.analyze_file(js_file)
if result['success']:
self.parsed_results['javascript'].append(result)
except Exception as e:
print(f"Error parsing {js_file}: {e}")
print(f"Successfully parsed {len(self.parsed_results['python'])} Python files")
print(f"Successfully parsed {len(self.parsed_results['javascript'])} JavaScript files")
def _find_files_by_extension(self, root_path: str, extension: str) -> list:
files = []
for dirpath, dirnames, filenames in os.walk(root_path):
if 'node_modules' in dirpath or '.git' in dirpath or 'venv' in dirpath:
continue
for filename in filenames:
if filename.endswith(extension):
files.append(os.path.join(dirpath, filename))
return files
async def _build_knowledge_graph(self):
kg_manager = self.system['kg_manager']
total_entities = 0
total_relationships = 0
for language, results in self.parsed_results.items():
for result in results:
kg_manager.add_code_entities(result)
total_entities += len(result['entities'])
total_relationships += len(result['relationships'])
print(f"Added {total_entities} entities to knowledge graph")
print(f"Added {total_relationships} relationships to knowledge graph")
async def _populate_vector_database(self):
vector_db = self.system['vector_db_manager']
code_chunks = []
for language, results in self.parsed_results.items():
for result in results:
for entity_id, entity_data in result['entities'].items():
if entity_data['type'] in ['class', 'function', 'method']:
chunk = {
'text': f"{entity_data['name']}\n{entity_data.get('docstring', '')}",
'filepath': entity_data['filepath'],
'start_line': entity_data.get('line_number', 0),
'end_line': entity_data.get('line_number', 0),
'entity_type': entity_data['type'],
'entity_name': entity_data['name'],
'language': language
}
code_chunks.append(chunk)
vector_db.add_code_chunks(code_chunks)
print(f"Added {len(code_chunks)} code chunks to vector database")
doc_chunks = self._load_documentation_chunks()
if doc_chunks:
vector_db.add_documentation_chunks(doc_chunks)
print(f"Added {len(doc_chunks)} documentation chunks to vector database")
def _load_documentation_chunks(self) -> list:
return [
{
'text': 'The e-commerce platform uses a microservices architecture with separate services for user management, product catalog, shopping cart, order processing, and payment.',
'document_type': 'architecture',
'document_path': 'docs/architecture.md',
'section': 'Architecture Overview',
'subsection': 'Service Decomposition'
},
{
'text': 'User authentication is handled by the Auth Service using JWT tokens. All API requests must include a valid JWT in the Authorization header.',
'document_type': 'architecture',
'document_path': 'docs/architecture.md',
'section': 'Security',
'subsection': 'Authentication'
},
{
'text': 'The system must support at least 10,000 concurrent users during peak shopping periods. Response time for product searches must be under 200ms.',
'document_type': 'requirements',
'document_path': 'docs/requirements.md',
'section': 'Non-Functional Requirements',
'subsection': 'Performance'
}
]
async def _execute_analysis(self) -> Dict[str, Any]:
orchestrator = self.system['orchestrator']
analysis_request = {
'type': 'full_system_analysis',
'parameters': {
'repository_name': self.config['repository']['name'],
'scope': self.config['analysis_scope']
}
}
results = await orchestrator.execute_analysis(analysis_request)
return results
async def _generate_report(self, analysis_results: Dict[str, Any]) -> Dict[str, Any]:
report_generator = self.system['orchestrator'].agents['report']
report = await report_generator.generate_comprehensive_report(
analysis_results,
self.config['repository']
)
return report
async def _generate_visualizations(self, report: Dict[str, Any]):
print("Generating dependency graph visualization...")
print("Generating findings distribution chart...")
print("Generating risk heatmap...")
print("Visualizations generated successfully")
async def demonstrate_specific_queries(self):
print("\n=== Demonstrating Specific Analysis Queries ===\n")
await self._demo_requirement_traceability()
await self._demo_security_analysis()
await self._demo_architecture_pattern_detection()
await self._demo_code_quality_assessment()
async def _demo_requirement_traceability(self):
print("Query: Trace requirement 'User Authentication' through the system")
orchestrator = self.system['orchestrator']
trace_request = {
'type': 'requirement_traceability',
'requirement_id': 'REQ-AUTH-001'
}
result = await orchestrator.execute_analysis(trace_request)
print(f"Found {len(result.get('design_elements', []))} design elements")
print(f"Found {len(result.get('code_components', []))} code components")
print(f"Found {len(result.get('tests', []))} related tests")
async def _demo_security_analysis(self):
print("\nQuery: Analyze security posture of authentication subsystem")
orchestrator = self.system['orchestrator']
security_request = {
'type': 'security_analysis',
'subsystem': 'authentication'
}
result = await orchestrator.execute_analysis(security_request)
print(f"Identified {len(result.get('vulnerabilities', []))} potential vulnerabilities")
print(f"Security requirements coverage: {result.get('requirements_coverage', 0)}%")
async def _demo_architecture_pattern_detection(self):
print("\nQuery: Identify architectural patterns in use")
orchestrator = self.system['orchestrator']
arch_request = {
'type': 'architecture_analysis'
}
result = await orchestrator.execute_analysis(arch_request)
patterns = result.get('detected_patterns', {})
print(f"Detected patterns: {', '.join(patterns.keys())}")
async def _demo_code_quality_assessment(self):
print("\nQuery: Assess overall code quality")
kg_manager = self.system['kg_manager']
hotspots = kg_manager.find_architectural_hotspots(min_dependencies=5)
print(f"Found {len(hotspots)} architectural hotspots")
if hotspots:
print(f"Top hotspot: {hotspots[0]['name']} with {hotspots[0]['dependency_count']} dependencies")
async def main():
example = ECommerceAnalysisExample()
print("="*80)
print("MULTI-AGENT CODE REPOSITORY ANALYSIS SYSTEM")
print("E-Commerce Platform Analysis Example")
print("="*80)
try:
final_report = await example.run_complete_analysis()
print("\n" + "="*80)
print("ANALYSIS SUMMARY")
print("="*80)
print(f"\nTotal Findings: {final_report['findings_summary']['total_findings']}")
print(f"Critical Issues: {final_report['findings_summary']['by_severity'].get('critical', 0)}")
print(f"High Priority Issues: {final_report['findings_summary']['by_severity'].get('high', 0)}")
print("\n" + "="*80)
print("SWOT ANALYSIS SUMMARY")
print("="*80)
swot = final_report['swot_analysis']
print(f"\nStrengths: {len(swot['strengths'])}")
for strength in swot['strengths'][:3]:
print(f" - {strength.title}")
print(f"\nWeaknesses: {len(swot['weaknesses'])}")
for weakness in swot['weaknesses'][:3]:
print(f" - {weakness.title}")
print(f"\nOpportunities: {len(swot['opportunities'])}")
for opportunity in swot['opportunities'][:3]:
print(f" - {opportunity.title}")
print(f"\nThreats: {len(swot['threats'])}")
for threat in swot['threats'][:3]:
print(f" - {threat.title}")
print("\n" + "="*80)
print("EXECUTIVE SUMMARY")
print("="*80)
print(f"\n{final_report['executive_summary']}")
print("\n" + "="*80)
print("TOP RECOMMENDATIONS")
print("="*80)
for i, rec in enumerate(final_report['recommendations'][:5], 1):
print(f"\n{i}. [{rec['priority'].upper()}] {rec['recommendation']}")
print(f" Related to: {rec['related_finding']}")
await example.demonstrate_specific_queries()
print("\n" + "="*80)
print("ANALYSIS COMPLETE")
print("="*80)
print("\nFull report available in structured format")
print("Stakeholder-specific views generated for:")
print(" - Executives")
print(" - Architects")
print(" - Developers")
print(" - Project Managers")
except Exception as e:
print(f"\nError during analysis: {e}")
import traceback
traceback.print_exc()
finally:
if example.system:
await example.system['llm_manager'].close_all()
example.system['kg_manager'].close()
if __name__ == "__main__":
asyncio.run(main())
This complete running example demonstrates the entire multi-agent system analyzing an e-commerce platform. The example shows how all components work together through a realistic workflow that ingests code, builds knowledge representations, executes multi-agent analysis, and generates comprehensive reports. The system handles multiple programming languages, coordinates specialized agents, synthesizes findings into SWOT analysis, and produces stakeholder-specific views.
The example illustrates key capabilities including requirement traceability that traces business requirements through design, implementation, and testing; security analysis that examines specific subsystems for vulnerabilities; architecture pattern detection that identifies the architectural styles in use; and code quality assessment that finds hotspots and technical debt. The multi-stakeholder report generation produces tailored views for executives who need business impact summaries, architects who need technical depth, developers who need actionable items, and project managers who need effort estimates and timelines.
Deployment Considerations and Operational Best Practices
Deploying the multi-agent analysis system in production environments requires careful attention to scalability, reliability, and maintainability. The system should support both on-demand analysis triggered by user requests and scheduled periodic analysis that tracks system evolution over time. For large repositories, analysis can take hours, so the system needs robust job management with progress tracking, error recovery, and result persistence.
The knowledge graph and vector database require sufficient storage and memory resources. A large enterprise repository might generate millions of nodes and edges in the knowledge graph. The vector database might store millions of embeddings. Proper indexing and query optimization ensure acceptable performance. Regular maintenance tasks like graph compaction and index rebuilding keep the system performant as data accumulates.
LLM costs can become significant with heavy usage, especially when using premium cloud models. The task routing configuration should balance cost and quality by using expensive models only for tasks that benefit from their advanced capabilities while routing routine tasks to cheaper or local models. Usage tracking and budget alerts prevent unexpected cost overruns. Caching LLM responses for identical queries reduces redundant API calls.
Security considerations include protecting API keys for cloud LLM providers, securing access to the knowledge graph and vector database, and ensuring that analysis reports containing potentially sensitive information about system weaknesses are properly access-controlled. The system should support role-based access control so that different stakeholders see only the information appropriate for their roles.
Continuous improvement of the system involves collecting feedback on analysis quality, tracking which findings lead to actual code improvements, and refining agent prompts and reasoning patterns based on real-world performance. The multi-agent architecture facilitates incremental improvement because individual agents can be enhanced independently without disrupting the entire system.
The complete system demonstrates how to combine traditional software engineering techniques like parsing and knowledge graphs with modern AI capabilities like large language models and vector embeddings to create powerful analytical tools that serve diverse stakeholders from executives to developers. The multi-agent architecture provides flexibility, scalability, and the ability to leverage different AI models for different tasks while maintaining coherent analysis across the entire system.
CHAPTER 7: INTEGRATING ARCHITECTURE AND CODE ANALYSIS TOOLS
Leveraging Existing Static Analysis Tools
The multi-agent system achieves significantly greater analytical depth by integrating established static analysis tools rather than attempting to reimplement their functionality. Tools like SonarQube, ESLint, Pylint, PMD, and SpotBugs have been refined over years to detect specific code quality issues, security vulnerabilities, and style violations. These tools employ sophisticated pattern matching, data flow analysis, and control flow analysis that would be expensive and error-prone to recreate. The multi-agent system orchestrates these tools, aggregates their outputs, and uses language models to interpret findings in context and synthesize insights across multiple tools.
The integration architecture treats static analysis tools as specialized information sources that feed into the knowledge graph and inform agent reasoning. When analyzing a Python codebase, the system invokes Pylint to check code quality, Bandit to detect security issues, and mypy to verify type correctness. Each tool produces findings in its own format. The integration layer normalizes these diverse formats into a unified representation that the knowledge graph can store and agents can query. This normalization enables cross-tool correlation where findings from different tools that affect the same code location are grouped together, providing a more complete picture of issues in specific components.
The language model agents add interpretive value on top of raw tool outputs. Static analysis tools report that a particular function has cyclomatic complexity of thirty-five, which exceeds recommended thresholds. The tool identifies the problem but provides limited context about why this matters or what to do about it. An LLM-powered agent can examine the function in context, understand its business purpose from documentation and naming, assess whether the complexity is inherent to the problem domain or represents poor design, and generate specific recommendations for refactoring that preserve functionality while reducing complexity. This combination of precise automated detection with contextual interpretation provides analysis that neither approach could achieve alone.
Here is an implementation of the static analysis tool integration framework:
import subprocess
import json
import xml.etree.ElementTree as ET
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from abc import ABC, abstractmethod
import re
@dataclass
class ToolFinding:
tool_name: str
rule_id: str
severity: str
category: str
message: str
filepath: str
line_number: int
column_number: Optional[int]
code_snippet: Optional[str]
suggestion: Optional[str]
raw_data: Dict[str, Any]
class StaticAnalysisTool(ABC):
@abstractmethod
def analyze(self, target_path: str, config: Optional[Dict[str, Any]] = None) -> List[ToolFinding]:
pass
@abstractmethod
def get_tool_name(self) -> str:
pass
@abstractmethod
def is_available(self) -> bool:
pass
class PylintAnalyzer(StaticAnalysisTool):
def __init__(self):
self.tool_name = "pylint"
def get_tool_name(self) -> str:
return self.tool_name
def is_available(self) -> bool:
try:
result = subprocess.run(
['pylint', '--version'],
capture_output=True,
timeout=5
)
return result.returncode == 0
except (FileNotFoundError, subprocess.TimeoutExpired):
return False
def analyze(self, target_path: str, config: Optional[Dict[str, Any]] = None) -> List[ToolFinding]:
if not self.is_available():
raise Exception("Pylint is not installed or not available in PATH")
cmd = [
'pylint',
target_path,
'--output-format=json',
'--disable=C0111' # Disable missing docstring warnings for brevity
]
if config and 'rcfile' in config:
cmd.extend(['--rcfile', config['rcfile']])
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=300
)
if result.stdout:
pylint_output = json.loads(result.stdout)
return self._parse_pylint_output(pylint_output)
else:
return []
except subprocess.TimeoutExpired:
raise Exception(f"Pylint analysis timed out for {target_path}")
except json.JSONDecodeError as e:
raise Exception(f"Failed to parse Pylint output: {e}")
def _parse_pylint_output(self, pylint_data: List[Dict[str, Any]]) -> List[ToolFinding]:
findings = []
severity_map = {
'convention': 'low',
'refactor': 'medium',
'warning': 'medium',
'error': 'high',
'fatal': 'critical'
}
for item in pylint_data:
finding = ToolFinding(
tool_name=self.tool_name,
rule_id=item.get('message-id', ''),
severity=severity_map.get(item.get('type', 'warning').lower(), 'medium'),
category='code_quality',
message=item.get('message', ''),
filepath=item.get('path', ''),
line_number=item.get('line', 0),
column_number=item.get('column', 0),
code_snippet=None,
suggestion=None,
raw_data=item
)
findings.append(finding)
return findings
class BanditAnalyzer(StaticAnalysisTool):
def __init__(self):
self.tool_name = "bandit"
def get_tool_name(self) -> str:
return self.tool_name
def is_available(self) -> bool:
try:
result = subprocess.run(
['bandit', '--version'],
capture_output=True,
timeout=5
)
return result.returncode == 0
except (FileNotFoundError, subprocess.TimeoutExpired):
return False
def analyze(self, target_path: str, config: Optional[Dict[str, Any]] = None) -> List[ToolFinding]:
if not self.is_available():
raise Exception("Bandit is not installed or not available in PATH")
cmd = [
'bandit',
'-r',
target_path,
'-f', 'json'
]
if config and 'config_file' in config:
cmd.extend(['-c', config['config_file']])
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=300
)
if result.stdout:
bandit_output = json.loads(result.stdout)
return self._parse_bandit_output(bandit_output)
else:
return []
except subprocess.TimeoutExpired:
raise Exception(f"Bandit analysis timed out for {target_path}")
except json.JSONDecodeError as e:
raise Exception(f"Failed to parse Bandit output: {e}")
def _parse_bandit_output(self, bandit_data: Dict[str, Any]) -> List[ToolFinding]:
findings = []
severity_map = {
'LOW': 'low',
'MEDIUM': 'medium',
'HIGH': 'high'
}
for result in bandit_data.get('results', []):
finding = ToolFinding(
tool_name=self.tool_name,
rule_id=result.get('test_id', ''),
severity=severity_map.get(result.get('issue_severity', 'MEDIUM'), 'medium'),
category='security',
message=result.get('issue_text', ''),
filepath=result.get('filename', ''),
line_number=result.get('line_number', 0),
column_number=None,
code_snippet=result.get('code', ''),
suggestion=result.get('more_info', ''),
raw_data=result
)
findings.append(finding)
return findings
class ESLintAnalyzer(StaticAnalysisTool):
def __init__(self):
self.tool_name = "eslint"
def get_tool_name(self) -> str:
return self.tool_name
def is_available(self) -> bool:
try:
result = subprocess.run(
['eslint', '--version'],
capture_output=True,
timeout=5
)
return result.returncode == 0
except (FileNotFoundError, subprocess.TimeoutExpired):
return False
def analyze(self, target_path: str, config: Optional[Dict[str, Any]] = None) -> List[ToolFinding]:
if not self.is_available():
raise Exception("ESLint is not installed or not available in PATH")
cmd = [
'eslint',
target_path,
'--format', 'json'
]
if config and 'config_file' in config:
cmd.extend(['--config', config['config_file']])
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=300
)
if result.stdout:
eslint_output = json.loads(result.stdout)
return self._parse_eslint_output(eslint_output)
else:
return []
except subprocess.TimeoutExpired:
raise Exception(f"ESLint analysis timed out for {target_path}")
except json.JSONDecodeError as e:
raise Exception(f"Failed to parse ESLint output: {e}")
def _parse_eslint_output(self, eslint_data: List[Dict[str, Any]]) -> List[ToolFinding]:
findings = []
severity_map = {
0: 'info',
1: 'medium',
2: 'high'
}
for file_result in eslint_data:
filepath = file_result.get('filePath', '')
for message in file_result.get('messages', []):
finding = ToolFinding(
tool_name=self.tool_name,
rule_id=message.get('ruleId', ''),
severity=severity_map.get(message.get('severity', 1), 'medium'),
category='code_quality',
message=message.get('message', ''),
filepath=filepath,
line_number=message.get('line', 0),
column_number=message.get('column', 0),
code_snippet=None,
suggestion=message.get('fix', {}).get('text') if message.get('fix') else None,
raw_data=message
)
findings.append(finding)
return findings
class SonarQubeAnalyzer(StaticAnalysisTool):
def __init__(self, server_url: str, token: str):
self.tool_name = "sonarqube"
self.server_url = server_url
self.token = token
def get_tool_name(self) -> str:
return self.tool_name
def is_available(self) -> bool:
try:
import requests
response = requests.get(
f"{self.server_url}/api/system/status",
timeout=5
)
return response.status_code == 200
except:
return False
def analyze(self, target_path: str, config: Optional[Dict[str, Any]] = None) -> List[ToolFinding]:
project_key = config.get('project_key', 'default-project') if config else 'default-project'
self._run_sonar_scanner(target_path, project_key)
findings = self._fetch_issues(project_key)
return findings
def _run_sonar_scanner(self, target_path: str, project_key: str):
cmd = [
'sonar-scanner',
f'-Dsonar.projectKey={project_key}',
f'-Dsonar.sources={target_path}',
f'-Dsonar.host.url={self.server_url}',
f'-Dsonar.login={self.token}'
]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=600
)
if result.returncode != 0:
raise Exception(f"SonarQube scanner failed: {result.stderr}")
except subprocess.TimeoutExpired:
raise Exception("SonarQube scanner timed out")
def _fetch_issues(self, project_key: str) -> List[ToolFinding]:
import requests
findings = []
page = 1
page_size = 500
while True:
response = requests.get(
f"{self.server_url}/api/issues/search",
params={
'componentKeys': project_key,
'p': page,
'ps': page_size
},
headers={'Authorization': f'Bearer {self.token}'}
)
if response.status_code != 200:
break
data = response.json()
issues = data.get('issues', [])
for issue in issues:
finding = ToolFinding(
tool_name=self.tool_name,
rule_id=issue.get('rule', ''),
severity=issue.get('severity', 'MEDIUM').lower(),
category=issue.get('type', 'CODE_SMELL').lower(),
message=issue.get('message', ''),
filepath=issue.get('component', ''),
line_number=issue.get('line', 0),
column_number=None,
code_snippet=None,
suggestion=None,
raw_data=issue
)
findings.append(finding)
if len(issues) < page_size:
break
page += 1
return findings
class ToolIntegrationManager:
def __init__(self, knowledge_graph_manager, llm_manager):
self.kg_manager = knowledge_graph_manager
self.llm_manager = llm_manager
self.tools: Dict[str, StaticAnalysisTool] = {}
self.findings_cache: Dict[str, List[ToolFinding]] = {}
def register_tool(self, tool: StaticAnalysisTool):
self.tools[tool.get_tool_name()] = tool
async def run_all_tools(self, target_path: str, language: str) -> Dict[str, List[ToolFinding]]:
results = {}
tool_mapping = {
'python': ['pylint', 'bandit'],
'javascript': ['eslint'],
'java': ['pmd', 'spotbugs']
}
applicable_tools = tool_mapping.get(language, [])
for tool_name in applicable_tools:
if tool_name in self.tools:
tool = self.tools[tool_name]
if tool.is_available():
try:
findings = tool.analyze(target_path)
results[tool_name] = findings
self.findings_cache[f"{tool_name}:{target_path}"] = findings
print(f"{tool_name}: Found {len(findings)} issues")
except Exception as e:
print(f"Error running {tool_name}: {e}")
results[tool_name] = []
else:
print(f"{tool_name} is not available, skipping")
return results
async def integrate_findings_into_knowledge_graph(self, findings_by_tool: Dict[str, List[ToolFinding]]):
for tool_name, findings in findings_by_tool.items():
for finding in findings:
await self._add_finding_to_graph(finding)
await self._correlate_findings()
async def _add_finding_to_graph(self, finding: ToolFinding):
pass
async def _correlate_findings(self):
location_groups = {}
for cache_key, findings in self.findings_cache.items():
for finding in findings:
location_key = f"{finding.filepath}:{finding.line_number}"
if location_key not in location_groups:
location_groups[location_key] = []
location_groups[location_key].append(finding)
correlated_issues = []
for location, findings in location_groups.items():
if len(findings) > 1:
correlated_issues.append({
'location': location,
'findings': findings,
'severity': max(f.severity for f in findings),
'tools': [f.tool_name for f in findings]
})
print(f"Found {len(correlated_issues)} locations with multiple tool findings")
return correlated_issues
async def interpret_findings_with_llm(self, findings: List[ToolFinding], context: Dict[str, Any]) -> Dict[str, Any]:
if not findings:
return {'interpretation': 'No findings to interpret', 'recommendations': []}
findings_summary = self._format_findings_for_llm(findings[:10])
interpretation_prompt = f"""
Analyze the following static analysis tool findings and provide interpretation in context:
{findings_summary}
Context:
- Component: {context.get('component_name', 'Unknown')}
- Purpose: {context.get('component_purpose', 'Unknown')}
- Language: {context.get('language', 'Unknown')}
Provide:
1. An overall assessment of the code quality based on these findings
2. Identification of the most critical issues that should be addressed first
3. Patterns or systemic issues indicated by multiple findings
4. Specific, actionable recommendations for improvement
5. Assessment of whether the issues are superficial or indicate deeper design problems
Be concise but thorough. Focus on actionable insights.
"""
response = await self.llm_manager.generate(LLMRequest(
prompt=interpretation_prompt,
system_prompt="You are an expert code reviewer analyzing static analysis results. Provide practical, actionable insights.",
temperature=0.3,
max_tokens=1500,
task_type="code_interpretation"
))
recommendations = await self._extract_recommendations(response.text)
return {
'interpretation': response.text,
'recommendations': recommendations,
'critical_findings': [f for f in findings if f.severity in ['high', 'critical']],
'finding_count': len(findings)
}
def _format_findings_for_llm(self, findings: List[ToolFinding]) -> str:
formatted = []
for i, finding in enumerate(findings, 1):
formatted.append(f"\n{i}. [{finding.tool_name}] {finding.severity.upper()}")
formatted.append(f" Rule: {finding.rule_id}")
formatted.append(f" Location: {finding.filepath}:{finding.line_number}")
formatted.append(f" Message: {finding.message}")
if finding.code_snippet:
formatted.append(f" Code: {finding.code_snippet[:100]}")
return '\n'.join(formatted)
async def _extract_recommendations(self, interpretation_text: str) -> List[str]:
lines = interpretation_text.split('\n')
recommendations = []
in_recommendations = False
for line in lines:
if 'recommendation' in line.lower():
in_recommendations = True
continue
if in_recommendations:
line = line.strip()
if line and (line[0].isdigit() or line.startswith('-') or line.startswith('•')):
clean_line = re.sub(r'^[\d\-•.\s]+', '', line)
if clean_line:
recommendations.append(clean_line)
return recommendations[:10]
async def generate_quality_report(self, findings_by_tool: Dict[str, List[ToolFinding]]) -> Dict[str, Any]:
total_findings = sum(len(findings) for findings in findings_by_tool.values())
severity_distribution = {
'critical': 0,
'high': 0,
'medium': 0,
'low': 0,
'info': 0
}
category_distribution = {}
for findings in findings_by_tool.values():
for finding in findings:
severity_distribution[finding.severity] = severity_distribution.get(finding.severity, 0) + 1
category_distribution[finding.category] = category_distribution.get(finding.category, 0) + 1
quality_score = self._calculate_quality_score(severity_distribution, total_findings)
return {
'total_findings': total_findings,
'severity_distribution': severity_distribution,
'category_distribution': category_distribution,
'quality_score': quality_score,
'tools_used': list(findings_by_tool.keys()),
'critical_issues': severity_distribution.get('critical', 0) + severity_distribution.get('high', 0)
}
def _calculate_quality_score(self, severity_dist: Dict[str, int], total: int) -> float:
if total == 0:
return 100.0
weights = {
'critical': 10,
'high': 5,
'medium': 2,
'low': 1,
'info': 0.5
}
weighted_sum = sum(
severity_dist.get(severity, 0) * weight
for severity, weight in weights.items()
)
max_score = 100
penalty = min(weighted_sum, max_score)
score = max(0, max_score - penalty)
return round(score, 2)
This comprehensive tool integration framework demonstrates how to orchestrate multiple static analysis tools, normalize their outputs, correlate findings across tools, and use language models to interpret results in context. The framework supports Python tools like Pylint and Bandit, JavaScript tools like ESLint, and enterprise platforms like SonarQube. The LLM integration adds interpretive value by analyzing patterns across findings and generating actionable recommendations.
Integrating Architecture Analysis Tools and Frameworks
Beyond code-level static analysis, architecture analysis tools provide insights into system-level structure, dependencies, and design quality. Tools like ArchUnit for architecture testing, Dependency Cruiser for JavaScript dependency analysis, and jdepend for Java package dependencies detect architectural violations and structural problems that code-level tools miss. The multi-agent system integrates these tools to build a comprehensive understanding of system architecture.
Architecture analysis tools typically examine dependencies between modules, packages, or services. They can detect circular dependencies, layering violations where lower-level components depend on higher-level components, and violations of architectural rules like "presentation layer components must not directly access data access layer components." These structural issues indicate architectural erosion where the implemented architecture diverges from the intended design.
The integration of architecture tools with the knowledge graph creates powerful analytical capabilities. The knowledge graph already contains code-level entities and relationships from parsing. Adding architecture-level findings enriches the graph with higher-level structural information. An agent can query the graph to trace a path from a business requirement through architectural components to specific code classes and functions, providing end-to-end traceability that spans multiple abstraction levels.
Here is an implementation integrating architecture analysis tools:
import subprocess
import json
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
@dataclass
class ArchitectureFinding:
tool_name: str
finding_type: str
severity: str
description: str
source_component: str
target_component: Optional[str]
violation_details: Dict[str, Any]
architectural_rule: str
class ArchitectureAnalysisTool(ABC):
@abstractmethod
def analyze_architecture(self, target_path: str, config: Optional[Dict[str, Any]] = None) -> List[ArchitectureFinding]:
pass
@abstractmethod
def get_tool_name(self) -> str:
pass
class DependencyCruiserAnalyzer(ArchitectureAnalysisTool):
def __init__(self):
self.tool_name = "dependency-cruiser"
def get_tool_name(self) -> str:
return self.tool_name
def analyze_architecture(self, target_path: str, config: Optional[Dict[str, Any]] = None) -> List[ArchitectureFinding]:
cmd = [
'depcruise',
'--output-type', 'json',
target_path
]
if config and 'config_file' in config:
cmd.extend(['--config', config['config_file']])
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=300
)
if result.stdout:
depcruise_output = json.loads(result.stdout)
return self._parse_dependency_cruiser_output(depcruise_output)
else:
return []
except subprocess.TimeoutExpired:
raise Exception("Dependency Cruiser analysis timed out")
except json.JSONDecodeError as e:
raise Exception(f"Failed to parse Dependency Cruiser output: {e}")
def _parse_dependency_cruiser_output(self, data: Dict[str, Any]) -> List[ArchitectureFinding]:
findings = []
violations = data.get('summary', {}).get('violations', [])
for violation in violations:
finding = ArchitectureFinding(
tool_name=self.tool_name,
finding_type='dependency_violation',
severity=violation.get('severity', 'warn'),
description=violation.get('rule', {}).get('name', 'Unknown rule'),
source_component=violation.get('from', ''),
target_component=violation.get('to', ''),
violation_details=violation,
architectural_rule=violation.get('rule', {}).get('name', '')
)
findings.append(finding)
modules = data.get('modules', [])
circular_deps = self._detect_circular_dependencies(modules)
for cycle in circular_deps:
finding = ArchitectureFinding(
tool_name=self.tool_name,
finding_type='circular_dependency',
severity='high',
description=f"Circular dependency detected: {' -> '.join(cycle)}",
source_component=cycle[0],
target_component=cycle[-1],
violation_details={'cycle': cycle},
architectural_rule='no_circular_dependencies'
)
findings.append(finding)
return findings
def _detect_circular_dependencies(self, modules: List[Dict[str, Any]]) -> List[List[str]]:
cycles = []
return cycles
class MavenDependencyAnalyzer(ArchitectureAnalysisTool):
def __init__(self):
self.tool_name = "maven-dependency"
def get_tool_name(self) -> str:
return self.tool_name
def analyze_architecture(self, target_path: str, config: Optional[Dict[str, Any]] = None) -> List[ArchitectureFinding]:
cmd = [
'mvn',
'dependency:tree',
'-DoutputType=dot',
'-DoutputFile=dependency-tree.dot',
'-f', target_path
]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=300,
cwd=target_path
)
if result.returncode == 0:
return self._analyze_dependency_tree(f"{target_path}/dependency-tree.dot")
else:
return []
except subprocess.TimeoutExpired:
raise Exception("Maven dependency analysis timed out")
def _analyze_dependency_tree(self, dot_file_path: str) -> List[ArchitectureFinding]:
findings = []
try:
with open(dot_file_path, 'r') as f:
dot_content = f.read()
findings.extend(self._detect_version_conflicts(dot_content))
findings.extend(self._detect_transitive_dependency_issues(dot_content))
except FileNotFoundError:
pass
return findings
def _detect_version_conflicts(self, dot_content: str) -> List[ArchitectureFinding]:
findings = []
return findings
def _detect_transitive_dependency_issues(self, dot_content: str) -> List[ArchitectureFinding]:
findings = []
return findings
class ArchitectureIntegrationAgent:
def __init__(self, knowledge_graph_manager, llm_manager):
self.kg_manager = knowledge_graph_manager
self.llm_manager = llm_manager
self.arch_tools: Dict[str, ArchitectureAnalysisTool] = {}
def register_architecture_tool(self, tool: ArchitectureAnalysisTool):
self.arch_tools[tool.get_tool_name()] = tool
async def analyze_system_architecture(self, repository_path: str, language: str) -> Dict[str, Any]:
architecture_findings = []
tool_mapping = {
'javascript': ['dependency-cruiser'],
'java': ['maven-dependency'],
'python': []
}
applicable_tools = tool_mapping.get(language, [])
for tool_name in applicable_tools:
if tool_name in self.arch_tools:
tool = self.arch_tools[tool_name]
try:
findings = tool.analyze_architecture(repository_path)
architecture_findings.extend(findings)
print(f"{tool_name}: Found {len(findings)} architecture issues")
except Exception as e:
print(f"Error running {tool_name}: {e}")
dependency_graph = await self._build_dependency_graph(architecture_findings)
layering_analysis = await self._analyze_layering(architecture_findings)
modularity_metrics = await self._calculate_modularity_metrics(dependency_graph)
interpretation = await self._interpret_architecture_findings(
architecture_findings,
dependency_graph,
layering_analysis
)
return {
'findings': architecture_findings,
'dependency_graph': dependency_graph,
'layering_analysis': layering_analysis,
'modularity_metrics': modularity_metrics,
'interpretation': interpretation,
'summary': {
'total_violations': len(architecture_findings),
'circular_dependencies': len([f for f in architecture_findings if f.finding_type == 'circular_dependency']),
'layering_violations': len([f for f in architecture_findings if 'layer' in f.architectural_rule.lower()])
}
}
async def _build_dependency_graph(self, findings: List[ArchitectureFinding]) -> Dict[str, Any]:
nodes = set()
edges = []
for finding in findings:
if finding.source_component:
nodes.add(finding.source_component)
if finding.target_component:
nodes.add(finding.target_component)
edges.append({
'source': finding.source_component,
'target': finding.target_component,
'type': finding.finding_type
})
return {
'nodes': list(nodes),
'edges': edges,
'node_count': len(nodes),
'edge_count': len(edges)
}
async def _analyze_layering(self, findings: List[ArchitectureFinding]) -> Dict[str, Any]:
layering_violations = [f for f in findings if 'layer' in f.architectural_rule.lower()]
return {
'violations_found': len(layering_violations),
'violations': layering_violations,
'has_clear_layering': len(layering_violations) == 0
}
async def _calculate_modularity_metrics(self, dependency_graph: Dict[str, Any]) -> Dict[str, Any]:
node_count = dependency_graph['node_count']
edge_count = dependency_graph['edge_count']
if node_count == 0:
return {'coupling': 0.0, 'cohesion': 0.0}
avg_dependencies = edge_count / node_count if node_count > 0 else 0
coupling_score = min(avg_dependencies / 10.0, 1.0)
return {
'average_dependencies': round(avg_dependencies, 2),
'coupling_score': round(coupling_score, 2),
'total_components': node_count,
'total_dependencies': edge_count
}
async def _interpret_architecture_findings(self,
findings: List[ArchitectureFinding],
dependency_graph: Dict[str, Any],
layering_analysis: Dict[str, Any]) -> str:
if not findings:
return "No significant architecture issues detected. The system demonstrates good architectural practices."
findings_summary = self._format_architecture_findings(findings[:10])
interpretation_prompt = f"""
Analyze the following architectural findings and provide expert interpretation:
Architecture Findings:
{findings_summary}
Dependency Metrics:
- Total Components: {dependency_graph['node_count']}
- Total Dependencies: {dependency_graph['edge_count']}
- Average Dependencies per Component: {dependency_graph['edge_count'] / max(dependency_graph['node_count'], 1):.2f}
Layering Analysis:
- Layering Violations: {layering_analysis['violations_found']}
Provide:
1. Overall assessment of the architectural health
2. Most critical architectural issues that need immediate attention
3. Patterns indicating systemic architectural problems
4. Impact on maintainability, scalability, and evolvability
5. Specific recommendations for architectural improvements
Focus on actionable insights that architects and technical leads can use to improve the system.
"""
response = await self.llm_manager.generate(LLMRequest(
prompt=interpretation_prompt,
system_prompt="You are a senior software architect analyzing system architecture. Provide strategic, actionable insights.",
temperature=0.3,
max_tokens=1500,
task_type="architecture_interpretation"
))
return response.text
def _format_architecture_findings(self, findings: List[ArchitectureFinding]) -> str:
formatted = []
for i, finding in enumerate(findings, 1):
formatted.append(f"\n{i}. {finding.finding_type.upper()} - {finding.severity.upper()}")
formatted.append(f" Description: {finding.description}")
formatted.append(f" Source: {finding.source_component}")
if finding.target_component:
formatted.append(f" Target: {finding.target_component}")
formatted.append(f" Rule: {finding.architectural_rule}")
return '\n'.join(formatted)
This architecture integration framework demonstrates how to incorporate architecture-level analysis tools that examine system structure, dependencies, and design quality. The framework detects circular dependencies, layering violations, and excessive coupling. The LLM integration interprets these findings in the context of architectural principles, explaining how structural issues impact system quality attributes like maintainability and evolvability.
Synthesizing Multi-Tool Analysis Results
The true power of tool integration emerges when synthesizing results across multiple tools and analysis levels. A component might pass all code-level quality checks but participate in circular dependencies at the architecture level. A security vulnerability detected by Bandit might be mitigated by architectural controls that static analysis cannot see. The synthesis layer correlates findings across tools, identifies conflicts and confirmations, and generates holistic assessments that consider multiple perspectives.
The synthesis process builds a comprehensive picture by combining code-level findings from tools like Pylint and ESLint, security findings from tools like Bandit and SonarQube, and architecture findings from dependency analyzers. The knowledge graph serves as the integration point where findings from different tools are associated with the same code entities. When multiple tools report issues in the same component, the synthesis layer recognizes this correlation and elevates the component's priority for remediation.
The language model agents perform the interpretive synthesis that transforms raw tool outputs into strategic insights. They identify patterns across findings, such as recognizing that multiple security vulnerabilities in different components all stem from inadequate input validation, suggesting a systemic training or process issue rather than isolated coding mistakes. They assess the relative importance of findings based on business context, architectural significance, and remediation effort. They generate prioritized action plans that sequence remediation work to maximize impact while respecting dependencies between fixes.
Here is an implementation of the synthesis layer:
from typing import Dict, List, Any
from dataclasses import dataclass
@dataclass
class SynthesizedFinding:
title: str
severity: str
category: str
description: str
contributing_tools: List[str]
affected_components: List[str]
business_impact: str
technical_impact: str
recommendations: List[str]
evidence: List[Dict[str, Any]]
confidence: float
class MultiToolSynthesisAgent:
def __init__(self, llm_manager, knowledge_graph_manager):
self.llm_manager = llm_manager
self.kg_manager = knowledge_graph_manager
async def synthesize_analysis_results(self,
code_findings: Dict[str, List[ToolFinding]],
architecture_findings: List[ArchitectureFinding],
context: Dict[str, Any]) -> Dict[str, Any]:
print("Synthesizing findings across all analysis tools...")
component_analysis = await self._analyze_by_component(code_findings, architecture_findings)
pattern_analysis = await self._identify_systemic_patterns(code_findings, architecture_findings)
priority_assessment = await self._assess_priorities(component_analysis, pattern_analysis, context)
synthesized_findings = await self._generate_synthesized_findings(
component_analysis,
pattern_analysis,
priority_assessment
)
strategic_recommendations = await self._generate_strategic_recommendations(
synthesized_findings,
context
)
return {
'synthesized_findings': synthesized_findings,
'component_analysis': component_analysis,
'pattern_analysis': pattern_analysis,
'priority_assessment': priority_assessment,
'strategic_recommendations': strategic_recommendations,
'summary': {
'total_synthesized_findings': len(synthesized_findings),
'high_priority_items': len([f for f in synthesized_findings if f.severity in ['high', 'critical']]),
'systemic_patterns_identified': len(pattern_analysis),
'components_analyzed': len(component_analysis)
}
}
async def _analyze_by_component(self,
code_findings: Dict[str, List[ToolFinding]],
architecture_findings: List[ArchitectureFinding]) -> Dict[str, Any]:
component_map = {}
for tool_name, findings in code_findings.items():
for finding in findings:
component = finding.filepath
if component not in component_map:
component_map[component] = {
'component': component,
'code_findings': [],
'architecture_findings': [],
'total_issues': 0,
'severity_score': 0
}
component_map[component]['code_findings'].append({
'tool': tool_name,
'finding': finding
})
component_map[component]['total_issues'] += 1
severity_weights = {'critical': 10, 'high': 5, 'medium': 2, 'low': 1, 'info': 0.5}
component_map[component]['severity_score'] += severity_weights.get(finding.severity, 1)
for arch_finding in architecture_findings:
component = arch_finding.source_component
if component not in component_map:
component_map[component] = {
'component': component,
'code_findings': [],
'architecture_findings': [],
'total_issues': 0,
'severity_score': 0
}
component_map[component]['architecture_findings'].append(arch_finding)
component_map[component]['total_issues'] += 1
component_map[component]['severity_score'] += 5
sorted_components = sorted(
component_map.values(),
key=lambda x: x['severity_score'],
reverse=True
)
return {comp['component']: comp for comp in sorted_components}
async def _identify_systemic_patterns(self,
code_findings: Dict[str, List[ToolFinding]],
architecture_findings: List[ArchitectureFinding]) -> List[Dict[str, Any]]:
patterns = []
all_code_findings = []
for findings in code_findings.values():
all_code_findings.extend(findings)
rule_frequency = {}
for finding in all_code_findings:
rule_id = finding.rule_id
if rule_id:
rule_frequency[rule_id] = rule_frequency.get(rule_id, 0) + 1
frequent_violations = {rule: count for rule, count in rule_frequency.items() if count >= 5}
for rule_id, count in sorted(frequent_violations.items(), key=lambda x: x[1], reverse=True):
pattern = {
'pattern_type': 'frequent_violation',
'rule_id': rule_id,
'occurrence_count': count,
'severity': 'medium',
'description': f"Rule {rule_id} violated {count} times across the codebase",
'implication': 'Suggests systemic issue or training gap'
}
patterns.append(pattern)
security_findings = [f for f in all_code_findings if f.category == 'security']
if len(security_findings) >= 3:
patterns.append({
'pattern_type': 'security_concern',
'occurrence_count': len(security_findings),
'severity': 'high',
'description': f"Multiple security vulnerabilities detected ({len(security_findings)} issues)",
'implication': 'Security practices may need improvement'
})
circular_deps = [f for f in architecture_findings if f.finding_type == 'circular_dependency']
if len(circular_deps) >= 2:
patterns.append({
'pattern_type': 'architectural_erosion',
'occurrence_count': len(circular_deps),
'severity': 'high',
'description': f"Multiple circular dependencies detected ({len(circular_deps)} cycles)",
'implication': 'Architecture is degrading, refactoring needed'
})
return patterns
async def _assess_priorities(self,
component_analysis: Dict[str, Any],
pattern_analysis: List[Dict[str, Any]],
context: Dict[str, Any]) -> List[Dict[str, Any]]:
priority_items = []
for component_name, component_data in list(component_analysis.items())[:20]:
if component_data['severity_score'] > 10:
priority_items.append({
'type': 'component',
'name': component_name,
'priority': 'high' if component_data['severity_score'] > 20 else 'medium',
'score': component_data['severity_score'],
'issue_count': component_data['total_issues'],
'rationale': f"Component has {component_data['total_issues']} issues with severity score {component_data['severity_score']}"
})
for pattern in pattern_analysis:
if pattern['severity'] in ['high', 'critical']:
priority_items.append({
'type': 'pattern',
'name': pattern['description'],
'priority': pattern['severity'],
'score': pattern['occurrence_count'] * 5,
'rationale': pattern['implication']
})
sorted_priorities = sorted(priority_items, key=lambda x: x['score'], reverse=True)
return sorted_priorities[:15]
async def _generate_synthesized_findings(self,
component_analysis: Dict[str, Any],
pattern_analysis: List[Dict[str, Any]],
priority_assessment: List[Dict[str, Any]]) -> List[SynthesizedFinding]:
synthesized = []
for pattern in pattern_analysis:
if pattern['severity'] in ['high', 'critical']:
finding = SynthesizedFinding(
title=pattern['description'],
severity=pattern['severity'],
category='systemic_issue',
description=f"{pattern['description']}. {pattern['implication']}",
contributing_tools=['multiple'],
affected_components=[],
business_impact=await self._assess_business_impact(pattern),
technical_impact=pattern['implication'],
recommendations=await self._generate_pattern_recommendations(pattern),
evidence=[pattern],
confidence=0.8
)
synthesized.append(finding)
high_priority_components = [p for p in priority_assessment if p['type'] == 'component' and p['priority'] == 'high']
for priority_comp in high_priority_components[:5]:
comp_name = priority_comp['name']
comp_data = component_analysis.get(comp_name, {})
finding = SynthesizedFinding(
title=f"Critical Issues in {comp_name}",
severity='high',
category='component_quality',
description=f"Component has {comp_data.get('total_issues', 0)} issues requiring attention",
contributing_tools=list(set(f['tool'] for f in comp_data.get('code_findings', []))),
affected_components=[comp_name],
business_impact="Component quality issues increase maintenance costs and defect risk",
technical_impact=f"Severity score: {comp_data.get('severity_score', 0)}",
recommendations=await self._generate_component_recommendations(comp_data),
evidence=comp_data.get('code_findings', [])[:5],
confidence=0.9
)
synthesized.append(finding)
return synthesized
async def _assess_business_impact(self, pattern: Dict[str, Any]) -> str:
impact_prompt = f"""
Pattern identified: {pattern['description']}
Implication: {pattern['implication']}
Occurrence count: {pattern['occurrence_count']}
What is the business impact of this systemic pattern? Provide a concise 1-2 sentence statement focusing on:
- Risk to business operations
- Cost implications
- Impact on time-to-market or agility
Be specific and business-focused.
"""
response = await self.llm_manager.generate(LLMRequest(
prompt=impact_prompt,
system_prompt="You translate technical patterns into business impact statements for executives.",
temperature=0.3,
max_tokens=150,
task_type="business_impact"
))
return response.text.strip()
async def _generate_pattern_recommendations(self, pattern: Dict[str, Any]) -> List[str]:
return [
f"Address the root cause of {pattern['pattern_type']}",
"Implement automated checks to prevent recurrence",
"Provide team training on best practices"
]
async def _generate_component_recommendations(self, component_data: Dict[str, Any]) -> List[str]:
return [
"Prioritize refactoring this component",
"Add comprehensive test coverage before making changes",
"Consider breaking into smaller, more focused modules"
]
async def _generate_strategic_recommendations(self,
synthesized_findings: List[SynthesizedFinding],
context: Dict[str, Any]) -> List[Dict[str, Any]]:
findings_summary = self._format_synthesized_findings(synthesized_findings[:10])
strategy_prompt = f"""
Based on comprehensive multi-tool analysis, generate strategic recommendations:
Key Findings:
{findings_summary}
Project Context:
- Repository: {context.get('repository_name', 'Unknown')}
- Team Size: {context.get('team_size', 'Unknown')}
- Business Criticality: {context.get('business_criticality', 'Unknown')}
Provide 5-7 strategic recommendations that:
1. Address the most critical systemic issues
2. Are prioritized by business impact
3. Are realistic given typical resource constraints
4. Include both immediate actions and longer-term improvements
5. Consider dependencies between recommendations
Format each recommendation as:
- Title: Brief, action-oriented title
- Priority: High/Medium/Low
- Description: 2-3 sentences explaining what and why
- Expected Impact: Concrete benefits
- Effort Estimate: Rough sizing (Small/Medium/Large)
"""
response = await self.llm_manager.generate(LLMRequest(
prompt=strategy_prompt,
system_prompt="You are a technical strategist helping organizations improve their software systems. Provide practical, prioritized recommendations.",
temperature=0.4,
max_tokens=2000,
task_type="strategic_planning"
))
recommendations = await self._parse_strategic_recommendations(response.text)
return recommendations
def _format_synthesized_findings(self, findings: List[SynthesizedFinding]) -> str:
formatted = []
for i, finding in enumerate(findings, 1):
formatted.append(f"\n{i}. [{finding.severity.upper()}] {finding.title}")
formatted.append(f" Category: {finding.category}")
formatted.append(f" Impact: {finding.business_impact}")
formatted.append(f" Affected: {', '.join(finding.affected_components[:3])}")
return '\n'.join(formatted)
async def _parse_strategic_recommendations(self, llm_response: str) -> List[Dict[str, Any]]:
recommendations = []
sections = llm_response.split('\n\n')
for section in sections:
if 'Title:' in section or 'Priority:' in section:
rec = {
'title': '',
'priority': 'medium',
'description': '',
'expected_impact': '',
'effort_estimate': 'medium'
}
lines = section.split('\n')
for line in lines:
if line.startswith('Title:') or line.startswith('- Title:'):
rec['title'] = line.split(':', 1)[1].strip()
elif 'Priority:' in line:
priority_text = line.split(':', 1)[1].strip().lower()
if 'high' in priority_text:
rec['priority'] = 'high'
elif 'low' in priority_text:
rec['priority'] = 'low'
elif 'Description:' in line:
rec['description'] = line.split(':', 1)[1].strip()
elif 'Impact:' in line or 'Expected Impact:' in line:
rec['expected_impact'] = line.split(':', 1)[1].strip()
elif 'Effort:' in line:
effort_text = line.split(':', 1)[1].strip().lower()
if 'large' in effort_text:
rec['effort_estimate'] = 'large'
elif 'small' in effort_text:
rec['effort_estimate'] = 'small'
if rec['title']:
recommendations.append(rec)
return recommendations[:7]
This synthesis framework demonstrates how to combine findings from multiple tools and analysis levels into coherent, actionable insights. The system identifies systemic patterns across findings, assesses priorities based on severity and business impact, generates synthesized findings that represent holistic issues rather than isolated tool outputs, and produces strategic recommendations that guide remediation efforts.
The multi-tool integration approach transforms the multi-agent system from a collection of independent analyzers into a comprehensive quality assessment platform that provides value far beyond what any single tool could deliver. By correlating findings, identifying patterns, and interpreting results in business context, the system generates insights that drive meaningful improvements in software quality, architecture, and development practices.
CONCLUSION
The Transformative Potential of Multi-Agent AI Systems for Software Analysis
The multi-agent AI system architecture presented in this comprehensive guide represents a fundamental shift in how organizations can understand, assess, and improve their software systems. By combining the precision of traditional static analysis tools with the interpretive capabilities of large language models, orchestrated through specialized agents with distinct analytical responsibilities, we have created a framework that transcends the limitations of conventional code analysis approaches.
Traditional code analysis tools excel at detecting specific, well-defined issues—syntax errors, style violations, known vulnerability patterns, and complexity metrics. However, they struggle with contextual interpretation, cross-cutting concerns, and the synthesis of findings into actionable strategic insights. Human experts provide this interpretive layer but face scalability constraints and consistency challenges when analyzing large, complex systems. The multi-agent architecture bridges this gap by automating sophisticated analytical reasoning while maintaining the depth and nuance that only intelligent interpretation can provide.
The system we have explored operates across multiple dimensions of software quality simultaneously. It parses code in multiple programming languages to build comprehensive knowledge graphs that capture structural relationships. It employs vector databases to enable semantic search across code and documentation. It integrates established static analysis tools to leverage their specialized detection capabilities. It uses architecture analysis tools to assess system-level structure and dependencies. Most importantly, it orchestrates specialized AI agents that reason about findings in context, identify patterns across multiple sources, and generate insights tailored to different stakeholder needs.
Key Architectural Principles and Design Decisions
Several fundamental architectural principles underpin the system's effectiveness. The separation of concerns through specialized agents allows each component to focus on its domain of expertise while the orchestration layer manages their coordination. The architecture agent understands architectural patterns, quality attributes, and structural principles. The security agent knows common vulnerabilities, attack vectors, and defensive patterns. The domain analysis agent employs natural language processing to extract business concepts and assess domain model alignment. Each agent brings deep, focused capability to its analytical domain.
The knowledge graph serves as the central integration point where information from diverse sources converges into a unified, queryable representation. Code entities, architectural components, requirements, test cases, and analysis findings all exist as nodes in this graph, connected by relationships that capture dependencies, implementations, violations, and traceability links. This graph-based representation enables sophisticated queries that span multiple abstraction levels, answering questions like "Which components implement security requirement REQ-SEC-042, and are they adequately tested?" that would be difficult or impossible with traditional relational databases.
The vector database complements the knowledge graph by enabling semantic search over unstructured text. When an agent needs to understand the purpose of a component, find relevant documentation, or identify similar code patterns, vector-based similarity search retrieves contextually relevant information even when exact keyword matches don't exist. This retrieval-augmented generation approach grounds the language models' reasoning in factual information from the codebase, dramatically reducing hallucination and improving the accuracy of generated insights.
The flexible LLM integration architecture supports multiple providers and deployment models, from cloud-based services offering cutting-edge capabilities to local models running on consumer hardware for privacy and cost control. The task routing mechanism directs different types of analytical work to appropriate models, balancing quality, cost, and latency. The system can use expensive, powerful models for complex reasoning tasks while routing routine analysis to cheaper or local alternatives, optimizing the cost-performance tradeoff.
Advanced Reasoning Patterns Enable Sophisticated Analysis
The reasoning patterns implemented in the system—chain-of-thought, self-reflection, and tree-of-thought—elevate the quality of analysis beyond what simple prompt-response interactions could achieve. Chain-of-thought reasoning breaks complex analytical questions into sequences of simpler steps, with each step building on previous results. This structured approach enables the system to tackle problems that would overwhelm a single prompt, such as assessing whether a system can support a new feature by systematically analyzing involved components, current capabilities, gaps, and required effort.
Self-reflection adds a critical evaluation layer where the system examines its own reasoning for potential flaws. After generating an analysis, a self-reflection prompt asks the model to critique its work, identifying unsupported claims, alternative interpretations, and overlooked factors. This metacognitive capability catches errors and acknowledges uncertainty appropriately, producing more reliable and nuanced analyses than systems that lack self-evaluation mechanisms.
Tree-of-thought reasoning explores multiple hypotheses in parallel, evaluating alternatives before committing to conclusions. When analyzing the cause of a system issue, the system might explore hypotheses about architectural problems, code quality issues, and operational factors simultaneously, assessing each based on available evidence before determining which explanation best fits the facts. This multi-path exploration reduces premature commitment to potentially incorrect conclusions.
Integration with Existing Tools Maximizes Value
The integration of established static analysis and architecture tools demonstrates a pragmatic approach to building AI systems. Rather than attempting to reimplement the sophisticated pattern matching and data flow analysis that tools like SonarQube, Pylint, and Bandit have refined over years, the multi-agent system orchestrates these tools and adds interpretive value on top of their outputs. This division of labor leverages the strengths of both traditional static analysis and modern AI capabilities.
The normalization layer that translates diverse tool outputs into a unified representation enables cross-tool correlation and synthesis. When multiple tools report issues affecting the same code location, the system recognizes this correlation and elevates the component's priority. When findings from code-level tools align with architecture-level issues, the system identifies this as evidence of systemic problems requiring strategic intervention rather than isolated fixes.
The LLM-powered interpretation layer transforms raw tool outputs into contextualized insights. A static analysis tool reports that a function has cyclomatic complexity of thirty-five. The AI agent examines the function in context, understands its business purpose, assesses whether the complexity is inherent or represents poor design, and generates specific refactoring recommendations. This combination of precise automated detection with contextual interpretation provides analysis that neither approach could achieve alone.
Stakeholder-Specific Views Address Diverse Needs
The report generation system's ability to produce tailored views for different stakeholders represents a crucial capability for organizational adoption. Executives need high-level summaries that highlight business impacts and strategic implications. They care about risk to operations, cost implications, and competitive positioning, not the technical details of circular dependencies or cyclomatic complexity. The executive view translates technical findings into business language and presents them through frameworks like SWOT analysis that resonate with business decision-makers.
Architects need technical depth with comprehensive findings, architectural assessments, and detailed recommendations. The architect view provides access to all findings, dependency graphs, pattern detection results, and quality metrics. It presents information at the level of abstraction where architects work—components, layers, services, and architectural patterns—rather than individual functions or classes.
Developers need actionable items with clear priorities and specific guidance. The developer view groups findings by component, highlights issues in code they own or maintain, and provides concrete recommendations for remediation. It focuses on what needs to be fixed and how to fix it, supporting the practical work of improving code quality.
Project managers need effort estimates, timelines, and risk assessments to plan remediation work. The project manager view provides prioritized recommendations with rough effort estimates, suggested timelines that sequence work appropriately, and risk assessments that inform resource allocation decisions.
Deployment Considerations and Operational Realities
Deploying the multi-agent analysis system in production environments requires attention to scalability, reliability, and cost management. Large enterprise repositories can take hours to analyze comprehensively, necessitating robust job management with progress tracking, error recovery, and result persistence. The system should support both on-demand analysis triggered by user requests and scheduled periodic analysis that tracks system evolution over time.
The knowledge graph and vector database require sufficient infrastructure resources. A large repository might generate millions of nodes and edges in the knowledge graph and millions of embeddings in the vector database. Proper indexing, query optimization, and regular maintenance ensure acceptable performance as data accumulates. Cloud-based deployments can leverage managed graph database and vector database services, while on-premises deployments require dedicated infrastructure and operational expertise.
LLM costs can become significant with heavy usage, particularly when using premium cloud models. The task routing configuration should balance cost and quality by reserving expensive models for tasks that truly benefit from their advanced capabilities. Usage tracking and budget alerts prevent unexpected cost overruns. Caching LLM responses for identical queries reduces redundant API calls. For organizations with strict data privacy requirements or cost constraints, local model deployment provides an alternative, though typically with reduced analytical sophistication.
Security considerations include protecting API keys for cloud LLM providers, securing access to knowledge graphs and vector databases, and ensuring that analysis reports containing potentially sensitive information about system weaknesses are properly access-controlled. The system should support role-based access control so different stakeholders see only information appropriate for their roles. Audit logging tracks who accessed what information and when, supporting compliance and security investigations.
Future Directions and Evolving Capabilities
The multi-agent architecture provides a foundation for continuous enhancement as AI capabilities advance. As language models improve in reasoning ability, context window size, and domain-specific knowledge, the system's analytical depth will increase correspondingly. The modular agent architecture allows upgrading individual components independently—replacing the security agent with a more sophisticated version without disrupting other agents or the orchestration framework.
Emerging capabilities like multimodal models that process both code and architectural diagrams could enhance the system's ability to assess alignment between documented and implemented architecture. Models fine-tuned on specific programming languages or domains could provide more accurate and relevant analysis for specialized contexts. Reinforcement learning approaches could enable agents to improve their analytical strategies based on feedback about which findings led to actual improvements.
The integration of runtime observability data—logs, metrics, traces, and error reports from production systems—could enable the system to correlate static analysis findings with actual operational issues. A component flagged for high complexity in static analysis that also shows frequent errors or performance problems in production would receive elevated priority. This fusion of static and dynamic analysis provides a more complete picture of system health.
The system could evolve to support interactive analysis where human experts collaborate with AI agents in real-time. An architect exploring a design question could engage in a dialogue with the architecture agent, asking follow-up questions, requesting deeper analysis of specific aspects, and providing domain knowledge that refines the agent's reasoning. This human-AI collaboration model combines the scalability and consistency of automated analysis with the creativity and contextual understanding of human experts.
Broader Implications for Software Engineering Practice
Beyond its immediate analytical capabilities, the multi-agent system represents a shift in how organizations approach software quality and technical debt management. Traditional approaches treat code quality as a periodic concern addressed through manual code reviews and occasional tool-based scans. The multi-agent system enables continuous, comprehensive quality assessment that provides ongoing visibility into system health.
This continuous assessment supports data-driven decision-making about technical investments. When business stakeholders ask whether the team should prioritize new features or technical debt reduction, the system provides objective evidence about the current state of technical debt, its impact on development velocity, and the risks it poses. This evidence-based approach elevates technical quality discussions from subjective opinions to factual analysis.
The system also supports knowledge transfer and team development. New team members can query the system to understand why architectural decisions were made, how components interact, and where technical debt exists. The system's explanations, grounded in actual code and documentation, provide a form of interactive, always-available documentation that complements traditional written artifacts.
For organizations managing multiple systems or product portfolios, the multi-agent architecture enables comparative analysis across systems. Which systems have the highest technical debt? Which follow best practices most consistently? Which pose the greatest security risks? This portfolio-level visibility supports strategic decisions about where to invest in modernization, which systems to sunset, and how to allocate scarce technical resources.
Concluding Reflections
The multi-agent AI system for code repository analysis represents a convergence of multiple technological advances: sophisticated language models with strong reasoning capabilities, graph databases that efficiently represent complex relationships, vector databases that enable semantic search, and established static analysis tools with proven detection capabilities. The architecture we have explored orchestrates these technologies into a coherent system that provides value greater than the sum of its parts.
The system addresses a fundamental challenge in software engineering: the difficulty of maintaining comprehensive understanding of large, complex systems as they evolve. Human experts cannot hold all details in their heads. Documentation becomes outdated. Tribal knowledge resides with individuals who may leave the organization. The multi-agent system provides a scalable, consistent, always-available analytical capability that complements human expertise rather than attempting to replace it.
As organizations grapple with increasing software complexity, accelerating change, and growing technical debt, tools that provide deep, contextual understanding of systems become not just valuable but essential. The multi-agent architecture presented in this guide offers a practical path forward, demonstrating how to build systems that leverage cutting-edge AI capabilities while remaining grounded in proven software engineering practices.
The code examples, architectural patterns, and integration strategies provided throughout this guide are intended to serve as a foundation for practitioners building similar systems. The field of AI-powered software analysis is rapidly evolving, and the specific tools and models will continue to advance. However, the fundamental architectural principles—specialized agents with focused responsibilities, knowledge graphs for integration, retrieval-augmented generation for grounding, multi-tool synthesis for comprehensive analysis—provide a durable framework that will remain relevant as the underlying technologies improve.
Organizations that invest in building these capabilities position themselves to manage software complexity more effectively, make better technical decisions, and maintain higher quality systems. The multi-agent approach transforms code analysis from a periodic, tool-driven activity into an ongoing, intelligence-augmented process that provides continuous insight into system health and guides continuous improvement.
The journey from concept to production-ready multi-agent analysis system requires significant effort, but the value delivered—in improved code quality, reduced technical debt, better architectural decisions, and enhanced team productivity—justifies the investment. As AI capabilities continue to advance and integration patterns mature, these systems will become increasingly sophisticated and accessible, ultimately transforming how organizations understand and improve their software systems.
No comments:
Post a Comment