CodeGraph transforms client codebases into AI-readable knowledge graphs, enabling agents to access precise code context instead of entire repositories. This is NOT for Automatos's own code (though we can index it for meta purposes) - it's for client/user codebases.
Molecules: Symbol relationships (calls, imports, dependencies)
Cells: Contextual code chunks with embeddings
Organs: Multi-file analysis with call graphs
Organisms: Complete codebase understanding
1.3 Value Proposition
"Turn any codebase into an AI-readable knowledge graph. Instead of dumping 10,000 files into context, agents get laser-focused, relevant code snippets."
-- Projects table
CREATE TABLE codegraph_projects (
id SERIAL PRIMARY KEY,
name VARCHAR(255) UNIQUE NOT NULL,
source_type VARCHAR(50) NOT NULL, -- local, github, gitlab
source_url TEXT,
branch VARCHAR(255),
language VARCHAR(50),
total_files INTEGER DEFAULT 0,
total_symbols INTEGER DEFAULT 0,
total_relationships INTEGER DEFAULT 0,
last_indexed TIMESTAMP,
index_duration_seconds FLOAT,
status VARCHAR(50) DEFAULT 'active',
auto_reindex BOOLEAN DEFAULT false,
exclude_patterns JSONB DEFAULT '[]',
metadata JSONB DEFAULT '{}',
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
-- Code symbols (functions, classes, etc.)
CREATE TABLE codegraph_symbols (
id SERIAL PRIMARY KEY,
project_id INTEGER REFERENCES codegraph_projects(id) ON DELETE CASCADE,
symbol_type VARCHAR(50) NOT NULL, -- function, class, interface, etc.
name VARCHAR(500) NOT NULL,
qualified_name VARCHAR(1000), -- Full path: module.Class.method
file_path VARCHAR(1000) NOT NULL,
line_number INTEGER NOT NULL,
signature TEXT,
docstring TEXT,
code_snippet TEXT,
embedding VECTOR(1536), -- For semantic search
metadata JSONB DEFAULT '{}',
created_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_codegraph_symbols_project ON codegraph_symbols(project_id);
CREATE INDEX idx_codegraph_symbols_type ON codegraph_symbols(symbol_type);
CREATE INDEX idx_codegraph_symbols_name ON codegraph_symbols(name);
CREATE INDEX idx_codegraph_symbols_embedding ON codegraph_symbols
USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);
-- Code relationships (calls, imports, inheritance)
CREATE TABLE codegraph_relationships (
id SERIAL PRIMARY KEY,
project_id INTEGER REFERENCES codegraph_projects(id) ON DELETE CASCADE,
from_symbol_id INTEGER REFERENCES codegraph_symbols(id) ON DELETE CASCADE,
to_symbol_id INTEGER REFERENCES codegraph_symbols(id) ON DELETE CASCADE,
relationship_type VARCHAR(50) NOT NULL, -- calls, imports, extends, implements
metadata JSONB DEFAULT '{}',
created_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_codegraph_relationships_project ON codegraph_relationships(project_id);
CREATE INDEX idx_codegraph_relationships_from ON codegraph_relationships(from_symbol_id);
CREATE INDEX idx_codegraph_relationships_to ON codegraph_relationships(to_symbol_id);
-- Query analytics
CREATE TABLE codegraph_query_logs (
id SERIAL PRIMARY KEY,
project_id INTEGER REFERENCES codegraph_projects(id) ON DELETE CASCADE,
query_type VARCHAR(50) NOT NULL, -- symbol, semantic, call_graph
query_text TEXT NOT NULL,
results_count INTEGER,
execution_time_ms FLOAT,
user_id VARCHAR(255),
metadata JSONB DEFAULT '{}',
created_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_codegraph_query_logs_project ON codegraph_query_logs(project_id);
CREATE INDEX idx_codegraph_query_logs_created ON codegraph_query_logs(created_at DESC);
-- File metadata
CREATE TABLE codegraph_files (
id SERIAL PRIMARY KEY,
project_id INTEGER REFERENCES codegraph_projects(id) ON DELETE CASCADE,
file_path VARCHAR(1000) NOT NULL,
file_hash VARCHAR(64), -- SHA-256 for change detection
file_size INTEGER,
lines_of_code INTEGER,
complexity_score INTEGER,
language VARCHAR(50),
last_modified TIMESTAMP,
indexed_at TIMESTAMP DEFAULT NOW(),
metadata JSONB DEFAULT '{}',
UNIQUE(project_id, file_path)
);
CREATE INDEX idx_codegraph_files_project ON codegraph_files(project_id);
CREATE INDEX idx_codegraph_files_path ON codegraph_files(file_path);
class CodeGraphIndexer:
"""
Main indexing engine using tree-sitter
"""
async def index_project(
self,
project_id: int,
source_type: str,
source_path: str,
**options
):
"""
1. Clone/load source code
2. Discover files
3. Parse each file
4. Extract symbols
5. Build relationships
6. Generate embeddings
7. Store in database
"""
# Step 1: Source loading
if source_type == "github":
repo_path = await self.clone_repo(source_path, options)
elif source_type == "local":
repo_path = source_path
else:
raise ValueError(f"Unknown source type: {source_type}")
# Step 2: File discovery
files = self.discover_files(
repo_path,
exclude=options.get("exclude_patterns", [])
)
logger.info(f"Found {len(files)} files to index")
# Step 3: Parse files in parallel
symbols_batch = []
relationships_batch = []
for file_path in files:
try:
result = await self.parse_file(file_path)
symbols_batch.extend(result.symbols)
relationships_batch.extend(result.relationships)
except Exception as e:
logger.error(f"Failed to parse {file_path}: {e}")
# Step 4: Generate embeddings
await self.generate_embeddings(symbols_batch)
# Step 5: Store in database
await self.store_symbols(project_id, symbols_batch)
await self.store_relationships(project_id, relationships_batch)
# Step 6: Update project stats
await self.update_project_stats(project_id)
logger.info(f"Indexed {len(symbols_batch)} symbols")
async def parse_file(self, file_path: str):
"""
Parse single file using tree-sitter
"""
# Detect language
language = self.detect_language(file_path)
parser = self.get_parser(language)
# Read file
with open(file_path, 'r', encoding='utf-8') as f:
code = f.read()
# Parse AST
tree = parser.parse(bytes(code, 'utf8'))
# Extract symbols
extractor = SymbolExtractor(language)
symbols = extractor.extract_symbols(tree, file_path)
# Extract relationships
relationships = extractor.extract_relationships(tree, symbols)
return ParseResult(
symbols=symbols,
relationships=relationships,
tree=tree
)
class CodeGraphSearch:
"""
Search engine for code queries
"""
async def symbol_search(
self,
project_id: int,
query: str,
symbol_type: Optional[str] = None,
limit: int = 10
):
"""
Exact/fuzzy symbol name search
"""
query_filter = text("""
SELECT
id,
symbol_type,
name,
qualified_name,
file_path,
line_number,
signature,
docstring,
code_snippet
FROM codegraph_symbols
WHERE project_id = :project_id
AND (:symbol_type IS NULL OR symbol_type = :symbol_type)
AND (
name ILIKE :query
OR qualified_name ILIKE :query
)
ORDER BY
CASE
WHEN name = :exact_query THEN 1
WHEN name ILIKE :starts_with THEN 2
ELSE 3
END,
name
LIMIT :limit
""")
results = db.execute(
query_filter,
{
"project_id": project_id,
"symbol_type": symbol_type,
"query": f"%{query}%",
"exact_query": query,
"starts_with": f"{query}%",
"limit": limit
}
).fetchall()
return [self._format_symbol(r) for r in results]
async def semantic_search(
self,
project_id: int,
query: str,
limit: int = 10
):
"""
Vector similarity search for semantic code queries
"""
# Generate query embedding
query_embedding = await self.generate_embedding(query)
# Vector search
similarity_query = text("""
SELECT
cs.*,
1 - (cs.embedding <=> :query_embedding::vector) as similarity
FROM codegraph_symbols cs
WHERE cs.project_id = :project_id
AND cs.embedding IS NOT NULL
ORDER BY cs.embedding <=> :query_embedding::vector
LIMIT :limit
""")
results = db.execute(
similarity_query,
{
"project_id": project_id,
"query_embedding": query_embedding,
"limit": limit
}
).fetchall()
return [self._format_symbol(r) for r in results]
async def call_graph_search(
self,
project_id: int,
symbol_name: str,
depth: int = 2,
direction: str = "both" # "calls", "called_by", "both"
):
"""
Build call graph starting from symbol
"""
# Find root symbol
root = await self.find_symbol(project_id, symbol_name)
if not root:
return {"error": "Symbol not found"}
# Traverse graph
nodes = [root]
edges = []
visited = {root.id}
queue = [(root.id, 0)]
while queue:
symbol_id, current_depth = queue.pop(0)
if current_depth >= depth:
continue
# Get relationships
if direction in ["calls", "both"]:
outgoing = await self.get_calls_from(symbol_id)
for rel in outgoing:
if rel.to_symbol_id not in visited:
nodes.append(rel.to_symbol)
edges.append(rel)
visited.add(rel.to_symbol_id)
queue.append((rel.to_symbol_id, current_depth + 1))
if direction in ["called_by", "both"]:
incoming = await self.get_calls_to(symbol_id)
for rel in incoming:
if rel.from_symbol_id not in visited:
nodes.append(rel.from_symbol)
edges.append(rel)
visited.add(rel.from_symbol_id)
queue.append((rel.from_symbol_id, current_depth + 1))
return {
"root": symbol_name,
"nodes": [self._format_node(n) for n in nodes],
"edges": [self._format_edge(e) for e in edges],
"depth": depth
}
# In agent execution
async def execute_with_codegraph(
self,
task: str,
context: dict
):
"""
Execute agent task with CodeGraph context
"""
codegraph_project = context.get("codegraph_project")
if codegraph_project:
# Search for relevant code
code_results = await codegraph_search.semantic_search(
project=codegraph_project,
query=task,
limit=5
)
# Format code context
code_context = self.format_code_context(code_results)
# Augment agent prompt
augmented_prompt = f"""
{task}
## Relevant Code Context:
{code_context}
Use the above code as reference when completing this task.
"""
# Execute with enhanced context
result = await self.llm.generate(augmented_prompt)
return result
// Frontend: Chat with code awareness
POST /api/chat/query
{
"message": "How do I authenticate users in this codebase?",
"project": "automatos-ai",
"include_code": true,
"max_results": 3
}
{
"answer": "To authenticate users in this codebase, use the `authenticate_user` function in `services/auth_service.py`. Here's how it works...",
"code_references": [
{
"symbol": "authenticate_user",
"file": "services/auth_service.py",
"line": 45,
"snippet": "def authenticate_user(username: str, password: str) -> User:\n ...",
"relevance": 0.95
}
]
}
