Spaces:
Sleeping
Sleeping
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() | |