发现优质的 AI Agent 技能
聚合 Claude Skills、LangChain、AutoGPT 等优质资源，助力开发者快速构建智能应用
One user blocks all other users!
queue.enqueue(process_task, task) ``` **Correct (per-user limits with partitioning):**
Every worker processes both queues
if __name__ == "__main__": DBOS(config=config) DBOS.launch() ``` **Correct (workers listen to specific queues):**
Multiple requests = multiple workflows for same user!
queue.enqueue(process_workflow, user_id) ``` **Correct (deduplicated by user):**
Uses lots of memory
pass ``` **Correct (worker concurrency):**
Starting many workflows without control
for task in tasks: DBOS.start_workflow(process_task, task) # Could overwhelm resources ``` **Correct (using queue):**
Regular sleep is lost on restart!
time.sleep(delay_seconds) run_task(task) ``` **Correct (durable sleep):**
Don't use external cron or manual timers
import schedule schedule.every(1).minute.do(my_task) ``` **Correct (DBOS scheduled workflow):**
Multiple clicks = multiple payments!
handle = DBOS.start_workflow(payment_workflow, order_id) return handle.get_result() ``` **Correct (idempotent with workflow ID):**
Expensive processing
analyze(user_input) @app.post("/input") def on_input(user_id: str, input: str): DBOS.start_workflow(process_input, input) ```
instance_name must be unique and passed to super()
super().__init__(instance_name=url) @DBOS.workflow() def fetch_workflow(self): return self.fetch_url()
Don't configure at module level!
config: DBOSConfig = {"name": "my-app"} DBOS(config=config) @app.get("/") @DBOS.workflow() def endpoint():
Don't configure at module level!
config: DBOSConfig = { "name": "my-app", } DBOS(config=config) @DBOS.workflow()