version: "3.9"

services:

  redis:
    image: redis
    ports:
      - "6379:6379"

  api:
    build: .
    command: python api/main.py
    ports:
      - "8000:8000"
    depends_on:
      - redis

  scheduler:
    build: .
    command: python scheduler/scheduler.py
    depends_on:
      - redis

  worker:
    build: .
    command: python worker/worker.py
    depends_on:
      - redis
    deploy:
      replicas: 3   # 🔥 自动扩Worker

# queue/redis_queue.py
import redis
import json

r = redis.Redis(host="redis", port=6379, decode_responses=True)

QUEUE = "dlos_queue"

def push(task):
    priority = task.get("priority", 1)
    r.zadd(QUEUE, {json.dumps(task): priority})

def pop():
    items = r.zpopmax(QUEUE)
    if items:
        return items[0][0]
    return None

# api/main.py
from fastapi import FastAPI
from queue.redis_queue import push
import uuid

app = FastAPI()

@app.post("/run")
def run(data: dict):

    task = {
        "id": str(uuid.uuid4()),
        "input": data["input"],
        "priority": data.get("priority", 1),
        "retry": 0
    }

    push(task)

    return {"task_id": task["id"], "status": "queued"}

# scheduler/scheduler.py
from queue.redis_queue import pop, push
from core.planner import plan
import time
import json

def scheduler():

    while True:

        task_raw = pop()

        if not task_raw:
            time.sleep(1)
            continue

        task = json.loads(task_raw)

        steps = plan(task["input"])

        for step in steps:
            sub_task = {
                "id": task["id"] + "_" + step["id"],
                "action": step["action"],
                "depends": step.get("depends", []),
                "context": task["input"],
                "retry": 0,
                "priority": 1
            }
            push(sub_task)

if __name__ == "__main__":
    scheduler()

# worker/worker.py
from queue.redis_queue import pop, push
from core.executor import execute
from validator.validator import validate
import json
import time

MAX_RETRY = 3

def worker():

    while True:

        task_raw = pop()

        if not task_raw:
            time.sleep(1)
            continue

        task = json.loads(task_raw)

        result = execute(task, task.get("context", ""))

        if not validate(task, result):

            task["retry"] += 1

            if task["retry"] <= MAX_RETRY:
                print("Retry:", task["id"])
                push(task)
            else:
                print("FAILED:", task["id"])

            continue

        print("DONE:", task["id"], result)

if __name__ == "__main__":
    worker()

# core/executor.py
from core.tools import run_tool
from openai import OpenAI

client = OpenAI()

def run_llm(prompt):
    res = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}]
    )
    return res.choices[0].message.content

def execute(task, context):

    action = task["action"]

    tool_res = run_tool(action, context)
    if tool_res:
        return tool_res

    return run_llm(f"{action} with {context}")

# core/tools.py
TOOLS = {}

def register(name):
    def wrap(fn):
        TOOLS[name] = fn
        return fn
    return wrap

def run_tool(name, ctx):
    return TOOLS.get(name, lambda x: None)(ctx)


@register("fetch_data")
def fetch(ctx):
    return "data"

@register("analyze")
def analyze(ctx):
    return f"analysis of {ctx}"

# validator/validator.py
def validate(task, result):

    if not result:
        return False

    if len(result) < 5:
        return False

    return True

# core/planner.py
def plan(input_text):

    if "report" in input_text:
        return [
            {"id": "A", "action": "fetch_data"},
            {"id": "B", "action": "analyze", "depends": ["A"]}
        ]

    return [{"id": "A", "action": "llm_task"}]

FROM python:3.10

WORKDIR /app

COPY . .

RUN pip install fastapi uvicorn redis openai

CMD ["python", "api/main.py"]

docker-compose up --build

curl -X POST http://localhost:8000/run \
-H "Content-Type: application/json" \
-d '{"input":"write report","priority":2}'