diff --git a/chatbot.py b/chatbot.py
index 81ba06a..97e58e8 100644
--- a/chatbot.py
+++ b/chatbot.py
@@ -49,7 +49,9 @@ SYSTEM_PROMPT = (
"identifiers, filenames, and commands mentioned in prose. Use the "
"web_search tool when a question depends on current information you "
"may not know (recent releases, changelogs, current best practices, "
- "error messages you don't recognize)."
+ "error messages you don't recognize). If the user provides a local "
+ "file path or asks to inspect a specific file, prefer the read_file "
+ "tool instead of web_search."
)
# Number of prior user/assistant exchanges to keep in the prompt. Bounds the
@@ -75,6 +77,29 @@ WEB_SEARCH_TOOL = {
},
}
+FILE_READ_TOOL = {
+ "type": "function",
+ "function": {
+ "name": "read_file",
+ "description": "Read a local file from disk and return its contents.",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "path": {"type": "string", "description": "Absolute or relative path to the file."},
+ "start_line": {
+ "type": "integer",
+ "description": "Optional 1-based line number to start reading from.",
+ },
+ "end_line": {
+ "type": "integer",
+ "description": "Optional 1-based line number to stop reading at.",
+ },
+ },
+ "required": ["path"],
+ },
+ },
+}
+
# Matches the ... blocks
# Qwen3.5's chat template trains it to emit when it wants to invoke a tool -
# an XML-style syntax, not Qwen2.5's JSON-in- format.
@@ -109,7 +134,35 @@ def web_search(query: str, max_results: int = 5) -> str:
return "\n\n".join(f"{r['title']}\n{r['href']}\n{r['body']}" for r in results)
-TOOLS = {"web_search": web_search}
+def read_file(path: str, start_line: int | None = None, end_line: int | None = None) -> str:
+ """Read a local file from disk. If start_line/end_line are provided, read
+ only that line range (1-based, inclusive). Returns a helpful error string if
+ the file cannot be read."""
+ file_path = Path(path)
+ if not file_path.exists():
+ return f"File not found: {path}"
+ if not file_path.is_file():
+ return f"Not a file: {path}"
+
+ try:
+ lines = file_path.read_text(encoding="utf-8").splitlines()
+ except UnicodeDecodeError:
+ try:
+ lines = file_path.read_text(encoding="utf-8-sig").splitlines()
+ except Exception as exc: # pragma: no cover - defensive fallback
+ return f"Could not decode file: {path} ({exc})"
+
+ if start_line is None and end_line is None:
+ return "\n".join(lines) + ("\n" if lines and file_path.read_bytes() else "")
+
+ start = 1 if start_line is None else max(1, int(start_line))
+ end = len(lines) if end_line is None else min(len(lines), int(end_line))
+ if start > end:
+ return ""
+ return "\n".join(lines[start - 1 : end]) + ("\n" if start <= end and end <= len(lines) else "")
+
+
+TOOLS = {"web_search": web_search, "read_file": read_file}
def extract_tool_calls(text: str) -> list[dict]:
@@ -140,6 +193,9 @@ class ChatSession:
created_at: str
updated_at: str
history: list[tuple[str, str]] = field(default_factory=list)
+ # Tokens occupied in this chat's own context after its last model call.
+ # None until that chat has been evaluated in some session of the app.
+ context_tokens: int | None = None
def new_session() -> ChatSession:
@@ -167,6 +223,7 @@ def save_session(session: ChatSession) -> None:
"created_at": session.created_at,
"updated_at": session.updated_at,
"history": [[question, answer] for question, answer in session.history],
+ "context_tokens": session.context_tokens,
}
_session_path(session.id).write_text(json.dumps(payload, indent=2), encoding="utf-8")
@@ -191,6 +248,7 @@ def list_sessions() -> list[ChatSession]:
created_at=data["created_at"],
updated_at=data["updated_at"],
history=[tuple(turn) for turn in data["history"]],
+ context_tokens=data.get("context_tokens"),
)
)
except (json.JSONDecodeError, OSError, KeyError, TypeError):
@@ -367,7 +425,13 @@ def ask(
invoked the same way but for text inside the model's block,
letting a caller render its reasoning trace separately from the answer."""
for _ in range(MAX_TOOL_ROUNDS):
- content = _stream_content(llm, messages, [WEB_SEARCH_TOOL], on_token, on_thinking_token)
+ content = _stream_content(
+ llm,
+ messages,
+ [WEB_SEARCH_TOOL, FILE_READ_TOOL],
+ on_token,
+ on_thinking_token,
+ )
if on_usage is not None:
on_usage({"total_tokens": llm.n_tokens})
calls = extract_tool_calls(content)
@@ -385,7 +449,7 @@ def ask(
result = f"Unknown tool: {name}"
else:
if on_tool_call is not None:
- on_tool_call(args.get("query", ""))
+ on_tool_call(name, args.get("query") or args.get("path", ""))
result = func(**args)
messages.append({"role": "tool", "content": result})
diff --git a/gui.py b/gui.py
index 5336e60..6b3c69a 100644
--- a/gui.py
+++ b/gui.py
@@ -439,7 +439,7 @@ class ModelLoaderThread(QThread):
class AskThread(QThread):
- tool_call = pyqtSignal(str)
+ tool_call = pyqtSignal(str, str)
usage = pyqtSignal(dict)
token = pyqtSignal(str)
thinking_token = pyqtSignal(str)
@@ -479,6 +479,11 @@ class ChatWindow(QMainWindow):
self._sessions[0] if self._sessions else chatbot.new_session()
)
self._history: list[tuple[str, str]] = list(self._current_session.history)
+ # The loaded model has a single KV cache. Each chat keeps its own
+ # snapshot of that cache so switching chats restores that chat's
+ # context instead of continuing whichever chat ran last.
+ self._context_states: dict[str, object] = {}
+ self._context_owner: str | None = None
self._pending_question = ""
self._ask_thread: AskThread | None = None
# Position in chat_view's document where the in-progress streamed
@@ -718,6 +723,7 @@ class ChatWindow(QMainWindow):
return # already on an empty, unsaved chat
self._current_session = chatbot.new_session()
self._history = []
+ self._activate_session_context(self._current_session)
self.chat_view.clear()
self._append_system("Dev assist chatbot ready. Ask a development question.")
self.session_list.clearSelection()
@@ -734,6 +740,7 @@ class ChatWindow(QMainWindow):
return
self._current_session = session
self._history = list(session.history)
+ self._activate_session_context(session)
self._render_session_history(session)
self.statusBar().showMessage("Ready.")
@@ -759,6 +766,9 @@ class ChatWindow(QMainWindow):
if reply != QMessageBox.StandardButton.Yes:
return
chatbot.delete_session(session_id)
+ self._context_states.pop(session_id, None)
+ if self._context_owner == session_id:
+ self._context_owner = None
self._sessions = [s for s in self._sessions if s.id != session_id]
if self._current_session.id == session_id:
if self._sessions:
@@ -766,6 +776,7 @@ class ChatWindow(QMainWindow):
else:
self._current_session = chatbot.new_session()
self._history = list(self._current_session.history)
+ self._activate_session_context(self._current_session)
self.chat_view.clear()
if self._current_session.history:
self._render_session_history(self._current_session)
@@ -776,7 +787,7 @@ class ChatWindow(QMainWindow):
def _on_model_loaded(self, llm) -> None:
self._llm = llm
self._n_ctx = llm.n_ctx()
- self.context_status_label.setText(f"Context: 0 / {self._n_ctx}")
+ self._activate_session_context(self._current_session)
self.statusBar().showMessage("Ready.")
self._append_system("Dev assist chatbot ready. Ask a development question.")
self._set_input_enabled(True)
@@ -799,6 +810,7 @@ class ChatWindow(QMainWindow):
self._answer_started = False
self._stream_thinking_id = None
+ self._activate_session_context(self._current_session)
messages = chatbot.build_messages(self._history, question)
self._pending_question = question
@@ -936,16 +948,47 @@ class ChatWindow(QMainWindow):
return
self._render_thinking(block_id, expanded=not block["expanded"])
- def _on_tool_call(self, query: str) -> None:
+ def _on_tool_call(self, tool_name: str, target: str) -> None:
self._collapse_open_thinking()
self._discard_stream_card()
- self.statusBar().showMessage(f"Searching: {query}")
- self._append_system(f"Searching the web for: {query}")
+ if tool_name == "read_file":
+ self.statusBar().showMessage(f"Reading file: {target}")
+ self._append_system(f"Reading file: {target}")
+ else:
+ self.statusBar().showMessage(f"Searching: {target}")
+ self._append_system(f"Searching the web for: {target}")
+
+ def _show_context_usage(self, total: int | None) -> None:
+ if total is None or not self._n_ctx:
+ shown = "--" if total is None else total
+ limit = self._n_ctx or "--"
+ self.context_status_label.setText(f"Context: {shown} / {limit}")
+ return
+ pct = 100 * total / self._n_ctx
+ self.context_status_label.setText(f"Context: {total} / {self._n_ctx} ({pct:.0f}%)")
+
+ def _activate_session_context(self, session: "chatbot.ChatSession") -> None:
+ """Make the loaded model's KV cache match this chat.
+
+ Leaving a chat stores its cache; entering one restores it, or clears
+ the cache when that chat has no snapshot yet (first visit this run).
+ """
+ self._show_context_usage(session.context_tokens)
+ if self._llm is None or self._context_owner == session.id:
+ return
+ if self._context_owner is not None:
+ self._context_states[self._context_owner] = self._llm.save_state()
+ state = self._context_states.get(session.id)
+ if state is None:
+ self._llm.reset()
+ else:
+ self._llm.load_state(state)
+ self._context_owner = session.id
def _on_usage(self, usage: dict) -> None:
total = usage["total_tokens"]
- pct = 100 * total / self._n_ctx if self._n_ctx else 0
- self.context_status_label.setText(f"Context: {total} / {self._n_ctx} ({pct:.0f}%)")
+ self._current_session.context_tokens = total
+ self._show_context_usage(total)
def _on_answer(self, answer: str) -> None:
# Normally already collapsed by the first answer token (_on_token),
diff --git a/test_chatbot_tools.py b/test_chatbot_tools.py
new file mode 100644
index 0000000..5b1128b
--- /dev/null
+++ b/test_chatbot_tools.py
@@ -0,0 +1,57 @@
+import chatbot
+
+
+def test_read_file_tool_returns_file_contents(tmp_path):
+ sample = tmp_path / "note.txt"
+ sample.write_text("hello\nworld\n", encoding="utf-8")
+
+ assert "read_file" in chatbot.TOOLS
+ assert chatbot.read_file(str(sample)) == "hello\nworld\n"
+
+
+def test_ask_uses_tool_name_in_callback(tmp_path):
+ sample = tmp_path / "note.txt"
+ sample.write_text("hello\nworld\n", encoding="utf-8")
+
+ class DummyLLM:
+ n_tokens = 0
+
+ def __init__(self):
+ self.calls = 0
+
+ def create_chat_completion(self, *args, **kwargs):
+ self.calls += 1
+ if self.calls == 1:
+ yield {
+ "choices": [{
+ "delta": {
+ "content": (
+ f""
+ f"{sample}"
+ )
+ }
+ }]
+ }
+ return
+ yield {"choices": [{"delta": {"content": "Final answer"}}]}
+
+ seen = []
+ llm = DummyLLM()
+ result = chatbot.ask(
+ llm,
+ [{"role": "user", "content": "read this file"}],
+ on_tool_call=lambda tool_name, target: seen.append((tool_name, target)),
+ )
+
+ assert result == "Final answer"
+ assert seen == [("read_file", str(sample))]
+
+
+def test_gui_tool_signal_accepts_name_and_target():
+ import gui
+ from PyQt6.QtWidgets import QApplication
+
+ app = QApplication.instance() or QApplication([])
+ thread = gui.AskThread(None, [])
+ thread.tool_call.emit("read_file", "C:/temp/example.py")
+ assert app is not None