INFINA-RD
refactor: DI infrastructure, service decomposition, repository helpers, test suite
e066621
import argparse
import sys
from pathlib import Path
from agent.graph import AgentRunner
from agent.states import AgentConfig, ModelBackend
COMMAND_HELP = """\
Commands:
:help / :h Show this message.
:context Refresh and display the project summary.
:clear Reset the conversation history (system instructions stay).
:history Print the last few exchanges.
:exit / :quit Leave the CLI.
"""
def parse_backend(value: str) -> ModelBackend:
try:
return ModelBackend(value.lower())
except ValueError as exc:
valid = ", ".join(b.value for b in ModelBackend)
raise argparse.ArgumentTypeError(f"Backend must be one of: {valid}") from exc
def resolve_project_directory(arg_dir: str | None, default_dir: Path) -> str:
if arg_dir:
project_directory_input = arg_dir.strip()
else:
prompt = (
f"Enter the absolute or relative path to your project root "
f"(press Enter for default: {default_dir}): "
)
user_value = input(prompt).strip()
project_directory_input = user_value or str(default_dir)
return str(Path(project_directory_input).expanduser().resolve())
def build_agent(args, project_directory: str) -> AgentRunner:
config = AgentConfig(
project_directory=project_directory,
backend=args.backend if isinstance(args.backend, ModelBackend) else parse_backend(args.backend),
model=args.model,
temperature=args.temperature,
recursion_limit=args.recursion_limit,
auto_context=not args.no_auto_context,
)
return AgentRunner(config)
def handle_command(command: str, runner: AgentRunner) -> bool:
cmd = command.lower()
if cmd in {":help", ":h"}:
print(COMMAND_HELP)
return True
if cmd in {":exit", ":quit"}:
print("Exiting...")
sys.exit(0)
if cmd == ":clear":
runner.reset_history()
print("Conversation cleared.")
return True
if cmd == ":context":
summary = runner.refresh_summary()
print("\n--- Workspace summary ---")
print(summary)
print("-------------------------\n")
return True
if cmd == ":history":
history = runner.conversation_history()
if not history:
print("No conversation history yet.")
return True
print("\n--- Conversation history ---")
for message in history:
role = message.get("role", "").upper()
if role == "SYSTEM":
continue
content = message.get("content", "")
print(f"[{role}]\n{content}\n")
print("-----------------------------\n")
return True
return False
def main():
default_project_dir = Path.cwd() / "generated_project"
parser = argparse.ArgumentParser(description="Gemini/Qwen-style engineering agent CLI")
parser.add_argument("--recursion-limit", "-r", type=int, default=100,
help="Recursion limit for LangGraph (default: 100)")
parser.add_argument("--project-directory", "-p", default=None,
help="Absolute or relative path to the project root. Prompts interactively when omitted.")
parser.add_argument("--backend", "-b", default=ModelBackend.GROQ.value, type=parse_backend,
help="LLM backend: groq, gemini, openrouter (default: groq)")
parser.add_argument("--model", "-m", default=None, help="Override the LLM model name for the selected backend.")
parser.add_argument("--temperature", "-t", type=float, default=0.2,
help="Sampling temperature for the agent (default: 0.2)")
parser.add_argument("--no-auto-context", action="store_true",
help="Disable automatic summarize_project calls before each request.")
args = parser.parse_args()
project_directory = resolve_project_directory(args.project_directory, default_project_dir)
runner = build_agent(args, project_directory)
print("\nProject Engineering Agent (Gemini/Qwen style)")
print("Type your request and press Enter. Use :help for CLI commands.\n")
while True:
try:
user_prompt = input("agent> ").strip()
except KeyboardInterrupt:
print("\nInterrupted. Type :exit to leave.")
continue
if not user_prompt:
continue
if user_prompt.lower() in ("exit", "quit", "q"):
print("Exiting...")
break
if user_prompt.startswith(":") and handle_command(user_prompt, runner):
continue
try:
result_state = runner.invoke(user_prompt)
final_message = result_state["messages"][-1] if result_state.get("messages") else {}
response = final_message.get("content", "(no response)")
print(f"\n{response}\n")
except Exception as exc:
print(f"Error: {exc}", file=sys.stderr)
sys.exit(0)
if __name__ == "__main__":
main()