发现优质的 AI Agent 技能
聚合 Claude Skills、LangChain、AutoGPT 等优质资源,助力开发者快速构建智能应用
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()
Client waits for entire response
response = call_llm(prompt) return response ``` **Correct (streaming results):**
Polling is inefficient and not durable
while True: status = check_payment_status() if status == "paid": break time.sleep(1)
Client code to read events
@app.post("/start") def start_workflow(): handle = DBOS.start_workflow(long_workflow) return {"workflow_id": handle.get_workflow_id()} @app.get("/status/{workflow_id}")
Connection leaked - no destroy()!
``` **Correct (with cleanup):**
Missing workflow_name and queue_name!
handle = client.enqueue({}, task_data) ``` **Correct (with required options):**