发现优质的 AI Agent 技能
聚合 Claude Skills、LangChain、AutoGPT 等优质资源，助力开发者快速构建智能应用
Random number in workflow breaks recovery!
choice = random.randint(0, 1) if choice == 0: step_one() else: step_two()
Never start workflows from inside a step!
DBOS.start_workflow(another_workflow) ``` **Incorrect (modifying global state):**
Long running work
pass thread = threading.Thread(target=long_task, args=(data,)) thread.start() ```
DBOS state from previous test!
result = another_workflow() ``` **Correct (reset fixture):**
For Postgres, use transactions instead of steps
engine.execute("INSERT INTO table VALUES (?)", data) ``` **Correct (using transaction):**
Manual retry logic is error-prone
for attempt in range(3): try: return requests.get("https://api.example.com").json() except Exception: if attempt == 2:
All tasks treated equally - urgent tasks may wait
for task in tasks: queue.enqueue(process_task, task) ``` **Correct (with priority):**
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):**