44 lines
1.8 KiB
Python
44 lines
1.8 KiB
Python
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"} |