Project: a codebase Q&A agent

import json
import re
from pathlib import Path

ROOT = Path(".").resolve()
SKIP = {".git", "node_modules", ".venv", "__pycache__"}


def project_path(path):
    p = (ROOT / path).resolve()
    if not p.is_relative_to(ROOT):
        raise ValueError(f"{path} is outside the project")
    return p


def walk(directory):
    for p in sorted(project_path(directory).rglob("*")):
        if p.is_file() and not SKIP & set(p.relative_to(ROOT).parts):
            yield p


def list_files(directory="."):
    files = [str(p.relative_to(ROOT)) for p in walk(directory)]
    more = f"\n…and {len(files) - 50} more. List a subdirectory." if len(files) > 50 else ""
    return "\n".join(files[:50]) + more or "No files."


def search(pattern, directory="."):
    regex, hits = re.compile(pattern), []
    for p in walk(directory):
        for n, line in enumerate(p.read_text(errors="ignore").splitlines(), 1):
            if regex.search(line):
                hits.append(f"{p.relative_to(ROOT)}:{n}: {line.strip()[:120]}")
    if not hits:
        return "No matches. Try a shorter pattern, or list_files to look around."
    more = f"\n…{len(hits) - 30} more. Use a narrower pattern." if len(hits) > 30 else ""
    return "\n".join(hits[:30]) + more


def read_file(path, start=1):
    lines = project_path(path).read_text(errors="ignore").splitlines()
    end = min(start + 79, len(lines))
    shown = "\n".join(f"{n}: {lines[n - 1]}" for n in range(start, end + 1))
    return shown + (f"\n…{len(lines)} lines in total." if end < len(lines) else "")


TOOLS = {"list_files": list_files, "search": search, "read_file": read_file}

def schema(name, description, required=(), **params):
    return {
        "name": name,
        "description": description,
        "parameters": {"type": "object", "properties": params, "required": list(required)},
    }


SCHEMAS = [
    schema("list_files", "List files under a directory of the project (max 50).",
           directory={"type": "string", "description": "Relative path, e.g. 'src'. Default '.'"}),
    schema("search", "Regex search across the project. Returns path:line: text for each match.", ["pattern"],
           pattern={"type": "string", "description": "Python regex, e.g. 'def apply_coupon'"},
           directory={"type": "string", "description": "Limit to this directory. Default '.'"}),
    schema("read_file", "Read 80 numbered lines of a file, starting at `start`.", ["path"],
           path={"type": "string", "description": "Relative path, as returned by search"},
           start={"type": "integer", "description": "First line to show. Default 1"}),
]

SYSTEM = (
    "You answer questions about the code in this project. Use the tools to look; never guess. "
    "Cite every claim as path:line, e.g. src/app.py:42. If you can't find it, say so."
)
MAX_STEPS, BUDGET_USD, MAX_REPEATS = 20, 0.30, 2

def run(question, call_model):
    messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": question}]
    seen, spent = {}, 0.0
    for step in range(MAX_STEPS):
        reply = call_model(messages, tools=SCHEMAS)
        messages.append(reply)
        spent += cost_of(reply.usage)
        if not reply.tool_calls:
            return reply.content
        if spent > BUDGET_USD:
            return f"Stopped: spent ${spent:.2f} without an answer."
        for call in reply.tool_calls:
            key = (call.name, json.dumps(call.arguments, sort_keys=True))
            seen[key] = seen.get(key, 0) + 1
            if seen[key] > MAX_REPEATS:
                result = "ERROR: you already made this exact call. Try something different."
            else:
                try:
                    result = TOOLS[call.name](**call.arguments)
                except Exception as e:
                    result = f"ERROR: {type(e).__name__}: {e}"
            messages.append({"role": "tool", "tool_call_id": call.id, "content": str(result)})
    return "Stopped: step limit reached."

CITATION = re.compile(r"([\w./-]+\.\w+):(\d+)")


def citation_problems(answer):
    cited = CITATION.findall(answer)
    if not cited:
        return ["the answer has no path:line citations"]
    problems = []
    for path, line in cited:
        try:
            lines = project_path(path).read_text(errors="ignore").splitlines()
        except (OSError, ValueError):
            problems.append(f"{path} does not exist")
            continue
        if not 1 <= int(line) <= len(lines):
            problems.append(f"{path} has no line {line}")
    return problems


def ask(question, call_model):
    answer = run(question, call_model)
    if answer.startswith("Stopped:"):
        return answer
    problems = citation_problems(answer)
    if problems:
        answer = run(f"{question}\n\nA first answer had bad citations ({'; '.join(problems)}). "
                     "Check the files and answer again.", call_model)
    return answer

print(ask("Where do we validate coupon codes?", call_model))