-
Notifications
You must be signed in to change notification settings - Fork 0
agent tools
Factory that builds the 11 LangChain tools the agent can call, each closing over a shared VectorStore, modules dict, files list, dependency graph, and repo path.
| Term | Definition | Example |
|---|---|---|
| blast radius | All files that would be affected if a given file changes — found by following reverse import edges transitively. | If A imports B and C imports A, changing B has blast radius = {A, C}. |
| embedding | A numerical vector (list of numbers) that represents the meaning of text. Similar text → similar vectors. | The code def add(a, b): return a+b might become [0.12, -0.45, 0.78, ...] (768 numbers for jinaai/jina-code-embeddings-1.5b via Ollama/MPS). |
| vector store | A database optimized for storing embeddings and finding the most similar ones quickly. | ChromaDB stores code chunk embeddings and returns the 5 most similar chunks to your query. |
| ChromaDB | An open-source vector database for storing and searching embeddings. Used here to store code chunks. |
collection.query(query_texts=["scan files"], n_results=5) returns the 5 closest code chunks. |
| chunk | A piece of source code (usually one function or class) stored as a unit for search. | The function def scan_directory(root): ... (20 lines) is one chunk. |
| AST | Abstract Syntax Tree — a tree representation of source code structure, where each node is a language construct (function, class, if-statement, etc.). |
def add(a, b): return a+b becomes a tree: FunctionDef → [args: a, b] → [body: Return → BinOp(a + b)]. |
| RAG | Retrieval-Augmented Generation — instead of asking an LLM to answer from memory, first retrieve relevant documents, then include them in the prompt. | Question: "What does scan_directory do?" → retrieve the source code of scan_directory → include it in the LLM prompt → get an accurate answer. |
| LLM | Large Language Model — an AI model (like GPT-4, Claude) that generates text given a prompt. |
get_llm() returns a ChatOpenAI instance that can answer questions about code. |
| LangChain | A Python framework for building LLM-powered applications. Provides chains (prompt → LLM → parser), structured output, and more. |
prompt | llm | StrOutputParser() creates a chain that formats a prompt, sends to LLM, and extracts the string response. |
| diff | The set of changes between two versions of code, showing added (+) and removed (-) lines. |
- old_line\n+ new_line shows old_line was replaced with new_line. |
| hunk | A contiguous block of changes within a diff. One diff can contain multiple hunks (changes in different parts of a file). | A diff might have hunk 1 (lines 10-15 changed) and hunk 2 (lines 80-85 changed). |
Source: src/codewalk/agent/tools.py
Accepts pre-computed codebase data and returns a list of 11 @tool-decorated functions, each capturing the shared data via closure.
Input:
store = VectorStore(persist_dir="/tmp/chroma") # already has 500 chunks indexed
modules_result = {
"modules": {"api": {"files": ["src/api/routes.py"], "file_count": 1, "languages": {"Python": 1}}},
"module_graph": {"api": ["config"]},
"source_root": "src",
"stats": {"total_files": 10},
}
files = [{"file_path": "src/api/routes.py", "language": "Python"}]
deps = {"graph": {"src/api/routes.py": ["src/config.py"]}}
graph_runtime = None # optional GraphRuntime (igraph)
graph_store = None
repo_path = "/home/user/project"Line 35: tools = create_tools(store, modules_result, files=files, deps=deps, graph_runtime=None, graph_store=None, repo_path=repo_path)
What happens inside: 13 inner functions are defined, each decorated with @tool. All of them close over store, modules_result, files, deps, graph_runtime, graph_store, and repo_path.
Line (end): return [search_codebase, get_module_info, explain_function, get_overview, get_blast_radius_map, get_reading_order, get_execution_flow, load_guidelines, get_architecture_health, apply_fix, verify_fix] → a list of 11 tool functions.
Returns: [<function search_codebase>, <function get_module_info>, ..., <function verify_fix>]
Runs multi-query corrective RAG: expands the user question into 1-3 complementary search angles, runs corrective RAG for each in parallel, then synthesizes the results into one answer.
Input: query = "how does authentication work"
Internal flow:
-
expand_query("how does authentication work")→["how does authentication work", "authentication login flow", "verify user credentials"] -
_multi_query_search()runsask_corrective()for each angle in parallel. -
_synthesize_answers()merges the partial answers into one coherent response.
Returns: "Auth uses JWT tokens in middleware...\n\n---\n_Confident: True | Retries: 0 | Chunks: 4 | Confidence: 0.87_"
Delegates to module_info_text() from query.py to return a module's files, symbols, and dependencies.
Input: module_name = "analysis"
Line 74: return module_info_text(modules_result, "analysis", graph_runtime, graph_store)
→ Looks up modules_result["modules"]["analysis"], formats file list with symbols, dependency relationships.
Returns: "## Module: analysis\n**Files:** 5\n**Languages:** Python (5)\n**Depends on:** config\n**Depended on by:** pipeline\n### Files & Symbols\n- **scanner.py**: ..."
Looks up a specific function or class by name in the vector store and returns its source code with blast radius.
Input: function_name = "scan_directory"
Line 87: return explain_function_text(store, "scan_directory", deps, graph_runtime, graph_store)
→ Searches ChromaDB for chunks whose symbol_name metadata matches "scan_directory", retrieves source code, computes which files break if this function changes.
Returns: "## scan_directory (src/ingestion/scanner.py, L12-L45)\n\ndef scan_directory(path):\n ...\n\n### Blast Radius\n- src/pipeline.py\n- src/mcp/server.py"
Returns a high-level project summary: tech stack, modules, dependency flow, riskiest files.
Input: (no arguments)
Line 125: deps is None → False (deps was provided)
Line 127-129: if not repo_path: return "Error: No repo path available."
Line 129: return overview_text(repo_path, modules_result, deps, graph_runtime)
Returns: "## Project Overview\n**Tech Stack:** Python\n**Modules:** api, config, analysis\n**Entry Points:** api\n**Riskiest Files:**\n1. config.py — breaks 8 files\n..."
If deps is None:
Line 125-126: returns "Error: No analysis data available."
If repo_path is empty:
Line 127-128: returns "Error: No repo path available."
Shows which files break if you change a target file or module.
Input: target = "scanner.py"
Line 114: deps is None → False
Line 115: return blast_radius_map_text(modules_result, deps, "scanner.py", graph_runtime)
→ Finds scanner.py in the dependency graph, walks all reverse dependencies.
Returns: "## Blast Radius: scanner.py\nBreaks 3 files:\n - pipeline.py\n - mcp/server.py\n - api/routes.py"
When target = "" (empty): returns the top 30 riskiest files across the whole repo.
Returns files in dependency order — read leaf dependencies first, then files that depend on them.
Input: module_name = "analysis"
Line 129: files is None or deps is None → False
Line 130: return reading_order_text(files, deps, modules_result, "analysis", graph_runtime)
Returns: "## Reading Order (analysis)\n1. config.py — 0 deps, risk: low\n2. scanner.py — 1 dep (config.py), risk: medium\n3. dependency_graph.py — 2 deps, risk: high"
When module_name = "": returns reading order for the entire repo.
Shows how modules or files connect via imports.
Input: module_name = "" (empty = repo-level)
Line 145: return execution_flow_text(modules_result, deps, "")
Returns: "## Module Flow\ningestion → analysis → embeddings → rag → generation\n\n**Entry Modules:** api, mcp"
Input: module_name = "analysis" → returns file-level flow within that module.
Note:
review_diffandreview_filewere removed from the LangGraph agent tool set. Reviews are now triggered via MCP (codewalk_run_review,codewalk_review_file) or the API (POST /review,/review/file).
Loads team coding guidelines from a directory so they're available to agent-driven queries and review workflows.
Input: docs_path = "/home/user/project/guidelines"
Line 254: path = "/home/user/project/guidelines" (docs_path is non-empty, so it's used directly)
Line 260: os.path.isdir("/home/user/project/guidelines") → True
Line 263: gl_store = get_guidelines_store() → a VectorStore with embedded guideline chunks
Line 267: count = gl_store.chunk_count() → 15
Returns: "Loaded 15 guideline chunks from /home/user/project/guidelines"
When docs_path = "":
Returns: "No path provided. Pass docs_path."
Returns an architecture health report: graph stats, bottleneck files (betweenness centrality), key files (PageRank), and circular dependency detection with suggested fixes.
Input: (no arguments)
Line 343: graph_runtime is None → False (graph_runtime was provided)
Line 346: stats = graph_runtime.get_graph_stats() → file/edge counts and DAG status
Line 348: centrality = graph_runtime.centrality(top_n=5) → top bottleneck and PageRank files
Line 349: cycles = graph_runtime.detect_cycles() → cycle groups and edges to break
Returns: "Files: 120, Edges: 340, DAG: No\nBottlenecks: config.py (0.42), routes.py (0.31)\nKey files (PageRank): config.py, routes.py, pipeline.py\nCycles: 2 groups found\n Cycle 1: models.py ↔ schema.py\nFix — remove these imports:\n - schema.py → models.py"
When graph_runtime is None:
Line 343-344: returns "Error: No graph data available."
Applies a code fix by replacing old_code with new_code in a file. This tool edits files on disk and is interrupted by HITL before execution; the user must approve each fix via /chat/approve.
Input: file_path = "src/auth/login.py", old_code = "def login(u, p):\n return True", new_code = "def login(u, p):\n return verify(u, p)"
Line 406: if not repo_path: return "Error: No repo path available."
Line 408: result = apply_fix_to_file(repo_path, file_path, old_code, new_code)
Returns: "Applied fix to src/auth/login.py" (plus optional validation message)
When repo_path is empty:
Line 406-407: returns "Error: No repo path available."
Runs tests and static analysis to verify a fix. Should be called after apply_fix.
Input: file_paths = ["src/auth/login.py"] (or omitted to run the full suite)
Line 427: if not repo_path: return "Error: No repo path available."
Line 433: sa_issues = run_static_analysis(repo_path, file_paths or [], language_hint=None)
Line 443: test_result = run_tests(repo_path, file_paths or [])
Returns: "## Verification Results\n\nStatic analysis: no issues\n\nTests: PASSED\nCommand: pytest tests/..."
When repo_path is empty:
Line 427-428: returns "Error: No repo path available."