async def process(state: dict) -> dict: """ This node retrieves the source code of a repository given its URL. It uses an LLM to determine the appropriate steps to clone the repository and provides the local path where the repository is stored. Input: state["repository_url"]: The URL of the repository to retrieve. Output: Returns a dictionary with: - "local_repository_path": The local path where the repository is stored. - "phase": "complete" on success, or "failed" on error. - "error": Error message if an exception occurs. """ from app.agent import llm from langchain_core.messages import SystemMessage, HumanMessage import os import subprocess try: repository_url = state.get("repository_url", "") if not repository_url: raise ValueError("Missing 'repository_url' in input state.") # Use LLM to determine the appropriate steps to clone the repository messages = [ SystemMessage(content="You are a helpful assistant that provides instructions for retrieving repository source code."), HumanMessage(content=f"How do I clone the repository from this URL: {repository_url}?") ] response = await llm.ainvoke(messages) instructions = response.content.strip() # Execute the instructions to clone the repository local_path = os.path.join(os.getcwd(), "cloned_repository") if not os.path.exists(local_path): os.makedirs(local_path) clone_command = f"git clone {repository_url} {local_path}" subprocess.run(clone_command, shell=True, check=True) return {"local_repository_path": local_path, "phase": "complete"} except Exception as exc: return {"error": str(exc), "phase": "failed"}