DB-GPT Multi-Agent Framework Architecture: How Agents Collaborate

DB-GPT implements a pluggable, layered multi-agent framework where a ManagerAgent orchestrates specialized ConversableAgents through planning loops, dynamic speaker selection, and shared memory to solve complex database tasks.

The eosphoros-ai/DB-GPT repository provides a modular multi-agent system designed to coordinate autonomous AI agents for database operations. This architecture separates concerns into distinct layers—from HTTP API handling to agent execution—enabling seamless collaboration between planner agents, tool executors, and specialized assistants. Understanding the DB-GPT Multi-Agent framework architecture reveals how the system breaks down user requests into orchestrated sub-tasks executed by a coordinated team of agents.

Core Architecture Layers

The framework consists of six distinct layers, each with specific responsibilities and well-defined interfaces.

Orchestration Layer: MultiAgents

The entry point for all multi-agent interactions is the MultiAgents class in packages/dbgpt-serve/src/dbgpt_serve/agent/agents/controller.py. This component registers the HTTP API endpoint /api/agent_chat_v2 and handles the initial request routing. When a user query arrives, MultiAgents builds a per-conversation AgentMemory using get_or_build_agent_memory(), which combines a vector store for long-term persistence with a short-term buffer through HybridMemory.

Team Management: The Team Class

At the heart of collaboration lies the Team class defined in packages/dbgpt-core/src/dbgpt/agent/core/base_team.py. This container holds the collection of hired agents and tracks message history across the conversation. The hire() method appends agents to self.agents, making them available for task allocation. The Team instance maintains the shared state that allows agents to access previous outputs and coordinate dependencies.

Planning and Management: AutoPlanChatManager

The AutoPlanChatManager in packages/dbgpt-core/src/dbgpt/agent/core/plan/team_auto_plan.py serves as the central coordinator. This special ManagerAgent (inheriting from both ConversableAgent and Team) runs the main execution loop. It implements select_speaker() to determine which agent acts next and manages the planning lifecycle through process_rely_message() to resolve inter-agent dependencies.

Agent Interface: ConversableAgent and Agent

All concrete agents inherit from ConversableAgent (extending the abstract Agent class in packages/dbgpt-core/src/dbgpt/agent/core/agent.py). This interface requires implementations of:

• thinking() – Produce intermediate reasoning prompts
• generate_reply() – Create responses based on context
• act() – Execute actions returning ActionOutput
• verify() – Validate results against sub-task requirements

Memory and Knowledge Systems

The framework utilizes GptsMemory and HybridMemory for state management, storing both plan fragments (GptsPlan) and message history. For external knowledge retrieval, the KnowledgeSpaceRetriever in knowledge_space.py fetches relevant chunks from vector stores, appending them to agent prompt contexts during execution.

How Agents Collaborate: The Execution Flow

Agent collaboration follows a structured lifecycle from request ingestion to completion:

1. Request Initialization

The flow begins when MultiAgents.agent_chat_v2() receives a user query. The system constructs a conversation-specific memory instance combining vector storage and short-term buffers, then instantiates the AutoPlanChatManager to oversee the session.

2. Team Assembly

The manager executes hire() to recruit specialized agents such as PlannerAgent, ToolAssistantAgent, or CodeAssistantAgent. These agents are appended to the team roster with their specific roles and profiles registered for speaker selection.

3. Planning Phase

If no valid plan exists, the manager spawns a PlannerAgent and binds it to shared memory and agent context. Calling planner.generate_reply() produces a GptsPlan object—a list of sub-tasks persisted in GptsMemory for the duration of the conversation.

4. Dynamic Speaker Selection

Each round begins with AutoPlanChatManager.select_speaker(), which prompts the manager to analyze the current goal and available agents. The method returns the next agent's name, with mentioned_agents() resolving ambiguities (e.g., mapping "the analyst" to a concrete ConversableAgent instance).

5. Execution and Verification

The selected agent receives messages via receive(), processes them through thinking() and generate_reply(), then executes act() to produce an ActionOutput. The verify() method checks if the output satisfies the sub-task requirements. Successful results update both the plan memory and message history via process_rely_message(), making them available for dependent steps.

6. Completion Loop

The manager repeats this select-act-verify sequence for up to self.max_round iterations (defaulting to 100) or until the plan status reaches Status.COMPLETE. Throughout the loop, agents access shared memory to retrieve previous outputs and knowledge resources through MultiAgents.get_knowledge_resources().

Code Implementation Examples

Starting a Multi-Agent Chat

Use the high-level API to initiate conversations through the controller:

from dbgpt_serve.agent.agents.controller import multi_agents

async def ask_gpt(question: str, user_id: str):
    conv_id = "conv_123"
    # Streamed response (set stream=False for a single result)

    reply = await multi_agents.agent_chat_v2(
        conv_id=conv_id,
        new_order=0,
        gpts_name="my_multi_agent_app",
        user_query=question,
        user_code=user_id,
        stream=False,
    )
    return reply

Creating a Custom ConversableAgent

Extend ConversableAgent to implement domain-specific logic:

from dbgpt.agent.core.base_agent import ConversableAgent
from dbgpt.agent.core.agent import AgentMessage
from dbgpt.agent.core.action.base import ActionOutput

class SqlAssistantAgent(ConversableAgent):
    """Agent that can run SQL queries and return results."""
    
    async def thinking(self, messages, sender=None, prompt=None):
        # Use the last user message as the prompt for the LLM

        return messages[-1].content, None

    async def generate_reply(self, received_message, sender, **kwargs):
        # Simple echo – replace with actual SQL generation/inference

        return AgentMessage(content=f"Running SQL: {received_message.content}")

    async def act(self, message, sender, **kwargs):
        # Here you would execute the SQL against a DB and return the result

        return ActionOutput(is_exe_success=True, content="42 rows returned")

Register the custom agent with the manager:

from dbgpt.agent.core.plan.team_auto_plan import AutoPlanChatManager

manager = AutoPlanChatManager()
manager.hire([SqlAssistantAgent(name="SQLAgent", role="SQLAssistant")])

Accessing Shared Memory

Agents interact with the conversation memory directly during execution:

async def act(self, message, sender, **kwargs):
    # Store a custom flag in the long-term vector store

    self.memory.memory.store("last_query", message.content)
    # Retrieve later

    last = self.memory.memory.load("last_query")
    return ActionOutput(is_exe_success=True, content=f"Remembered: {last}")

Summary

• DB-GPT's architecture separates concerns into Orchestration, Team Management, Planning, Agent Interface, and Memory layers, enabling modular agent development.
• AutoPlanChatManager serves as the central coordinator in team_auto_plan.py, handling speaker selection and plan execution loops up to 100 rounds by default.
• Collaboration workflow involves hiring agents into a Team, generating a GptsPlan, dynamically selecting speakers via select_speaker(), and verifying actions through shared HybridMemory.
• Extensibility allows developers to subclass ConversableAgent in base_agent.py and register new agents via hire() without modifying core orchestration logic.
• Memory persistence combines short-term buffers with vector stores, accessible through AgentMemory interfaces for context retention across agent handoffs.

Frequently Asked Questions

What is the role of AutoPlanChatManager in DB-GPT's Multi-Agent framework?

The AutoPlanChatManager functions as the central nervous system of the multi-agent framework. It inherits from both ConversableAgent and Team, enabling it to both participate in conversations and manage other agents. According to the source code in packages/dbgpt-core/src/dbgpt/agent/core/plan/team_auto_plan.py, it runs the main execution loop, calls select_speaker() to determine which agent acts next, and uses process_rely_message() to handle dependencies between sub-tasks. It also manages the planning lifecycle by invoking PlannerAgent when new plans are required.

How does DB-GPT handle communication between agents?

Agent communication occurs through a shared memory architecture and message passing protocol. The Team class in base_team.py maintains a collective message history that all agents can access. When an agent completes an action, the verify() method checks the result, and successful outputs are stored in GptsMemory via the memory interfaces. Subsequent agents retrieve these results through the shared AgentMemory instance constructed by MultiAgents.get_or_build_agent_memory(), enabling seamless context transfer without direct agent-to-agent messaging.

Can I add custom agents to the DB-GPT Multi-Agent framework?

Yes, the architecture supports pluggable agents through inheritance. You create a custom agent by subclassing ConversableAgent from packages/dbgpt-core/src/dbgpt/agent/core/base_agent.py and implementing the required methods: thinking(), generate_reply(), act(), and optionally verify(). Once defined, you register the agent using manager.hire([YourAgent()]), which adds it to the team's agent pool. The AutoPlanChatManager will automatically include your agent in the select_speaker() rotation based on its role profile and the current task requirements.

How does the memory system work in DB-GPT's agent collaboration?

The memory system uses a hybrid approach implemented in HybridMemory and GptsMemory classes. Short-term conversation history is buffered for immediate context, while long-term plan fragments and knowledge are stored in vector databases. During initialization, MultiAgents builds per-conversation memory instances that persist across the chat loop. Agents access this memory through self.memory to store intermediate results and load dependencies, with KnowledgeSpaceRetriever providing optional retrieval-augmented generation capabilities for external knowledge bases.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}