"""
Agent-based semantic searcher using tools.
This searcher runs an LLM agent loop that can:
1. List directories to understand codebase structure
2. Run terminal commands (ripgrep, grep, etc.) to search code
3. Submit final answer with relevant code snippets
Supports multiple LLM providers:
- Claude (Anthropic)
- Gemini (Google)
"""
from __future__ import annotations
import os
import time
import uuid
from concurrent.futures import ThreadPoolExecutor, as_completed
from enum import Enum
from typing import Any, Dict, List, Optional, Tuple
from dotenv import load_dotenv
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
from langchain_core.tools import tool
import langsmith as ls
from langsmith import traceable
from langsmith.run_helpers import get_current_run_tree, trace as ls_trace
from .base import BaseSearcher, SearchItem, SearchResult
from .tools import (
ListDirInput,
RunTerminalCommandInput,
execute_list_dir,
execute_run_command,
)
# Load environment variables
load_dotenv()
class LLMProvider(str, Enum):
"""Supported LLM providers."""
CLAUDE = "claude"
GEMINI = "gemini"
SYSTEM_PROMPT = """
You are **FastContextCodeSearch**, a specialized *code retrieval subagent*.
Your ONLY job is to quickly find and return the most relevant code snippets for a query about a codebase.
You:
- Work from the repository root. All paths must be relative to the repo root.
- Use tools ONLY for searching and reading code (ListDir, RunTerminalCommand).
- Never try to solve the user's task fully — just find the right code.
- Finish by calling **SubmitAnswer exactly once**, with 1–10 high-quality snippets.
────────────────────────────────────
## 0. High-Level Behavior (like Fast Context / SWE-grep)
Treat yourself as a high-speed code search model (SWE-grep style):
- You are optimized for *rapid, parallel code retrieval*, not for long reasoning.
- You must aggressively use **parallel tool calls** and keep the number of turns small.
- You care about **precision and recall** of snippets, while avoiding irrelevant context.
### Hard Limits
- On every tool-using turn, you MUST issue **between 5 and 8 parallel tool calls**
(any combination of ListDir and RunTerminalCommand).
- After you find all information, you MUST call **SubmitAnswer** and stop.
────────────────────────────────────
## 1. Tools Overview
### ListDir(path: str = ".", max_depth: int = 1)
Use this to quickly understand the structure of the repo and locate likely code directories.
Typical usage patterns:
- Initial overview:
- ListDir(".", max_depth=1)
- Likely code dirs (in parallel):
- ListDir("src", max_depth=2)
- ListDir("lib", max_depth=2)
- ListDir("app", max_depth=2)
- ListDir("backend", max_depth=2)
- ListDir("frontend", max_depth=2)
- ListDir("server", max_depth=2)
- ListDir("client", max_depth=2)
- ListDir("packages", max_depth=2)
- ListDir("core", max_depth=2)
- ListDir("api", max_depth=2)
- ListDir("test", max_depth=2)
- ListDir("tests", max_depth=2)
Avoid wasting calls on:
- node_modules, .git, build, dist, coverage, logs, .venv, venv, .cache, etc.
### RunTerminalCommand(command: str)
Executes a shell command from repo root.
Allowed commands: **rg, grep, cat, find, head, tail, ls, wc, file**.
You may combine them with simple pipes when needed (like examples below).
Typical patterns:
- Ripgrep (preferred):
- rg "pattern" .
- rg -i "pattern" .
- rg "pattern" --type py .
- rg "pattern" path/to/dir
- rg "pattern" path/to/file.ext -n -C 8
- Grep:
- grep -rn "pattern" path/
- grep -r "pattern" .
- File discovery:
- find . -name "*foo*.py" -type f
- ls src/
- Reading code:
- cat path/to/file.py
- head -120 path/to/file.py
- tail -80 path/to/file.py
- head -200 path/to/file.py | tail -40 # approximate range
You MUST NOT run any other commands (no git, rm, curl, node, npm, python, etc.).
────────────────────────────────────
## 2. Overall Search Strategy (Turns 1–4)
You always follow this pattern:
### Turn 1 – Broad Repo Scan + Wide Search
Goal: quickly map the repo and find first candidate files.
On this turn, issue **5–8 parallel tool calls**, for example:
- Several ListDir calls to likely code dirs.
- Several wide ripgrep searches using different patterns from the query.
From the user query, extract 3–6 independent search patterns, such as:
- Exact function / class / method / variable names (if present).
- Domain terms (e.g. "payment", "checkout", "auth", "websocket", "retry", "timeout").
- Error messages, log messages, HTTP status codes, config keys.
Example call shapes (pseudocode):
- ListDir(".", max_depth=1)
- ListDir("src", max_depth=2)
- ListDir("lib", max_depth=2)
- RunTerminalCommand('rg -i "UserService" .')
- RunTerminalCommand('rg -i "user service" .')
- RunTerminalCommand('rg -i "get_user" .')
- RunTerminalCommand('rg -i "HTTP 500" .')
- RunTerminalCommand('rg -i "InternalServerError" .')
### Turn 2 – Focused / Targeted Search
Goal: zoom in on promising areas revealed on Turn 1.
On this turn, again issue **5–8 parallel RunTerminalCommand / ListDir calls**, such as:
- Targeted ripgrep within specific directories or file types:
- rg "pattern" src/feature/
- rg "pattern" --type ts src/
- rg "pattern" services/ api/
- Searches for **definitions and usages**:
- rg "class SomeName" .
- rg "def some_name" .
- rg "function someName" .
- rg "SomeName(" . # call sites
You should:
- Prefer directories and files that showed hits in Turn 1.
- Start differentiating between implementation, tests, configs, docs.
### Turn 3 – Reading Candidate Files / Extracting Snippets
Goal: read the most relevant files to extract high-quality snippets.
Again, use **5–8 tool calls**, focusing on reading:
- Use `rg ... -n -C 8` on specific files to get line numbers + local context.
- Use `cat`, `head`, `tail`, or `head | tail` combinations to see specific regions.
- Read only the parts you actually need to understand:
- Function/class definitions.
- Associated logic (e.g., helpers, private methods).
- Relevant configuration / routing / wiring code.
At this stage, you:
- Decide which 1–10 snippets you will eventually include in SubmitAnswer.
- For each snippet, identify:
- File path.
- Approximate line range (if possible).
- The exact code block(s) to include.
### Turn 4 (Optional) – Extra Connections / Sanity Checks
Only if needed and you still have uncertainty.
Goal: find related pieces:
- Usages of a function/class you already found.
- Error handlers, middleware, config flags that influence the behavior.
- Alternative implementations (e.g., different versions / platform-specific code).
Again, use **5–8 calls** focusing on:
- Additional rg/grep queries for the names you already discovered.
- Reading small additional chunks to clarify behavior.
After Turn 4 (or earlier if you are confident), you MUST stop searching and call SubmitAnswer.
────────────────────────────────────
## 3. Choosing Good Snippets
Your output is evaluated by:
- How directly it answers the query.
- How little irrelevant context it includes.
- How easy it is for another agent to use the snippets to answer the user.
### General Rules
- Prefer:
1. **Implementations** of functions/classes/APIs that directly relate to the query.
2. Relevant helper functions that are needed to understand that implementation.
3. Relevant configuration / routing / wiring.
4. Tests, ONLY if:
- They show expected behavior clearly, or
- No better implementation snippet is available.
- Avoid:
- Huge files or entire files if they are long.
- Auto-generated code, vendored libs, compiled artifacts.
- Repeated snippets of nearly identical code.
### Snippet Size & Count
- Return **1–10 snippets** in total.
- Each snippet's `content` should typically be:
- ~10–40 lines (or smaller), enough to understand the logic.
- Up to ~120 lines in extreme cases when necessary.
- It is better to have:
- A few high-signal snippets
- Than many noisy or overlapping ones.
────────────────────────────────────
## 4. SubmitAnswer Usage
When you are done searching (no later than after 4 tool-using turns), call:
SubmitAnswer(
items = [...],
reasoning = "..."
)
### items
`items` is a list of objects:
- `file_path` (str, required)
- Path to the file, relative to repo root.
- `content` (str, required)
- The extracted code snippet (exact text from the file).
- `line_start` (int, optional)
- `line_end` (int, optional)
Guidelines:
- If your search commands provide line numbers (e.g. via `rg -n`),
use them for `line_start` / `line_end` where possible.
- If you are not sure of exact line numbers, you may omit them.
- Each `content` should be a clean concatenation of the relevant code block(s),
not including command output noise.
### reasoning
`reasoning` is:
- A brief explanation (a few short paragraphs or bullet points) of:
- Why each snippet is relevant to the query.
- How the snippets fit together (if they are related).
- Do NOT repeat the code itself in `reasoning`.
- Do NOT explain the entire solution to the user — just explain why the snippets are good.
────────────────────────────────────
## 5. Parallel Tool Call Policy (CRITICAL)
You MUST aggressively parallelize tool calls:
- On EVERY tool-using turn, issue **5–8 tool calls in parallel**.
- NEVER do a single tool call if you can instead split work across multiple calls.
Examples of good parallel patterns:
1. First turn (structure + broad search):
- ListDir(".")
- ListDir("src", max_depth=2)
- ListDir("backend", max_depth=2)
- ListDir("frontend", max_depth=2)
- RunTerminalCommand('rg -i "keyword1" .')
- RunTerminalCommand('rg -i "keyword2" .')
- RunTerminalCommand('rg -i "keyword3" .')
- RunTerminalCommand('rg -i "keyword4" .')
2. Later turns (targeted search + reading):
- RunTerminalCommand('rg -n -C 8 "SomeFunction" src/')
- RunTerminalCommand('rg -n -C 8 "SomeFunction" tests/')
- RunTerminalCommand('rg -n "class SomeClass" src/')
- RunTerminalCommand('rg -n "interface SomeClass" src/')
- RunTerminalCommand('rg -n "SomeClass(" src/')
- RunTerminalCommand('cat src/module/file1.py')
- RunTerminalCommand('head -160 src/module/file2.py')
- RunTerminalCommand('head -260 src/module/file2.py | tail -80')
If you think fewer than 5 calls are needed, split your work anyway
(e.g., separate rg calls by directory, file type, or pattern variations).
────────────────────────────────────
## 6. Thinking About Symptoms vs Root Causes
When the query describes a symptom (e.g. "500 error on /checkout"):
- Search for files that define the route/endpoint/controller.
- Search for logging or error messages that match the symptom.
- Also search for:
- Middleware, interceptors, exception handlers.
- Config flags (timeouts, feature flags, toggles).
- Try to include both:
- The obvious direct code, and
- Code that likely *causes* the behavior.
────────────────────────────────────
## 7. Style of Your Own Messages
- Keep your non-tool messages very brief and focused.
- Do NOT write long natural-language explanations between tool calls.
- Never ask the user questions directly; you are a subagent invoked by another system.
- Your main value is the selection of high-quality, relevant snippets via SubmitAnswer.
"""
# Define tools using langchain's @tool decorator
@tool
def ListDir(path: str = ".", max_depth: int = 1) -> str:
"""List directory contents and get file statistics.
Args:
path: Relative path to directory (e.g., ".", "src", "src/utils"). Default is "." (repo root)
max_depth: Maximum depth to recurse (1 = immediate children only)
Returns:
Directory listing with stats
"""
return f"ListDir placeholder: {path}"
@tool
def RunTerminalCommand(command: str) -> str:
"""Execute shell commands from repo root. Commands run in the repository root directory.
Args:
command: Shell command to execute. Allowed: rg, grep, cat, find, head, tail, ls, wc, file.
Example: rg "pattern" . or cat src/main.py
Returns:
Command output
"""
return f"RunTerminalCommand placeholder: {command}"
@tool
def SubmitAnswer(items: List[Dict[str, Any]], reasoning: str) -> str:
"""Submit your final answer with found code snippets.
Args:
items: List of {file_path: str, content: str, line_start?: int, line_end?: int}
reasoning: Brief explanation of why these results are relevant
Returns:
Confirmation
"""
return "Answer submitted"
# Get tool schemas for binding
TOOLS = [ListDir, RunTerminalCommand, SubmitAnswer]
def create_llm(
provider: LLMProvider,
model: Optional[str] = None,
) -> BaseChatModel:
"""
Create an LLM instance for the specified provider.
Args:
provider: LLM provider (claude or gemini)
model: Model name (optional, uses defaults)
Returns:
LangChain chat model with tools bound
"""
if provider == LLMProvider.CLAUDE:
from langchain_anthropic import ChatAnthropic
api_key = os.getenv("CLAUDE_API_KEY") or os.getenv("ANTHROPIC_API_KEY")
if not api_key:
raise ValueError("CLAUDE_API_KEY or ANTHROPIC_API_KEY must be set")
model_name = model or "claude-sonnet-4-20250514"
llm = ChatAnthropic(
model=model_name,
api_key=api_key,
max_tokens=4096,
)
elif provider == LLMProvider.GEMINI:
from langchain_google_genai import ChatGoogleGenerativeAI
# Try different API key environment variables
api_key = (
os.getenv("GOOGLE_API_KEY") or
os.getenv("GEMINI_API_KEY") or
os.getenv("AI_STUDIO") or
os.getenv("VERTEX_AI_API_KEY")
)
if not api_key:
raise ValueError(
"GOOGLE_API_KEY, GEMINI_API_KEY, AI_STUDIO or VERTEX_AI_API_KEY must be set. "
"Get a key from https://aistudio.google.com/apikey"
)
model_name = model or "gemini-2.5-flash-lite"
llm = ChatGoogleGenerativeAI(
model=model_name,
google_api_key=api_key,
max_output_tokens=4096,
convert_system_message_to_human=True, # Gemini doesn't support system messages directly
)
else:
raise ValueError(f"Unknown provider: {provider}")
return llm.bind_tools(TOOLS)
class AgentSearcher(BaseSearcher):
"""
Agent-based searcher that uses tools to explore and search code.
Runs an LLM agent loop:
1. Agent receives query + tool results
2. Agent decides which tool to call
3. Tool is executed, result added to conversation
4. Repeat until SubmitAnswer is called or max iterations reached
"""
# Output size limits
MAX_OUTPUT_CHARS = 15000 # Max characters in tool output
MAX_OUTPUT_LINES = 300 # Max lines in tool output
def __init__(
self,
provider: LLMProvider = LLMProvider.CLAUDE,
model: Optional[str] = None,
max_iterations: int = 15,
total_timeout: float = 120.0,
llm_timeout: float = 30.0,
tool_timeout: float = 30.0,
max_output_chars: int = 15000,
max_output_lines: int = 300,
verbose: bool = False,
):
"""
Initialize the agent searcher.
Args:
provider: LLM provider to use (claude or gemini)
model: Model name (optional, uses provider defaults)
max_iterations: Maximum tool calls before forcing an answer
total_timeout: Total timeout for entire search in seconds (default: 120)
llm_timeout: Timeout for each LLM call in seconds (default: 30)
tool_timeout: Timeout for each tool execution in seconds (default: 30)
max_output_chars: Maximum characters in tool output (default: 15000)
max_output_lines: Maximum lines in tool output (default: 300)
verbose: Print agent actions to stdout
"""
self.provider = provider
self.model = model
self.max_iterations = max_iterations
self.total_timeout = total_timeout
self.llm_timeout = llm_timeout
self.tool_timeout = tool_timeout
self.max_output_chars = max_output_chars
self.max_output_lines = max_output_lines
self.verbose = verbose
# Create LLM
self.llm = create_llm(provider, model)
# Get actual model name for display
self._model_name = model or (
"claude-sonnet-4-20250514" if provider == LLMProvider.CLAUDE
else "gemini-2.5-flash-lite"
)
@property
def name(self) -> str:
return f"AgentSearcher ({self.provider.value}: {self._model_name})"
def _log(self, message: str) -> None:
"""Print message if verbose mode is enabled."""
if self.verbose:
print(f"[Agent] {message}")
def _execute_tools_parallel(
self,
tool_calls: List[Dict[str, Any]],
repo_path: str,
iteration: int,
parent_run: Optional[Any] = None,
) -> List[Tuple[str, str, Dict[str, Any], str]]:
"""
Execute multiple tool calls in parallel.
Returns list of (tool_id, tool_name, tool_args, result) tuples.
"""
results = []
# Capture parent run for passing to threads
current_parent = parent_run or get_current_run_tree()
def execute_single(tool_call: Dict[str, Any]) -> Tuple[str, str, Dict[str, Any], str]:
tool_name = tool_call["name"]
tool_args = tool_call["args"]
tool_id = tool_call.get("id", f"call_{iteration}_{tool_name}")
self._log(f"Tool (parallel): {tool_name}")
if tool_name == "RunTerminalCommand":
self._log(f" Command: {tool_args.get('command', '')[:60]}...")
elif tool_name == "ListDir":
self._log(f" Path: {tool_args.get('path', '.')}")
result = self._execute_tool(tool_name, tool_args, repo_path, current_parent)
return (tool_id, tool_name, tool_args, result)
# Use ThreadPoolExecutor for parallel execution (up to 8 parallel calls)
with ThreadPoolExecutor(max_workers=min(len(tool_calls), 8)) as executor:
futures = {
executor.submit(execute_single, tc): tc
for tc in tool_calls
}
for future in as_completed(futures):
try:
result = future.result()
results.append(result)
except Exception as e:
tc = futures[future]
tool_id = tc.get("id", "unknown")
results.append((
tool_id,
tc["name"],
tc["args"],
f"Error: {e}"
))
return results
def _validate_and_truncate_output(self, output: str, tool_name: str) -> str:
"""
Validate tool output size and truncate if too large.
Returns truncated output with a message to be more specific.
"""
lines = output.split('\n')
char_count = len(output)
line_count = len(lines)
truncated = False
truncation_reason = []
# Check line count
if line_count > self.max_output_lines:
lines = lines[:self.max_output_lines]
output = '\n'.join(lines)
truncated = True
truncation_reason.append(f"{line_count} lines")
# Check character count
if len(output) > self.max_output_chars:
output = output[:self.max_output_chars]
truncated = True
truncation_reason.append(f"{char_count} chars")
if truncated:
warning = (
f"\n\n⚠️ OUTPUT TRUNCATED (was {', '.join(truncation_reason)}). "
f"Too many results. Please use more specific search patterns:\n"
f"- Add file type filter: --type py, --type go, --type ts\n"
f"- Use more specific regex patterns\n"
f"- Search in specific directory: rg 'pattern' src/\n"
f"- Limit results: rg 'pattern' -m 10"
)
output = output + warning
self._log(f"Output truncated: {', '.join(truncation_reason)}")
return output
def _execute_tool(
self,
tool_name: str,
tool_args: Dict[str, Any],
repo_path: str,
parent_run: Optional[Any] = None,
) -> str:
"""Execute a tool and return result string."""
if tool_name == "ListDir":
output = self._execute_list_dir(tool_args, repo_path, parent_run)
elif tool_name == "RunTerminalCommand":
output = self._execute_terminal_command(tool_args, repo_path, parent_run)
elif tool_name == "SubmitAnswer":
return "__SUBMIT_ANSWER__"
else:
return f"Unknown tool: {tool_name}"
# Validate and truncate if needed
return self._validate_and_truncate_output(output, tool_name)
def _execute_list_dir(
self,
tool_args: Dict[str, Any],
repo_path: str,
parent_run: Optional[Any] = None,
) -> str:
"""Execute ListDir tool."""
with ls_trace(
name="ListDir",
run_type="tool",
inputs={"path": tool_args.get("path", ".")},
parent=parent_run,
) as run:
input_data = ListDirInput(
path=tool_args.get("path", "."),
max_depth=tool_args.get("max_depth", 1),
)
result = execute_list_dir(input_data, repo_path)
output = result.to_string()
run.outputs = {"result": output[:500]} # Truncate for trace
return output
def _execute_terminal_command(
self,
tool_args: Dict[str, Any],
repo_path: str,
parent_run: Optional[Any] = None,
) -> str:
"""Execute RunTerminalCommand tool."""
command = tool_args.get("command", "")
with ls_trace(
name="RunTerminalCommand",
run_type="tool",
inputs={"command": command[:100]},
parent=parent_run,
) as run:
input_data = RunTerminalCommandInput(
command=command,
working_dir=None, # Always run from repo root
)
result = execute_run_command(input_data, repo_path, timeout=self.tool_timeout)
output = result.to_string()
run.outputs = {"result": output[:500]} # Truncate for trace
return output
def search(
self,
query: str,
repo_path: str,
path: Optional[str] = None,
) -> SearchResult:
"""
Perform semantic search using agent loop.
"""
# Get project name from env or use default
project_name = os.getenv("LANGSMITH_PROJECT", "semantic-search")
# Run with tracing context
with ls.tracing_context(enabled=True, project_name=project_name):
return self._search_traced(query, repo_path, path)
@traceable(name="SemanticSearchAgent", run_type="chain")
def _search_traced(
self,
query: str,
repo_path: str,
path: Optional[str],
) -> SearchResult:
"""Main search method with tracing."""
start_time = time.time()
tool_time_total = 0.0 # Track time spent in tools (grep, etc.) - excluded from latency metric
# Build user message without exposing repo_path to LLM
if path:
user_message = f"""Find code that answers this question:
{query}
Search within subdirectory: {path}
Start by exploring the directory structure with ListDir("{path}"), then search for relevant code."""
else:
user_message = f"""Find code that answers this question:
{query}
Start by exploring the directory structure with ListDir("."), then search for relevant code."""
messages = [
SystemMessage(content=SYSTEM_PROMPT),
HumanMessage(content=user_message),
]
iterations = 0
tool_calls_made = []
while iterations < self.max_iterations:
# Check total timeout
elapsed = time.time() - start_time
if elapsed >= self.total_timeout:
self._log(f"Total timeout reached ({self.total_timeout}s)")
break
iterations += 1
remaining_time = self.total_timeout - elapsed
self._log(f"Iteration {iterations}/{self.max_iterations} (remaining: {remaining_time:.1f}s)")
try:
# Invoke LLM
response = self.llm.invoke(messages)
if not response.tool_calls:
self._log("No tool calls in response")
if response.content:
self._log(f"Content: {str(response.content)[:200]}...")
messages.append(response)
messages.append(HumanMessage(
content="Please use one of the available tools: ListDir, RunTerminalCommand, or SubmitAnswer."
))
continue
messages.append(response)
# Check for SubmitAnswer first (terminates the loop)
for tool_call in response.tool_calls:
if tool_call["name"] == "SubmitAnswer":
tool_args = tool_call["args"]
items_data = tool_args.get("items", [])
reasoning = tool_args.get("reasoning", "")
items = []
for item in items_data[:10]:
items.append(SearchItem(
file_path=item.get("file_path", ""),
content=item.get("content", ""),
line_start=item.get("line_start"),
line_end=item.get("line_end"),
))
total_time = (time.time() - start_time) * 1000
llm_time = total_time - tool_time_total # LLM + verification (excludes grep)
self._log(f"Done! Found {len(items)} items in {total_time:.0f}ms (LLM: {llm_time:.0f}ms, tools: {tool_time_total:.0f}ms)")
self._log(f"Reasoning: {reasoning[:100]}...")
return SearchResult(
items=items,
patterns_used=tool_calls_made,
execution_time_ms=llm_time, # Per task spec: excludes grep time
total_time_ms=total_time,
tool_time_ms=tool_time_total,
)
# Filter out SubmitAnswer, prepare other tool calls for parallel execution
tool_calls_to_execute = [
tc for tc in response.tool_calls
if tc["name"] != "SubmitAnswer"
]
# Get current run for proper trace nesting
current_run = get_current_run_tree()
# Track tool execution time (excluded from latency metric per task spec)
tool_start = time.time()
if len(tool_calls_to_execute) > 1:
# Execute multiple tool calls in parallel
self._log(f"Executing {len(tool_calls_to_execute)} tool calls in parallel")
tool_results = self._execute_tools_parallel(
tool_calls_to_execute, repo_path, iterations, current_run
)
else:
# Single tool call - execute directly
tool_results = []
for tool_call in tool_calls_to_execute:
tool_name = tool_call["name"]
tool_args = tool_call["args"]
tool_id = tool_call.get("id", f"call_{iterations}")
self._log(f"Tool: {tool_name}")
if tool_name == "RunTerminalCommand":
self._log(f" Command: {tool_args.get('command', '')[:80]}...")
elif tool_name == "ListDir":
self._log(f" Path: {tool_args.get('path', '.')}")
result = self._execute_tool(tool_name, tool_args, repo_path, current_run)
tool_results.append((tool_id, tool_name, tool_args, result))
tool_time_total += (time.time() - tool_start) * 1000 # Add to total tool time
# Add all tool results to messages
for tool_id, tool_name, tool_args, result in tool_results:
tool_calls_made.append(f"{tool_name}({str(tool_args)[:50]}...)")
messages.append(ToolMessage(
content=result,
tool_call_id=tool_id,
))
except Exception as e:
self._log(f"Error: {e}")
import traceback
traceback.print_exc()
messages.append(HumanMessage(
content=f"Error occurred: {e}\n\nPlease try a different approach or submit your answer."
))
# Max iterations - force submit
self._log("Max iterations reached, requesting final answer...")
messages.append(HumanMessage(
content="Maximum iterations reached. Please call SubmitAnswer now with whatever relevant code you have found."
))
try:
response = self.llm.invoke(messages)
if response.tool_calls:
for tool_call in response.tool_calls:
if tool_call["name"] == "SubmitAnswer":
tool_args = tool_call["args"]
items_data = tool_args.get("items", [])
items = []
for item in items_data[:10]:
items.append(SearchItem(
file_path=item.get("file_path", ""),
content=item.get("content", ""),
line_start=item.get("line_start"),
line_end=item.get("line_end"),
))
total_time = (time.time() - start_time) * 1000
llm_time = total_time - tool_time_total
return SearchResult(
items=items,
patterns_used=tool_calls_made,
execution_time_ms=llm_time,
total_time_ms=total_time,
tool_time_ms=tool_time_total,
)
except Exception as e:
self._log(f"Final error: {e}")
total_time = (time.time() - start_time) * 1000
llm_time = total_time - tool_time_total
return SearchResult(
items=[],
patterns_used=tool_calls_made,
execution_time_ms=llm_time,
total_time_ms=total_time,
tool_time_ms=tool_time_total,
error="Agent failed to produce results",
)
# Convenience classes for specific providers
class ClaudeAgentSearcher(AgentSearcher):
"""Agent searcher using Claude."""
def __init__(
self,
model: str = "claude-sonnet-4-20250514",
max_iterations: int = 15,
total_timeout: float = 120.0,
llm_timeout: float = 30.0,
tool_timeout: float = 30.0,
verbose: bool = False,
):
super().__init__(
provider=LLMProvider.CLAUDE,
model=model,
max_iterations=max_iterations,
total_timeout=total_timeout,
llm_timeout=llm_timeout,
tool_timeout=tool_timeout,
verbose=verbose,
)
class GeminiAgentSearcher(AgentSearcher):
"""Agent searcher using Gemini (default: gemini-2.5-flash-lite)."""
def __init__(
self,
model: str = "gemini-2.5-flash-lite",
max_iterations: int = 15,
total_timeout: float = 120.0,
llm_timeout: float = 30.0,
tool_timeout: float = 30.0,
verbose: bool = False,
):
super().__init__(
provider=LLMProvider.GEMINI,
model=model,
max_iterations=max_iterations,
total_timeout=total_timeout,
llm_timeout=llm_timeout,
tool_timeout=tool_timeout,
verbose=verbose,
)
class GeminiFlashLiteSearcher(AgentSearcher):
"""Agent searcher using Gemini Flash Lite (fastest, cheapest)."""
def __init__(
self,
max_iterations: int = 15,
total_timeout: float = 120.0,
llm_timeout: float = 30.0,
tool_timeout: float = 30.0,
verbose: bool = False,
):
super().__init__(
provider=LLMProvider.GEMINI,
model="gemini-2.5-flash-lite",
total_timeout=total_timeout,
llm_timeout=llm_timeout,
tool_timeout=tool_timeout,
max_iterations=max_iterations,
verbose=verbose,
)
class GeminiFlashSearcher(AgentSearcher):
"""Agent searcher using Gemini Flash (balanced speed/quality)."""
def __init__(
self,
max_iterations: int = 15,
total_timeout: float = 120.0,
llm_timeout: float = 30.0,
tool_timeout: float = 30.0,
verbose: bool = False,
):
super().__init__(
provider=LLMProvider.GEMINI,
model="gemini-2.5-flash",
max_iterations=max_iterations,
total_timeout=total_timeout,
llm_timeout=llm_timeout,
tool_timeout=tool_timeout,
verbose=verbose,
)
class GeminiProSearcher(AgentSearcher):
"""Agent searcher using Gemini 3 Pro Preview (highest quality)."""
def __init__(
self,
max_iterations: int = 15,
total_timeout: float = 180.0,
llm_timeout: float = 60.0,
tool_timeout: float = 30.0,
verbose: bool = False,
):
super().__init__(
provider=LLMProvider.GEMINI,
model="gemini-3-pro-preview",
max_iterations=max_iterations,
total_timeout=total_timeout,
llm_timeout=llm_timeout,
tool_timeout=tool_timeout,
verbose=verbose,
)
