DSCI

**File structure** ``` .dsci/ │ ├─ jobs.yaml │ ├─ job_setup/ │ ├─ config.yaml │ ├─ job.py │ └─ tasks/ │ ├─ install_deps/ │ │ └─ task.py │ └─ download_model/ │ └─ task.py │ ├─ job_deploy/ │ ├─ config.yaml │ ├─ job.py │ └─ tasks/ │ └─ start_agent/ │ └─ task.py │ └─ job_test/ ├─ job.bash └─ tasks/ └─ test_endpoint/ └─ task.bash ``` --- ## 1️⃣ `jobs.yaml` – only the pipeline definition ```yaml # .dsci/jobs.yaml jobs: - id: setup path: job_setup/ - id: deploy path: job_deploy/ - id: test path: job_test/ ``` *No matrix, no extra options – just three sequential jobs.* --- ## 2️⃣ Job **setup** – install dependencies and (optionally) download a tiny local model ### `job_setup/config.yaml` – default parameters for the job ```yaml # .dsci/job_setup/config.yaml # Python packages required for the agent requirements: - "langchain==0.1.9" - "uvicorn[standard]==0.27.0" - "fastapi==0.111.0" - "openai==1.30.1" # if you want to hit OpenAI API - "requests==2.32.3" # URL of a very small GGML model (e.g. tiny‑llama) – optional model_url: "https://huggingface.co/ggml-org/ggml-tiny-llama/resolve/main/ggml-tiny-llama.bin" # Where the model will be stored (inside the DSCI artifacts folder) model_path: "${HOME}/artifacts/tiny-llama.bin" ``` ### `job_setup/job.py` – orchestrates the two tasks ```python #!/usr/bin/env python3 # .dsci/job_setup/job.py # ------------------------------------------------- # 1️⃣ Install python requirements # 2️⃣ (Optional) download a tiny GGML model # ------------------------------------------------- run_task('install_deps') run_task('download_model') ``` ### `job_setup/tasks/install_deps/task.py` – pip‑install everything ```python #!/usr/bin/env python3 # .dsci/job_setup/tasks/install_deps/task.py import subprocess import sys # ----------------------------------------------------------------- # Read the list of packages from the job configuration # ----------------------------------------------------------------- cfg = config() requirements = cfg.get('requirements', []) if not requirements: print("⚠️ No requirements defined – skipping pip install") else: # Build a single pip command (more reliable than many calls) pip_cmd = [sys.executable, "-m", "pip", "install", "--quiet"] + requirements print(f"🔧 Installing python packages: {', '.join(requirements)}") subprocess.check_call(pip_cmd) # Remember that deps were installed – useful for downstream jobs update_state({"deps_installed": True}) ``` ### `job_setup/tasks/download_model/task.py` – fetch a tiny GGML model (optional) ```python #!/usr/bin/env python3 # .dsci/job_setup/tasks/download_model/task.py import os import pathlib import urllib.request cfg = config() model_url = cfg.get("model_url") model_path = cfg.get("model_path") # e.g. ${HOME}/artifacts/tiny-llama.bin if not model_url: print("⚠️ No model URL defined – skipping model download") update_state({"model_downloaded": False}) else: # Ensure the artifacts directory exists artifacts_dir = pathlib.Path.home() / "artifacts" artifacts_dir.mkdir(parents=True, exist_ok=True) # Resolve $HOME in the path model_path = os.path.expandvars(model_path) print(f"⬇️ Downloading model from {model_url}") urllib.request.urlretrieve(model_url, model_path) print(f"✅ Model saved to {model_path}") # Export the absolute path so later jobs can use it update_state({ "model_downloaded": True, "model_path": model_path }) ``` --- ## 3️⃣ Job **deploy** – spin‑up a FastAPI server that runs a LangChain agent ### `job_deploy/config.yaml` – runtime parameters for the agent ```yaml # .dsci/job_deploy/config.yaml # Port on which the agent will be exposed port: 8000 # If you want to use OpenAI instead of a local GGML model, set these: openai_api_key: "" # leave empty for local model openai_model: "gpt-4o-mini" # When using the tiny‑llama GGML model (downloaded by the previous job) # the task will read the path from the shared state. ``` ### `job_deploy/job.py` – single task that starts the server ```python #!/usr/bin/env python3 # .dsci/job_deploy/job.py # ------------------------------------------------- # Start a FastAPI + LangChain agent. # The server runs in the background (detached) so the job can finish # while the process keeps listening on localhost. # ------------------------------------------------- run_task('start_agent') ``` ### `job_deploy/tasks/start_agent/task.py` – the actual agent code ```python #!/usr/bin/env python3 # .dsci/job_deploy/tasks/start_agent/task.py import os import pathlib import subprocess import sys import time import json # ----------------------------------------------------------------- # 1️⃣ Gather configuration # ----------------------------------------------------------------- cfg = config() port = cfg.get("port", 8000) openai_key = cfg.get("openai_api_key", "").strip() openai_model = cfg.get("openai_model", "gpt-4o-mini") # ----------------------------------------------------------------- # 2️⃣ Retrieve the model path from the previous job (if any) # ----------------------------------------------------------------- state = config() # whole pipeline state model_path = state.get("_dsci_", {}) \ .get("setup", {}) \ .get("model_path") # may be None use_local_model = bool(model_path and pathlib.Path(model_path).exists()) # ----------------------------------------------------------------- # 3️⃣ Build a tiny FastAPI app on‑the‑fly # ----------------------------------------------------------------- app_code = pathlib.Path("/tmp/agent_app.py") app_code.write_text(f""" import os from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Any app = FastAPI() class Prompt(BaseModel): text: str # ----------------------------------------------------------------- # LangChain / OpenAI or local GGML setup # ----------------------------------------------------------------- if {json.dumps(use_local_model)}: # ---------- LOCAL GGML MODEL ---------- # Very simple placeholder – replace with your own inference wrapper # For demo purposes we just echo the prompt. def _run(prompt: str) -> str: return f"[LOCAL MODEL] {{prompt}}" else: # ---------- OPENAI ---------- import openai openai.api_key = {json.dumps(openai_key)} def _run(prompt: str) -> str: resp = openai.ChatCompletion.create( model={json.dumps(openai_model)}, messages=[{{"role": "user", "content": prompt}}] ) return resp.choices[0].message.content @app.post("/run") def run_prompt(p: Prompt) -> Any: try: answer = _run(p.text) return {{"answer": answer}} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) """) # ----------------------------------------------------------------- # 4️⃣ Launch the server in background (detached) # ----------------------------------------------------------------- # We keep a tiny “ready” file to signal that the service is up ready_file = pathlib.Path(os.path.expanduser("~/artifacts/agent_ready")) if ready_file.exists(): ready_file.unlink() # Use `nohup` + `&` so the job can exit while the server lives cmd = [ sys.executable, "-m", "uvicorn", "agent_app:app", "--host", "127.0.0.1", "--port", str(port), "--log-level", "error" ] print(f"🚀 Starting FastAPI agent on http://127.0.0.1:{port}") proc = subprocess.Popen( cmd, cwd="/tmp", stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, preexec_fn=os.setsid # detach from parent ) # Wait a couple of seconds for the server to bind for _ in range(10): try: import socket with socket.create_connection(("127.0.0.1", port), timeout=1): break except OSError: time.sleep(0.5) else: print("❌ Agent failed to start") sys.exit(1) # Touch the ready‑file so downstream jobs can see the service ready_file.touch() print("✅ Agent is ready") # Export useful state for later jobs (pid & port) update_state({ "agent_pid": proc.pid, "agent_port": port, "agent_ready_file": str(ready_file) }) ``` *The task creates a temporary `agent_app.py` (FastAPI) and launches it with `uvicorn`. It stores the PID, port and a “ready” flag in the DSCI state – downstream jobs can read those values.* --- ## 4️⃣ Job **test** – sanity‑check the running agent ### `job_test/job.bash` – simply run the single test task ```bash #!/usr/bin/env bash # .dsci/job_test/job.bash run_task "test_endpoint" ``` ### `job_test/tasks/test_endpoint/task.bash` – call the `/run` endpoint ```bash #!/usr/bin/env bash # .dsci/job_test/tasks/test_endpoint/task.bash # ----------------------------------------------------------------- # Grab configuration / state # ----------------------------------------------------------------- PORT=$(config port) # from job_deploy/config.yaml (inherited) READY_FILE=$(config agent_ready_file) # exported by the deploy job # Wait until the ready file appears (max 15 seconds) SECONDS_WAITED=0 while [[ ! -f "$READY_FILE" && $SECONDS_WAITED -lt 15 ]]; do echo "⏳ Waiting for agent to become ready..." sleep 1 ((SECONDS_WAITED++)) done if [[ ! -f "$READY_FILE" ]]; then echo "❌ Agent did not signal readiness – aborting test" exit 1 fi # ----------------------------------------------------------------- # Perform a simple request # ----------------------------------------------------------------- PROMPT="What is the current year?" RESPONSE=$(curl -s -X POST "http://127.0.0.1:${PORT}/run" \ -H "Content-Type: application/json" \ -d "$(printf '{"text":"%s"}' "$PROMPT")") echo "🗣️ Prompt sent: $PROMPT" echo "📨 Raw response: $RESPONSE" # Extract the answer field (jq is available in the default container) ANSWER=$(echo "$RESPONSE" | jq -r '.answer // "❓ no answer"') echo "✅ Agent answered: $ANSWER" ``` *The Bash task reads the port from the shared configuration