DEV Community: DBOS, Inc.

Build a Reliable Hacker News Deep Research AI Agent

Qian Li — Thu, 17 Jul 2025 00:04:24 +0000

In this example, we use DBOS to build an AI deep research agent that autonomously searches Hacker News for information on any topic.

This example demonstrates how to build reliable, durable AI agents with DBOS. The agent starts with a research topic, autonomously searches for related information, makes decisions about when to continue research, and synthesizes findings into a comprehensive report. Because the agent is implemented as a DBOS durable workflow, it can automatically recover from any failure and continue research from where it left off, ensuring no work is lost.

This example also demonstrates how easy it is to add DBOS to an existing agentic application. Adding DBOS to this agent to make it reliable and observable required changing <20 lines of code. All you have to do is annotate workflows and steps.

All source code is available on GitHub.

Main Research Workflow

The core of the agent is the main research workflow. It starts with a topic and autonomously explores related queries until it has enough information, then synthesizes a final report.

@DBOS.workflow()
def agentic_research_workflow(topic: str, max_iterations: int) -> Dict[str, Any]:
    """Main agentic workflow that autonomously researches a topic.

    This demonstrates a complete agentic workflow using DBOS.
    The agent starts with a research topic then:
    1. Searches Hacker News for information on that topic.
    2. Iteratively searches related queries, collecting information.
    3. Makes decisions about when to continue
    4. Synthesizes findings into a final report.

    The entire process is durable and can recover from any failure.
    """

    all_findings = []
    research_history = []
    current_iteration = 0
    current_query = topic

    # Main agentic research loop
    while current_iteration < max_iterations:
        current_iteration += 1

        # Research the next query in a child workflow
        iteration_result = research_query(topic, current_query, current_iteration)
        research_history.append(iteration_result)
        all_findings.append(iteration_result["evaluation"])

        # Handle cases where no results are found
        stories_found = iteration_result["stories_found"]
        if stories_found == 0:

            # Generate alternative queries when hitting dead ends
            alternative_query = generate_follow_ups_step(
                topic, all_findings, current_iteration
            )
            if alternative_query:
                current_query = alternative_query
                continue

        # Evaluate whether to continue research
        should_continue = should_continue_step(
            topic, all_findings, current_iteration, max_iterations
        )
        if not should_continue:
            break

        # Generate next research question based on findings
        if current_iteration < max_iterations:
            follow_up_query = generate_follow_ups_step(
                topic, all_findings, current_iteration
            )
            if follow_up_query:
                current_query = follow_up_query

    # Final step: Synthesize all findings into comprehensive report
    final_report = synthesize_findings_step(topic, all_findings)

    # Return complete research results
    return {
        "topic": topic,
        "total_iterations": current_iteration,
        "max_iterations": max_iterations,
        "research_history": research_history,
        "final_report": final_report,
        "summary": {
            "total_stories": sum(r["stories_found"] for r in research_history),
            "total_comments": sum(r["comments_analyzed"] for r in research_history),
            "queries_executed": [r["query"] for r in research_history],
            "avg_relevance": (
                sum(f.get("relevance_score", 0) for f in all_findings)
                / len(all_findings)
                if all_findings
                else 0
            ),
        },
    }

Research Query Workflow

Each iteration of the main research workflow calls a child workflow that searches Hacker News for information about a query, then evaluates and returns its findings.

@DBOS.workflow()
def research_query(topic: str, query: str, iteration: int) -> Dict[str, Any]:
    """Research a query selected by the main agentic workflow."""

    # Step 1: Search Hacker News for stories about the topic
    stories = search_hackernews_step(query, max_results=30)

    # Step 2: Gather comments from all stories found
    comments = []
    if stories:
        for i, story in enumerate(stories):
            story_id = story.get("objectID")
            title = story.get("title", "Unknown")[:50]
            num_comments = story.get("num_comments", 0)

            if story_id and num_comments > 0:
                story_comments = get_comments_step(story_id, max_comments=10)
                comments.extend(story_comments)

    # Step 3: Evaluate gathered data and return findings
    evaluation = evaluate_results_step(topic, query, stories, comments)

    return {
        "iteration": iteration,
        "query": query,
        "stories_found": len(stories),
        "comments_analyzed": len(comments),
        "evaluation": evaluation,
        "stories": stories,
        "comments": comments,
    }

Agent Decision-Making Steps

The agent's intelligence comes from three key step functions that handle decision-making:

Agent Evaluation Step

@DBOS.step()
def evaluate_results_step(
    topic: str,
    query: str,
    stories: List[Dict[str, Any]],
    comments: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    """Agent evaluates search results and extracts insights."""

    # Prepare content for analysis
    content_summary = f"Found {len(stories)} stories"
    if comments:
        content_summary += f" and {len(comments)} comments"

    # Create detailed content digest for LLM
    stories_text = ""
    top_stories = []

    # Evaluate only the top 10 most relevant (per HN search) stories
    for i, story in enumerate(stories[:10]):
        title = story.get("title", "No title")
        url = story.get("url", "No URL")
        hn_url = f"https://news.ycombinator.com/item?id={story.get('objectID', '')}"
        points = story.get("points", 0)
        num_comments = story.get("num_comments", 0)
        author = story.get("author", "Unknown")

        stories_text += f"Story {i+1}:\n"
        stories_text += f"  Title: {title}\n"
        stories_text += f"  Points: {points}, Comments: {num_comments}\n"
        stories_text += f"  URL: {url}\n"
        stories_text += f"  HN Discussion: {hn_url}\n"
        stories_text += f"  Author: {author}\n\n"

        # Store top stories for reference
        top_stories.append(
            {
                "title": title,
                "url": url,
                "hn_url": hn_url,
                "points": points,
                "num_comments": num_comments,
                "author": author,
                "objectID": story.get("objectID", ""),
            }
        )

    comments_text = ""

    if comments:
        for i, comment in enumerate(comments[:20]):  # Limit to top 20 comments
            comment_text = comment.get("comment_text", "")
            if comment_text:
                author = comment.get("author", "Unknown")
                # Get longer excerpts for better analysis
                excerpt = (
                    comment_text[:400] + "..."
                    if len(comment_text) > 400
                    else comment_text
                )

                comments_text += f"Comment {i+1}:\n"
                comments_text += f"  Author: {author}\n"
                comments_text += f"  Text: {excerpt}\n\n"

    prompt = f"""
    You are a research agent evaluating search results for: {topic}

    Query used: {query}

    Stories found:
    {stories_text}

    Comments analyzed:
    {comments_text}

    Provide a DETAILED analysis with specific insights, not generalizations. Focus on:
    - Specific technical details, metrics, or benchmarks mentioned
    - Concrete tools, libraries, frameworks, or techniques discussed
    - Interesting problems, solutions, or approaches described
    - Performance data, comparison results, or quantitative insights
    - Notable opinions, debates, or community perspectives
    - Specific use cases, implementation details, or real-world examples

    Return JSON with:
    - "insights": Array of specific, technical insights with context
    - "relevance_score": Number 1-10
    - "summary": Brief summary of findings
    - "key_points": Array of most important points discovered
    """

    messages = [
        {
            "role": "system",
            "content": "You are a research evaluation agent. Analyze search results and provide structured insights in JSON format.",
        },
        {"role": "user", "content": prompt},
    ]

    response = call_llm(messages, max_tokens=2000)

    try:
        cleaned_response = clean_json_response(response)
        evaluation = json.loads(cleaned_response)
        evaluation["query"] = query
        evaluation["top_stories"] = top_stories
        return evaluation
    except json.JSONDecodeError:
        return {
            "insights": [f"Found {len(stories)} stories about {topic}"],
            "relevance_score": 7,
            "summary": f"Basic search results for {query}",
            "key_points": [],
            "query": query,
        }

Follow-up Query Generation Step

@DBOS.step()
def generate_follow_ups_step(
    topic: str, current_findings: List[Dict[str, Any]], iteration: int
) -> Optional[str]:
    """Agent generates follow-up research queries based on current findings."""

    findings_summary = ""
    for finding in current_findings:
        findings_summary += f"Query: {finding.get('query', 'Unknown')}\n"
        findings_summary += f"Summary: {finding.get('summary', 'No summary')}\n"
        findings_summary += f"Key insights: {finding.get('insights', [])}\n"
        findings_summary += (
            f"Unanswered questions: {finding.get('unanswered_questions', [])}\n\n"
        )

    prompt = f"""
    You are a research agent investigating: {topic}

    This is iteration {iteration} of your research.

    Current findings:
    {findings_summary}

    Generate 2-4 SHORT KEYWORD-BASED search queries for Hacker News that explore DIVERSE aspects of {topic}.

    CRITICAL RULES:
    1. Use SHORT keywords (2-4 words max) - NOT long sentences
    2. Focus on DIFFERENT aspects of {topic}, not just one narrow area
    3. Use terms that appear in actual Hacker News story titles
    4. Avoid repeating previous focus areas
    5. Think about what tech people actually discuss about {topic}

    For {topic}, consider diverse areas like:
    - Performance/optimization
    - Tools/extensions
    - Comparisons with other technologies
    - Use cases/applications
    - Configuration/deployment
    - Recent developments

    GOOD examples: ["postgres performance", "database tools", "sql optimization"]
    BAD examples: ["What are the best practices for PostgreSQL optimization?"]

    Return only a JSON array of SHORT keyword queries: ["query1", "query2", "query3"]
    """

    messages = [
        {
            "role": "system",
            "content": "You are a research agent. Generate focused follow-up queries based on current findings. Return only JSON array.",
        },
        {"role": "user", "content": prompt},
    ]

    response = call_llm(messages)

    try:
        cleaned_response = clean_json_response(response)
        queries = json.loads(cleaned_response)
        return queries[0] if isinstance(queries, list) and len(queries) > 0 else None
    except json.JSONDecodeError:
        return None

Continuation Decision Step

@DBOS.step()
def should_continue_step(
    topic: str,
    all_findings: List[Dict[str, Any]],
    current_iteration: int,
    max_iterations: int,
) -> bool:
    """Agent decides whether to continue research or conclude."""

    if current_iteration >= max_iterations:
        return {
            "should_continue": False,
            "reason": f"Reached maximum iterations ({max_iterations})",
        }

    # Analyze findings completeness
    findings_summary = ""
    total_relevance = 0
    for finding in all_findings:
        findings_summary += f"Query: {finding.get('query', 'Unknown')}\n"
        findings_summary += f"Summary: {finding.get('summary', 'No summary')}\n"
        findings_summary += f"Relevance: {finding.get('relevance_score', 5)}/10\n"
        total_relevance += finding.get("relevance_score", 5)

    avg_relevance = total_relevance / len(all_findings) if all_findings else 0

    prompt = f"""
    You are a research agent investigating: {topic}

    Current iteration: {current_iteration}/{max_iterations}

    Findings so far:
    {findings_summary}

    Average relevance score: {avg_relevance:.1f}/10

    Decide whether to continue research or conclude. PRIORITIZE THOROUGH EXPLORATION - continue if:
    1. Current iteration is less than 75% of max_iterations
    2. Average relevance is above 6.0 and there are likely unexplored aspects
    3. Recent queries found significant new information
    4. The research seems to be discovering diverse perspectives on the topic

    Only stop early if:
    - Average relevance is below 5.0 for multiple iterations
    - No new meaningful information in the last 2 iterations
    - Research appears to be hitting diminishing returns

    Return JSON with:
    - "should_continue": boolean
    """

    messages = [
        {
            "role": "system",
            "content": "You are a research decision agent. Evaluate research completeness and decide whether to continue. Return JSON.",
        },
        {"role": "user", "content": prompt},
    ]

    response = call_llm(messages)

    try:
        cleaned_response = clean_json_response(response)
        decision = json.loads(cleaned_response)
        return decision.get("should_continue", True)
    except json.JSONDecodeError:
        return True

Search API Steps

After deciding what terms to search for, the agent calls these steps to retrieve stories and comments from Hacker News.

Hacker News API Steps

@DBOS.step()
def search_hackernews_step(query: str, max_results: int = 20) -> List[Dict[str, Any]]:
    """Search Hacker News stories using Algolia API."""
    params = {"query": query, "hitsPerPage": max_results, "tags": "story"}

    with httpx.Client(timeout=30.0) as client:
        response = client.get("https://hn.algolia.com/api/v1/search", params=params)
        response.raise_for_status()
        return response.json()["hits"]

@DBOS.step()
def get_comments_step(story_id: str, max_comments: int = 50) -> List[Dict[str, Any]]:
    """Get comments for a specific Hacker News story."""
    params = {"tags": f"comment,story_{story_id}", "hitsPerPage": max_comments}

    with httpx.Client(timeout=30.0) as client:
        response = client.get("https://hn.algolia.com/api/v1/search", params=params)
        response.raise_for_status()
        return response.json()["hits"]

Synthesize Findings Step

Finally, after concluding its research, the agentic workflow calls this step to synthesize its findings into a report.

Synthesize Findings Step

@DBOS.step()
def synthesize_findings_step(
    topic: str, all_findings: List[Dict[str, Any]]
) -> Dict[str, Any]:
    """Synthesize all research findings into a comprehensive report."""
    findings_text = ""
    story_links = []

    for i, finding in enumerate(all_findings, 1):
        findings_text += f"\n=== Finding {i} ===\n"
        findings_text += f"Query: {finding.get('query', 'Unknown')}\n"
        findings_text += f"Summary: {finding.get('summary', 'No summary')}\n"
        findings_text += f"Key Points: {finding.get('key_points', [])}\n"
        findings_text += f"Insights: {finding.get('insights', [])}\n"

        # Extract story links and details for reference
        if finding.get("top_stories"):
            for story in finding["top_stories"]:
                story_links.append(
                    {
                        "title": story.get("title", "Unknown"),
                        "url": story.get("url", ""),
                        "hn_url": f"https://news.ycombinator.com/item?id={story.get('objectID', '')}",
                        "points": story.get("points", 0),
                        "comments": story.get("num_comments", 0),
                    }
                )

    # Build comprehensive story and citation data
    story_citations = {}
    citation_id = 1

    for finding in all_findings:
        if finding.get("top_stories"):
            for story in finding["top_stories"]:
                story_id = story.get("objectID", "")
                if story_id and story_id not in story_citations:
                    story_citations[story_id] = {
                        "id": citation_id,
                        "title": story.get("title", "Unknown"),
                        "url": story.get("url", ""),
                        "hn_url": story.get("hn_url", ""),
                        "points": story.get("points", 0),
                        "comments": story.get("num_comments", 0),
                    }
                    citation_id += 1

    # Create citation references text
    citations_text = "\n".join(
        [
            f"[{cite['id']}] {cite['title']} ({cite['points']} points, {cite['comments']} comments) - {cite['hn_url']}"
            + (f" - {cite['url']}" if cite["url"] else "")
            for cite in story_citations.values()
        ]
    )

    prompt = f"""
    You are a research analyst. Synthesize the following research findings into a comprehensive, detailed report about: {topic}

    Research Findings:
    {findings_text}

    Available Citations:
    {citations_text}

    IMPORTANT: You must return ONLY a valid JSON object with no additional text, explanations, or formatting.

    Create a comprehensive research report that flows naturally as a single narrative. Include:
    - Specific technical details and concrete examples
    - Actionable insights practitioners can use
    - Interesting discoveries and surprising findings
    - Specific tools, libraries, or techniques mentioned
    - Performance metrics, benchmarks, or quantitative data when available
    - Notable opinions or debates in the community
    - INLINE LINKS: When making claims, include clickable links directly in the text using this format: [link text](HN_URL)
    - Use MANY inline links throughout the report. Aim for at least 4-5 links per paragraph.

    CRITICAL CITATION RULES - FOLLOW EXACTLY:

    1. NEVER replace words with bare URLs like "(https://news.ycombinator.com/item?id=123)"
    2. ALWAYS write complete sentences with all words present
    3. Add citations using descriptive link text in brackets: [descriptive text](URL)
    4. Every sentence must be grammatically complete and readable without the links
    5. Links should ALWAYS be to the Hacker News discussion, NEVER directly to the article.

    CORRECT examples:
    "PostgreSQL's performance improvements have been significant in recent versions, as discussed in [community forums](https://news.ycombinator.com/item?id=123456), with developers highlighting [specific optimizations](https://news.ycombinator.com/item?id=789012) in query processing."

    "Redis performance issues can stem from common configuration mistakes, which are well-documented in [troubleshooting guides](https://news.ycombinator.com/item?id=345678) and [community discussions](https://news.ycombinator.com/item?id=901234)."

    "React's licensing changes have sparked significant community debate, as seen in [detailed discussions](https://news.ycombinator.com/item?id=15316175) about the implications for open-source projects."

    WRONG examples (NEVER DO THIS):
    "Community discussions reveal a strong interest in the (https://news.ycombinator.com/item?id=18717168) and the common pitfalls"
    "One significant topic is the (https://news.ycombinator.com/item?id=15316175), which raises important legal considerations"

    Always link to relevant discussions for:
    - Every specific tool, library, or technology mentioned
    - Performance claims and benchmarks  
    - Community opinions and debates
    - Technical implementation details
    - Companies or projects referenced
    - Version releases or updates
    - Problem reports or solutions

    Return a JSON object with this exact structure:
    {{
        "report": "A comprehensive research report written as flowing narrative text with inline clickable links [like this](https://news.ycombinator.com/item?id=123). Include specific technical details, tools, performance metrics, community opinions, and actionable insights. Make it detailed and informative, not just a summary."
    }}
    """

    messages = [
        {
            "role": "system",
            "content": "You are a research analyst. Provide comprehensive synthesis in JSON format.",
        },
        {"role": "user", "content": prompt},
    ]

    response = call_llm(messages, max_tokens=3000)

    try:
        cleaned_response = clean_json_response(response)
        result = json.loads(cleaned_response)
        return result
    except json.JSONDecodeError as e:
        return {
            "report": "JSON parsing error, report could not be generated.",
            "error": f"JSON parsing failed, created basic synthesis. Error: {str(e)}",
        }

Try it Yourself!

Setting Up OpenAI

To run this agent, you need an OpenAI developer account.
Obtain an API key here and set up a payment method for your account here.
This agent uses gpt-4o-mini for decision-making.

Set your API key as an environment variable:

export OPENAI_API_KEY=<your_openai_key>

Running Locally

First, clone this repository:

git clone https://github.com/dbos-inc/dbos-docs.git
cd dbos-docs/hacker-news-agent

Then use uv to install dependencies and activate your virtual environment:

uv sync
source .venv/bin/activate

Start PostgreSQL (required for DBOS):

dbos postgres start

Run the agent with any research topic:

python -m hacker_news_agent "artificial intelligence"

Or try other topics:

python -m hacker_news_agent "rust programming"
python -m hacker_news_agent "database performance"
python -m hacker_news_agent "kubernetes scaling"

The agent will autonomously research your topic, make decisions about what to investigate next, and produce a comprehensive research report with insights from the Hacker News community.

If the agent fails at any point during its research, you can restart it using its workflow ID to recover it from where it left off:

python -m hacker_news_agent "artificial intelligence" --workflow-id <id>

💡Never forget again: Build a scheduled reminder app in <50 lines of Python using REST and Postgres

Qian Li — Mon, 28 Oct 2024 23:07:40 +0000

🚀 Ever forget something important? Let's build a reliable, scheduled reminder app in under 50 lines of Python.

In this tutorial, you'll create and deploy a crashproof Python backend that schedules reminder emails for any date in the future. You can try the live app here: simply enter your email address and a date, and you’ll receive a reminder on that date!

TL;DR

This application consists of:

A single Python backend service
Multiple REST APIs
A Postgres database

We'll be using:

FastAPI to define REST APIs and handle HTTP requests.
SendGrid to send emails.
DBOS to make the backend durable and serverlessly host it in the cloud.

All source code is available on GitHub.

Import and Initialize the App

Let's start off with imports and initializing the DBOS and FastAPI apps.

import os

from dbos import DBOS, SetWorkflowID
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, EmailStr
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail

app = FastAPI()
DBOS(fastapi=app)

Scheduling Emails

Next, we'll write the workflow for sending emails. First, it will send a quick confirmation email, then wait until the scheduled day to send the reminder.

Thanks to DBOS's durably executed workflow, waiting until the scheduled day is simple—no matter how far away that day is, just use sleep!

Under the hood, when you call DBOS.sleep, it saves the wake-up time in the database. This ensures that even if your program is interrupted or restarted during a days-long sleep, it will still wake up on time to send the reminder.

If you need to schedule recurring events rather than a one-time email, consider using scheduled workflows.

@DBOS.workflow()
def reminder_workflow(to_email: str, send_date: datetime, start_date: datetime):
    send_email(
        to_email,
        subject="DBOS Reminder Confirmation",
        message=f"Thank you for signing up for DBOS reminders! You will receive a reminder on {send_date}.",
    )
    days_to_wait = (send_date - start_date).days
    seconds_to_wait = days_to_wait * 24 * 60 * 60
    DBOS.sleep(seconds_to_wait)
    send_email(
        to_email,
        subject="DBOS Reminder",
        message=f"This is a reminder from DBOS! You requested this reminder on {start_date}.",
    )

Sending Emails

Next, let's write the email-sending code using SendGrid. This will require a few environment variables:

api_key = os.environ.get("SENDGRID_API_KEY", None)
if api_key is None:
    raise Exception("Error: SENDGRID_API_KEY is not set")

from_email = os.environ.get("SENDGRID_FROM_EMAIL", None)
if from_email is None:
    raise Exception("Error: SENDGRID_FROM_EMAIL is not set")

Next, we implement the send_email function using SendGrid's Python API. We’ll annotate this function with @DBOS.step so that the reminder workflow calls it durably and doesn’t re-execute it if restarted.

@DBOS.step()
def send_email(to_email: str, subject: str, message: str):
    message = Mail(
        from_email=from_email, to_emails=to_email, subject=subject, html_content=message
    )
    email_client = SendGridAPIClient(api_key)
    email_client.send(message)
    DBOS.logger.info(f"Email sent to {to_email}")

Serving the App

Next, let’s use FastAPI to create an HTTP endpoint for scheduling reminder emails. This endpoint will accept an email address and a scheduled date, then start a reminder workflow in the background.

As a basic anti-spam measure, we’ll use the provided email address and date as an idempotency key. This ensures that only one reminder can be sent to any given email address per day.

class RequestSchema(BaseModel):
    email: EmailStr
    date: str


@app.post("/email")
def email_endpoint(request: RequestSchema):
    send_date = datetime.strptime(request.date, "%Y-%m-%d").date()
    today_date = datetime.now().date()
    with SetWorkflowID(f"{request.email}-{request.date}"):
        DBOS.start_workflow(reminder_workflow, request.email, send_date, today_date)

Finally, let's serve the app's frontend from an HTML file using FastAPI. In production, we recommend using DBOS primarily for the backend, with your frontend deployed elsewhere.

@app.get("/")
def frontend():
    with open(os.path.join("html", "app.html")) as file:
        html = file.read()
    return HTMLResponse(html)

Try it Yourself!

Setting Up SendGrid

This app uses SendGrid to send reminder emails. Create a SendGrid account, verify an email for sending, and generate an API key. Then, set the API key and sender email as environment variables:

export SENDGRID_API_KEY=<your key>
export SENDGRID_FROM_EMAIL=<your email>

Deploying to the Cloud

To deploy this app to DBOS Cloud, first install the DBOS Cloud CLI (requires Node):

npm i -g @dbos-inc/dbos-cloud

Then clone the dbos-demo-apps repository and deploy:

git clone https://github.com/dbos-inc/dbos-demo-apps.git
cd python/scheduled-reminders
dbos-cloud app deploy

This command outputs a URL—visit it to schedule a reminder!
You can also visit the DBOS Cloud Console to see your app's status and logs.

Running Locally

First, clone and enter the dbos-demo-apps repository:

git clone https://github.com/dbos-inc/dbos-demo-apps.git
cd python/scheduled-reminders

Then create a virtual environment:

python3 -m venv .venv
source .venv/bin/activate

DBOS requires a Postgres database.
If you don't already have one, you can start one with Docker:

export PGPASSWORD=dbos
python3 start_postgres_docker.py

Then run the app in the virtual environment:

pip install -r requirements.txt
dbos migrate
dbos start

Visit http://localhost:8000 to schedule a reminder!

Next Steps

Check out how DBOS can make your applications more scalable and resilient:

Use durable execution to write crashproof workflows.
Use queues to gracefully manage API rate limits.
Use scheduled workflows to run your functions at recurring intervals.
Want to learn what you can build with DBOS? Explore other example applications.

Give it a try and let us know what you think 😊

How to build a crashproof customer service agent in <80 lines with Swarm 💪🐝

Qian Li — Tue, 22 Oct 2024 18:00:37 +0000

🚀 Want to build a crashproof AI agent in <80 lines of Python?

This tutorial shows you how to build a reliable AI-powered customer service agent with OpenAI's Swarm framework, and then enhance it to be crashproof with DBOS durable execution 💪🐝

This agent takes in a user's name, processes a refund for the user, and then applies a discount. Even if the agent is interrupted during refund processing, upon restart it automatically recovers, finishes processing the refund, then proceeds to the next step in its workflow.

Try running this agent and pressing Ctrl+C at any time. You can see that when it restarts, it automatically resumes from the last completed step!

All source code is available on GitHub.

Writing an AI-Powered Refund Agent

Let's start off with creating an AI-powered refund agent using OpenAI's lightweight multi-agent orchestration framework, Swarm. This agent contains two functions: process_refund to help a user to return an item, and apply_discount to apply a discount to the user's future purchases. These functions may be invoked based on OpenAI LLM's output. In addition, the process_refund function invokes several sub-steps (refund_step) which simulates a complex refund workflow.

from swarm import Agent
from dbos import DBOS

def process_refund(context_variables, item_id, reason="NOT SPECIFIED"):
    """Refund an item. Refund an item. Make sure you have the item_id of the form item_... Ask for user confirmation before processing the refund."""
    user_name = context_variables.get("user_name", "user")
    print(f"[mock] Refunding for {user_name}, item {item_id}, because {reason}...")
    for i in range(1, 6):
        refund_step(i)
        DBOS.sleep(1)
    print("[mock] Refund successfully processed!")
    return "Success!"

@DBOS.step()
def refund_step(step_id):
    print(f"[mock] Processing refund step {step_id}... Press Control + C to quit")

@DBOS.step()
def apply_discount():
    """Apply a discount to the user's cart."""
    print("[mock] Applying discount...")
    return "Applied discount of 11%"

refunds_agent = Agent(
    name="Refunds Agent",
    instructions="Help the user with a refund. If the reason is that it was too expensive, offer the user a refund code. If they insist, then process the refund.",
    functions=[process_refund, apply_discount],
)

We decorate the refund_step and apply_discount functions as DBOS steps in the agent's workflow. This way, if the agent's workflow is interrupted while processing a refund, when it restarts, it will resume from the last completed step. DBOS guarantees that once the agent's workflow starts, you will always get a refund, and never be refunded twice or get the discount twice!

Reliable Agentic Workflow Orchestration

Next, let's enhance Swarm with a few lines of DBOS code to make it resilient to any failure.

This code declares the main loop of Swarm (run) to be a durable DBOS workflow and each chat completion to be a DBOS step in that workflow. Therefore, if a workflow is interrupted, it will skip already finished chat completion steps and use the recorded outputs of those steps.

from dbos import DBOS, DBOSConfiguredInstance
from swarm import Swarm
from swarm.repl.repl import pretty_print_messages

DBOS()

@DBOS.dbos_class()
class DurableSwarm(Swarm, DBOSConfiguredInstance):
    def __init__(self, client=None):
        Swarm.__init__(self, client)
        DBOSConfiguredInstance.__init__(self, "openai_client")

    @DBOS.step()
    def get_chat_completion(self, *args, **kwargs):
        return super().get_chat_completion(*args, **kwargs)

    @DBOS.workflow()
    def run(self, *args, **kwargs):
        response = super().run(*args, **kwargs)
        pretty_print_messages(response.messages)
        return response

DBOS.launch()

Finally, let's create a DurableSwarm instance and use the refund agent to process refunds! This script creates an interactive CLI for your agent.

from agents import refunds_agent

def main():
    client = DurableSwarm()
    print("Connecting to Durable Refund Agent 💪🐝")

    user_name = input("\033[90mWhat's your name\033[0m: \n")
    if user_name.strip() == "":
        return  # Exit if user doesn't provide a name

    query = "I want to refund item 99 because it's too expensive and I don't like its color! I want to proceed with the refund and also get a discount for my next purchase!"
    context_variables = {"user_name": user_name}

    client.run(
        agent=refunds_agent,
        messages=[{"role": "user", "content": query}],
        context_variables=context_variables,
    )

if __name__ == "__main__":
    main()

Try it Yourself!

To run this app, you need an OpenAI developer account. Obtain an API key here and set up a payment method for your account here.

Set your API key as an environment variable:

export OPENAI_API_KEY=<your_openai_key>

Then, clone and enter the durable-swarm repository:

git clone https://github.com/dbos-inc/durable-swarm.git
cd examples/reliable_refund

Next, create a virtual environment and install Swarm and DBOS. Swarm requires Python >=3.10.

python3 -m venv .venv
source .venv/bin/activate
pip install dbos git+https://github.com/openai/swarm.git

DBOS requires a Postgres database. If you don't already have one, you can start one with Docker:

export PGPASSWORD=dbos
python3 start_postgres_docker.py

Finally, run the app in the virtual environment with a single command python3 main.py. You can crash and restart this application as many times as you want. An example output:

> python3 main.py

Connecting to Durable Refund Agent 💪🐝
What's your name:
Qian
[mock] Refunding for Qian, item item_99, because Too expensive and I don't like its color...
[mock] Processing refund step 1... Press Control + C to quit
[mock] Processing refund step 2... Press Control + C to quit
[mock] Processing refund step 3... Press Control + C to quit
^C⏎

# Resume from where the last completed step (step 3), continuing with step 4.
> python3 main.py

Connecting to Durable Refund Agent 💪🐝
[mock] Refunding for Qian, item item_99, because Too expensive and I don't like its color...
[mock] Processing refund step 4... Press Control + C to quit
[mock] Processing refund step 5... Press Control + C to quit
[mock] Refund successfully processed!
[mock] Applying discount...
Refunds Agent:
process_refund("item_id"= "item_99", "reason"= "Too expensive and I don't like its color")
apply_discount()
Refunds Agent: I've processed the refund for item 99 and also applied a discount of 11% for your next purchase. If there's anything else you need, feel free to ask!

Next Steps

Check out how DBOS can make your applications more scalable and resilient:

Learn more about Durable Swarm
Use durable execution to write crashproof workflows.
Use queues to gracefully manage API rate limits.
Use scheduled workflows to run your functions at recurring intervals.
Want to learn what you can build with DBOS? Explore other example applications.

Reliable Multi-Agent Orchestration with Durable Swarm 💪🐝

Qian Li — Thu, 17 Oct 2024 22:01:06 +0000

🚀 Want to build reliable AI agents? Excited to release Durable Swarm 💪🐝, a drop-in replacement for OpenAI's new Swarm framework.

GitHub repo: https://github.com/dbos-inc/durable-swarm
(Please give us a star if you like it!)

Overview

We enhance Swarm with durable execution to help you build reliable, scalable multi-agent systems. Durable Swarm makes your agentic workflows resilient to failures, so that if they are interrupted or restarted, they automatically resume from their last completed steps.

We believe that as multi-agent workflows become more common, longer-running, and more interactive, it's important to make them reliable. If an agent spends hours waiting for user inputs or processing complex workflows, it must be resilient to transient failures, such as a server restart. However, reliable multi-agent orchestration isn't easy—it requires complex rearchitecting like routing agent communication through SQS or Kafka.

Durable execution helps you write reliable agents while preserving the ease of use of a framework like Swarm. The idea is to automatically persist the execution state of your Swarm workflow in a Postgres database. That way, if your program is interrupted, it can automatically resume your agentic workflows from the last completed step.

Under the hood, we implemented Durable Swarm using DBOS and its lightweight durable execution. The entire implementation of Durable Swarm is <20 lines of code, declaring the main loop of Swarm to be a durable workflow and each chat completion or tool call to be a step in that workflow.

Making Swarm Durable

To add Durable Swarm to your project, simply create a durable_swarm.py file containing the following code:

from swarm import Swarm
from dbos import DBOS, DBOSConfiguredInstance

DBOS()

@DBOS.dbos_class()
class DurableSwarm(Swarm, DBOSConfiguredInstance):
    def __init__(self, client=None):
        Swarm.__init__(self, client)
        DBOSConfiguredInstance.__init__(self, "openai_client")

    @DBOS.step()
    def get_chat_completion(self, *args, **kwargs):
        return super().get_chat_completion(*args, **kwargs)

    @DBOS.step()
    def handle_tool_calls(self, *args, **kwargs):
        return super().handle_tool_calls(*args, **kwargs)

    @DBOS.workflow()
    def run(self, *args, **kwargs):
        return super().run(*args, **kwargs)

DBOS.launch()

Then use DurableSwarm instead of Swarm in your applications—it's a drop-in replacement.

Getting Started

To get started, install Swarm and DBOS and initialize DBOS. Swarm requires Python >=3.10.

pip install dbos git+https://github.com/openai/swarm.git
dbos init --config

You also need an OpenAI API key. You can obtain one here. Set it as an environment variable:

export OPENAI_API_KEY=<your-key>

To try Durable Swarm out, create durable_swarm.py as above then create a main.py file in the same directory containing this simple program:

from swarm import Agent
from durable_swarm import DurableSwarm

client = DurableSwarm()

def transfer_to_agent_b():
    return agent_b


agent_a = Agent(
    name="Agent A",
    instructions="You are a helpful agent.",
    functions=[transfer_to_agent_b],
)

agent_b = Agent(
    name="Agent B",
    instructions="Only speak in Haikus.",
)

response = client.run(
    agent=agent_a,
    messages=[{"role": "user", "content": "I want to talk to agent B."}],
)

print(response.messages[-1]["content"])

DBOS requires Postgres. If you already have a Postgres server, modify dbos-config.yaml to configure its connection information.
Otherwise, we provide a script to start Postgres using Docker:

export PGPASSWORD=swarm
python3 start_postgres_docker.py

Finally, run your agents:

> python3 main.py

Agent B is here,
Ready to help you today,
What do you need, friend?

Converting Existing Apps to DurableSwarm

You can convert any existing Swarm app to DurableSwarm in three simple steps:

Install dbos and initialize it with dbos init --config.
Add durable_swarm.py to your project.
Use DurableSwarm in place of Swarm in your application.

Note: DurableSwarm currently doesn't support streaming

Give it a try and let us know what you think!

Next Steps

Check out how DBOS can make your applications more scalable and resilient:

Use durable execution to write crashproof workflows.
Use queues to gracefully manage API rate limits.
Use scheduled workflows to run your functions at recurring intervals.
Want to learn what you can build with DBOS? Explore other example applications.

How to build a serverless AI chatbot in <100 lines of Python

Qian Li — Sun, 13 Oct 2024 19:49:11 +0000

🚀 Want to build chatbot in <100 lines of Python and serverlessly host it on the cloud with a single command?

This guide shows you how to build an interactive LLM-powered chatbot with DBOS and LangChain and serverlessly deploy it to DBOS Cloud.

You can see the chatbot live here.

In addition to chatting, this bot displays both the amount of CPU time and wall-clock time consumed by your requests. As you chat, you'll quickly notice that while your requests may take a long time, they consume very little CPU. That's because they spend most of their time idle waiting for the LLM to respond. This gap explains why DBOS is 50x more cost-efficient than other serverless platforms for AI workloads—because DBOS bills only for the CPU time you actually consume, while other platforms bill for the total request duration.

All source code is available on GitHub.

Import and Initialize the App

Let's start off with imports and initializing DBOS.
We'll also set up FastAPI to serve HTTP requests.

import os
import threading
import time
from collections import deque

import psutil
from dbos import DBOS
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from langchain_core.messages import HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import START, MessagesState, StateGraph
from psycopg_pool import ConnectionPool
from pydantic import BaseModel

from .schema import chat_history

app = FastAPI()
dbos = DBOS(fastapi=app)

Setting Up LangChain

Next, let's set up Langchain.
We'll use Langchain to answer each chat message using OpenAI's gpt-3.5-turbo model. We'll configure LangChain to store message history in Postgres so it persists across app restarts.

For fun, let's also instruct our chatbot to talk like a pirate.

def create_langchain():
    # We use gpt-3.5-turbo as our model.
    model = ChatOpenAI(model="gpt-3.5-turbo")

    # This prompt instructs the model how to act. We'll tell it to talk like a pirate!
    prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                "You talk like a pirate. Answer all questions to the best of your ability.",
            ),
            MessagesPlaceholder(variable_name="messages"),
        ]
    )

    # This function tells LangChain to invoke our model with our prompt.
    def call_model(state: MessagesState):
        chain = prompt | model
        response = chain.invoke(state)
        return {"messages": response}

    # Create a checkpointer LangChain can use to store message history in Postgres.
    db = DBOS.config["database"]
    connection_string = f"postgresql://{db['username']}:{db['password']}@{db['hostname']}:{db['port']}/{db['app_db_name']}"
    pool = ConnectionPool(connection_string)
    checkpointer = PostgresSaver(pool)

    # Finally, construct and return the graph LangChain uses to respond to each message.
    # This chatbot uses a simple one-node graph that just calls the model.
    graph = StateGraph(state_schema=MessagesState)
    graph.add_node("model", call_model)
    graph.add_edge(START, "model")
    return graph.compile(checkpointer=checkpointer)


chain = create_langchain()

Handling Chats

Now, let's chat! We'll first write the endpoint that handles each chat request.

This endpoint is a DBOS workflow with three steps:

Store the incoming chat message in Postgres.
Use LangChain to query the LLM to respond to the chat message.
Store the response in Postgres.

It also records the total duration of each request in an in-memory buffer.

class ChatSchema(BaseModel):
    message: str
    username: str


@app.post("/chat")
@DBOS.workflow()
def chat_workflow(chat: ChatSchema):
    start_time = time.time()
    insert_chat(chat.username, chat.message, True)
    response = query_model(chat.message, chat.username)
    insert_chat(chat.username, response, False)
    elapsed_time = time.time() - start_time
    wallclock_times_buffer.append((time.time(), elapsed_time))
    return {"content": response, "isUser": True}

Next, let's write the function that actually queries LangChain for each new message. It uses your username as a thread ID so different users can have different threads of conversation.

We annotate this function with @DBOS.step() to mark it as a step in our chat workflow.

@DBOS.step()
def query_model(message: str, username: str) -> str:
    config = {"configurable": {"thread_id": username}}
    input_messages = [HumanMessage(message)]
    output = chain.invoke({"messages": input_messages}, config)
    return output["messages"][-1].content

We also need a history endpoint that retrieves all past chats from the database for a particular user.

This function is called when we start the chatbot so it can display your chat history.

@app.get("/history/{username}")
def history_endpoint(username: str):
    return get_chats(username)

Then, let's use SQLAlchemy to write the functions that write chats to and read chats from the database.
We annotate these functions with @DBOS.transaction() to access DBOS's managed database connection.

@DBOS.transaction()
def insert_chat(username: str, content: str, is_user: bool):
    DBOS.sql_session.execute(
        chat_history.insert().values(
            username=username, content=content, is_user=is_user
        )
    )


@DBOS.transaction()
def get_chats(username: str):
    stmt = (
        chat_history.select()
        .where(chat_history.c.username == username)
        .order_by(chat_history.c.created_at.asc())
    )
    result = DBOS.sql_session.execute(stmt)
    return [{"content": row.content, "isUser": row.is_user} for row in result]

Additionally, let's serve the app's frontend from an HTML file using FastAPI. In production, we recommend using DBOS primarily for the backend, with your frontend deployed elsewhere.

@app.get("/")
def frontend():
    with open(os.path.join("html", "app.html")) as file:
        html = file.read()
    return HTMLResponse(html)

Tracking App Usage

Finally, let's write some code to track the CPU time and wall-clock time consumed by your requests so we can display those metrics in the app's UI. This code runs once a second in a background thread.

We track the CPU consumption of this process using psutil. We track wall-clock time by recording the end-to-end duration of each request.

When you first start the app, you'll notice some small residual CPU consumption from the HTTP server. However, as you start chatting, you'll quickly see that each chat only consumes ~10ms of CPU time, but 1-2 seconds of wall-clock time. This gap explains why DBOS is 50x cheaper than other serverless platforms for AI workloads—because DBOS bills only for the CPU time you actually consume, while other platforms bill for the total request duration.

last_cpu_time_ms = 0
cpu_times_buffer = deque()
wallclock_times_buffer = deque()


def update_cpu_usage():
    while True:
        time.sleep(1)
        global last_cpu_time_ms
        # Every second, record CPU time consumed by this process
        # in the last second.
        process = psutil.Process()
        cpu_times = process.cpu_times()
        cpu_time = cpu_times.system + cpu_times.user
        time_consumed = cpu_time - last_cpu_time_ms
        if last_cpu_time_ms > 0:
            cpu_times_buffer.append((time.time(), time_consumed))
        last_cpu_time_ms = cpu_time
        # We only track usage in the last minute, so
        # pop measurements more than 60 seconds old.
        for buf in [cpu_times_buffer, wallclock_times_buffer]:
            while buf and time.time() - buf[0][0] > 60:
                buf.popleft()


threading.Thread(target=update_cpu_usage).start()


@app.get("/times")
def times_endpoint():
    return {
        "cpu_time": sum([t for _, t in cpu_times_buffer]),
        "wall_clock_time": sum([t for _, t in wallclock_times_buffer]),
    }

Try it Yourself!

Creating an OpenAI Account

To run this app, you need an OpenAI developer account. Obtain an API key here and set up a payment method for your account here. This bot uses gpt-3.5-turbo for text generation. Make sure you have some credits (~$1) to use it.

Set your API key as an environment variable:

export OPENAI_API_KEY=<your_openai_key>

Deploying to the Cloud

To deploy this app to DBOS Cloud, first install the DBOS Cloud CLI (requires Node):

npm i -g @dbos-inc/dbos-cloud

Then clone the dbos-demo-apps repository and deploy:

git clone https://github.com/dbos-inc/dbos-demo-apps.git
cd python/chatbot
dbos-cloud app deploy

This command outputs a URL—visit it to see your chatbot! You can also visit the DBOS Cloud Console to see your app's status and logs.

Running Locally

First, clone and enter the dbos-demo-apps repository:

git clone https://github.com/dbos-inc/dbos-demo-apps.git
cd python/chatbot

Then create a virtual environment:

python3 -m venv .venv
source .venv/bin/activate

DBOS requires a Postgres database. If you don't already have one, you can start one with Docker:

export PGPASSWORD=dbos
python3 start_postgres_docker.py

Then run the app in the virtual environment:

pip install -r requirements.txt
dbos migrate
dbos start

Visit http://localhost:8000 to see your chatbot!

Next Steps

Check out how DBOS can make your applications more scalable and resilient:

Use durable execution to write crashproof workflows.
Use queues to gracefully manage API rate limits.
Use scheduled workflows to run your functions at recurring intervals.
Want to learn what you can build with DBOS? Explore other example applications.

Building a crashproof cloud backend in Python using REST and PostgreSQL

Qian Li — Fri, 11 Oct 2024 23:16:19 +0000

🚀 Want to build applications so reliable you can give them a crash button 💣?

This guide shows you how to build and deploy a crashproof Python backend for an online store. It's an online store where you can click the "Buy" button to purchase widgets. At any time, press the "Crash" button and within seconds, the store will recover as if nothing happened.

It consists of a single service, several REST APIs, and a PostgreSQL database. We'll be using DBOS and FastAPI to build and deploy our app in the cloud, for free!

All source code is available on GitHub. You can see the application live here.

Import and Initialize the App

Let's start off with imports and initializing the DBOS and FastAPI apps. We'll also define a few constants for later.

import os

from dbos import DBOS, SetWorkflowID
from fastapi import FastAPI, HTTPException, Response
from fastapi.responses import HTMLResponse

from .schema import OrderStatus, orders, products

app = FastAPI()

DBOS(fastapi=app)

WIDGET_ID = 1
PAYMENT_STATUS = "payment_status"
PAYMENT_ID = "payment_id"
ORDER_ID = "order_id"

The Checkout Workflow

Next, let's write the checkout workflow. This workflow is triggered whenever a customer buys a widget. It creates a new order, then reserves inventory, then processes payment, then marks the order as paid and dispatches the order for fulfillment. If any step fails, it backs out, returning reserved inventory and marking the order as cancelled.

DBOS durably executes this workflow: each of its steps executes exactly-once and if it's ever interrupted, it automatically resumes from where it left off.

You can try this yourself! On the live application, start an order and press the crash button at any time. Within seconds, your app will recover to exactly the state it was in before the crash and continue as if nothing happened.

@DBOS.workflow()
def checkout_workflow():
    # Create a new order
    order_id = create_order()

    # Attempt to reserve inventory, cancelling the order if no inventory remains.
    inventory_reserved = reserve_inventory()
    if not inventory_reserved:
        DBOS.logger.error(f"Failed to reserve inventory for order {order_id}")
        update_order_status(order_id=order_id, status=OrderStatus.CANCELLED.value)
        DBOS.set_event(PAYMENT_ID, None)
        return

    # Send a unique payment ID to the checkout endpoint so it
    # can redirect the customer to the payments page.
    DBOS.set_event(PAYMENT_ID, DBOS.workflow_id)

    # Wait for a message that the customer has completed payment.
    payment_status = DBOS.recv(PAYMENT_STATUS)

    # If payment succeeded, mark the order as paid and start the order dispatch workflow.
    # Otherwise, return reserved inventory and cancel the order.
    if payment_status == "paid":
        DBOS.logger.info(f"Payment successful for order {order_id}")
        update_order_status(order_id=order_id, status=OrderStatus.PAID.value)
        DBOS.start_workflow(dispatch_order_workflow, order_id)
    else:
        DBOS.logger.warn(f"Payment failed for order {order_id}")
        undo_reserve_inventory()
        update_order_status(order_id=order_id, status=OrderStatus.CANCELLED.value)

    # Finally, send the order ID to the payment endpoint so it
    # can redirect the customer to the order status page.
    DBOS.set_event(ORDER_ID, str(order_id))

The Checkout and Payment Endpoints

Now, let's use FastAPI to write the HTTP endpoint for checkout.

This endpoint receives a request when a customer presses the "Buy Now" button.
It starts the checkout workflow in the background, then waits for the workflow to generate and send it a unique payment ID.
It then returns the payment ID so the browser can redirect the user to the payments page.

The endpoint accepts an idempotency key so that even if the customer presses "buy now" multiple times, only one checkout workflow is started.

@app.post("/checkout/{idempotency_key}")
def checkout_endpoint(idempotency_key: str) -> Response:
    # Idempotently start the checkout workflow in the background.
    with SetWorkflowID(idempotency_key):
        handle = DBOS.start_workflow(checkout_workflow)
    # Wait for the checkout workflow to send a payment ID, then return it.
    payment_id = DBOS.get_event(handle.workflow_id, PAYMENT_ID)
    if payment_id is None:
        raise HTTPException(status_code=404, detail="Checkout failed to start")
    return Response(payment_id)

Let's also write the HTTP endpoint for payments. It uses the payment ID to signal the checkout workflow whether the payment succeeded or failed. It then retrieves the order ID from the checkout workflow so the browser can redirect the customer to the order status page.

@app.post("/payment_webhook/{payment_id}/{payment_status}")
def payment_endpoint(payment_id: str, payment_status: str) -> Response:
    # Send the payment status to the checkout workflow.
    DBOS.send(payment_id, payment_status, PAYMENT_STATUS)
    # Wait for the checkout workflow to send an order ID, then return it.
    order_url = DBOS.get_event(payment_id, ORDER_ID)
    if order_url is None:
        raise HTTPException(status_code=404, detail="Payment failed to process")
    return Response(order_url)

Database Operations

Next, let's write some database operations. Each of these functions performs a simple CRUD operation, like retrieving product information or updating inventory. We apply the @DBOS.transaction to each of them to give them access to a pre-configured database connection. We also make some of these functions HTTP endpoints with FastAPI so the frontend can access them, for example to display order status.

@DBOS.transaction()
def reserve_inventory() -> bool:
    rows_affected = DBOS.sql_session.execute(
        products.update()
        .where(products.c.product_id == WIDGET_ID)
        .where(products.c.inventory > 0)
        .values(inventory=products.c.inventory - 1)
    ).rowcount
    return rows_affected > 0


@DBOS.transaction()
def undo_reserve_inventory() -> None:
    DBOS.sql_session.execute(
        products.update()
        .where(products.c.product_id == WIDGET_ID)
        .values(inventory=products.c.inventory + 1)
    )


@DBOS.transaction()
def create_order() -> int:
    result = DBOS.sql_session.execute(
        orders.insert().values(order_status=OrderStatus.PENDING.value)
    )
    return result.inserted_primary_key[0]


@app.get("/order/{order_id}")
@DBOS.transaction()
def get_order(order_id: int):
    return (
        DBOS.sql_session.execute(orders.select().where(orders.c.order_id == order_id))
        .mappings()
        .first()
    )


@DBOS.transaction()
def update_order_status(order_id: int, status: int) -> None:
    DBOS.sql_session.execute(
        orders.update().where(orders.c.order_id == order_id).values(order_status=status)
    )


@app.get("/product")
@DBOS.transaction()
def get_product():
    return DBOS.sql_session.execute(products.select()).mappings().first()


@app.get("/orders")
@DBOS.transaction()
def get_orders():
    rows = DBOS.sql_session.execute(orders.select())
    return [dict(row) for row in rows.mappings()]


@app.post("/restock")
@DBOS.transaction()
def restock():
    DBOS.sql_session.execute(products.update().values(inventory=100))

Finishing Up

A few more functions to go!

First, let's write a workflow to dispatch orders that have been paid for. This function is responsible for the "progress bar" you see for paid orders on the live demo page. Every second, it updates the progress of a paid order, then dispatches the order if it is fully progressed.

@DBOS.workflow()
def dispatch_order_workflow(order_id):
    for _ in range(10):
        DBOS.sleep(1)
        update_order_progress(order_id)

@DBOS.transaction()
def update_order_progress(order_id):
    # Update the progress of paid orders.
    progress_remaining = DBOS.sql_session.execute(
        orders.update()
        .where(orders.c.order_id == order_id)
        .values(progress_remaining=orders.c.progress_remaining - 1)
        .returning(orders.c.progress_remaining)
    ).scalar_one()

    # Dispatch if the order is fully-progressed.
    if progress_remaining == 0:
        DBOS.sql_session.execute(
            orders.update()
            .where(orders.c.order_id == order_id)
            .values(order_status=OrderStatus.DISPATCHED.value)
        )

Let's also serve the app's frontend from an HTML file using FastAPI.
In production, we recommend using DBOS primarily for the backend, with your frontend deployed elsewhere.

@app.get("/")
def frontend():
    with open(os.path.join("html", "app.html")) as file:
        html = file.read()
    return HTMLResponse(html)

Finally, here is the crash endpoint. It crashes your app. Trigger it as many times as you want—DBOS always comes back, resuming from exactly where it left off!

@app.post("/crash_application")
def crash_application():
    os._exit(1)

Try it Yourself!

Deploying to the Cloud

To deploy this example to DBOS Cloud, first install the Cloud CLI (requires Node):

npm i -g @dbos-inc/dbos-cloud

Then clone the dbos-demo-apps repository and deploy:

git clone https://github.com/dbos-inc/dbos-demo-apps.git
cd python/widget-store
dbos-cloud app deploy

This command outputs a URL—visit it to see your app!
You can also visit the DBOS Cloud Console to see your app's status and logs.

Running Locally

First, clone and enter the dbos-demo-apps repository:

git clone https://github.com/dbos-inc/dbos-demo-apps.git
cd python/widget-store

Then create a virtual environment:

python3 -m venv .venv
source .venv/bin/activate

This demo requires a Postgres database.
If you don't already have one, you can start one with Docker:

export PGPASSWORD=dbos
python3 start_postgres_docker.py

Then run the app in the virtual environment:

pip install -r requirements.txt
dbos migrate
dbos start

Visit http://localhost:8000 to see your app!

Next Steps

Check out how DBOS can make your applications more scalable and resilient:

Use durable execution to write crashproof workflows.
Use queues to gracefully manage API rate limits.
Use scheduled workflows to run your functions at recurring intervals.
Want to learn what you can build with DBOS? Explore other example applications.

Give it a try and let us know what you think 😊

Build & Deploy a Serverless OpenAI App in 9 Lines of Code

Qian Li — Tue, 08 Oct 2024 18:44:41 +0000

🚀 Want to build and deploy an interactive AI app 𝘁𝗼 𝘁𝗵𝗲 𝗰𝗹𝗼𝘂𝗱 in just 𝟵 𝗹𝗶𝗻𝗲𝘀 𝗼𝗳 𝗰𝗼𝗱𝗲?

In this tutorial, you'll use LlamaIndex to create a Q&A engine, FastAPI to serve it over HTTP, and DBOS to deploy it serverlessly to the cloud.

It's based on LlamaIndex’s 5-line starter, with just 4 extra lines to make it cloud-ready. Simple, fast, and ready to scale!

Preparation

First, create a folder for your app and activate a virtual environment.

python3 -m venv ai-app/.venv
cd ai-app
source .venv/bin/activate
touch main.py

Then, install dependencies and initialize a DBOS config file.

pip install dbos llama-index
dbos init --config

Next, to run this app, you need an OpenAI developer account. Obtain an API key here. Set the API key as an environment variable.

export OPENAI_API_KEY=XXXXX

Declare the environment variable in dbos-config.yaml:

env:
  OPENAI_API_KEY: ${OPENAI_API_KEY}

Finally, let's download some data. This app uses the text from Paul Graham's "What I Worked On". You can download the text from this link and save it under data/paul_graham_essay.txt of your app folder.

Now, your app folder structure should look like this:

ai-app/
├── dbos-config.yaml
├── main.py
└── data/
    └── paul_graham_essay.txt

Load Data and Build a Q&A Engine

Now, let's use LlamaIndex to write a simple AI application in just 5 lines of code.
Add the following code to your main.py:

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(documents)

query_engine = index.as_query_engine()
response = query_engine.query("What did the author do growing up?")
print(response)

This script loads data and builds an index over the documents under the data/ folder, and it generates an answer by querying the index. You can run this script and it should give you a response, for example:

$ python3 main.py

The author worked on writing short stories and programming...

HTTP Serving

Now, let's add a FastAPI endpoint to serve responses through HTTP. Modify your main.py as follows:

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from fastapi import FastAPI

app = FastAPI()

documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()

@app.get("/")
def get_answer():
    response = query_engine.query("What did the author do growing up?")
    return str(response)

Now you can start your app with fastapi run main.py. To see that it's working, visit this URL: http://localhost:8000

The result may be slightly different every time you refresh your browser window!

Hosting on DBOS Cloud

To deploy your app to DBOS Cloud, you only need to add two lines to main.py:

from dbos import DBOS
DBOS(fastapi=app)

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from fastapi import FastAPI
from dbos import DBOS

app = FastAPI()
DBOS(fastapi=app)

documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()

@app.get("/")
def get_answer():
    response = query_engine.query("What did the author do growing up?")
    return str(response)

Now, install the DBOS Cloud CLI if you haven't already (requires Node.js):

npm i -g @dbos-inc/dbos-cloud

Then freeze dependencies to requirements.txt and deploy to DBOS Cloud:

pip freeze > requirements.txt
dbos-cloud app deploy

In less than a minute, it should print Access your application at <URL>.
To see that your app is working, visit <URL> in your browser.

Congratulations, you've successfully deployed your first AI app to DBOS Cloud! You can see your deployed app in the cloud console.

Next Steps

This is just the beginning of your DBOS journey. Next, check out how DBOS can make your AI applications more scalable and resilient:

Use durable execution to write crashproof workflows.
Use queues to gracefully manage AI/LLM API rate limits.
Want to build a more complex app? Check out the AI-Powered Slackbot.

Give it a try and let me know what you think 😊

How To Process Events Exactly-Once

Peter Kraft — Sun, 06 Oct 2024 21:16:08 +0000

Want to process incoming events exactly-once?

Well, any distributed systems pedant will say you can't, because it's theoretically impossible. And technically, they're right: if you send a message and don't get an answer, you have no way of knowing whether the receiver is offline or just slow, so eventually you have no choice but to send the message again if you want it processed.

So if exactly-once processing is impossible, why do many systems, including DBOS, claim to provide it?

The trick is to leverage another property: idempotence. If you design a message receiver to be idempotent, then you can deliver a message to it multiple times and it will be fine because the duplicate deliveries have no effect. Thus, the combination of at-least-once delivery and idempotence is identical to exactly-once semantics in practice.

Under the hood, this is exactly how DBOS event receivers (like for Kafka) work. They generate a unique key from an event (for example, from a Kafka topic + partition + offset) and use it as an idempotency key for the event processing workflow. That way, even if an event is delivered multiple times, the workflow only processes it once.

Here's all the code you need to process Kafka messages exactly-once:

from dbos import DBOS, KafkaMessage

@DBOS.kafka_consumer(config, ["topic"])
@DBOS.workflow()
def test_kafka_workflow(msg: KafkaMessage):
    DBOS.logger.info(f"Message received: {msg.value.decode()}")

Learn more here!

How we made applications observable by default with OpenTelemetry

max — Wed, 28 Aug 2024 18:06:45 +0000

Hey, I'm Max Demoulin, engineer at DBOS. Today I wanted to share my story about using opentelemetry and our node.js framework to make apps observable by default! Would love to get feedback & comments :)

Observable by default

Invariably, applications need observability – i.e., ways to know what is going on when they run. But observability, today, does not come easily. It requires a lot of engineering effort. Given observability is an absolute requirement, it should be an integral part of modern development frameworks, and require minimal engineering effort.

In this post, I will present typical questions observability must answer and highlight key telemetry features it requires. I will then explore common observability techniques and the steps required to make telemetry data actionable.

Finally, I will present how we designed DBOS Transact and DBOS Cloud to enable observability by default. On the development side, DBOS Transact automatically generates observability code and exports telemetry signals using OpenTelemetry. On DBOS Cloud, we provide a default observability backend to every application.

The result is thorough observability that takes orders of magnitude less engineering effort to achieve.

What is an Observable Application?

Observability requires three types of important telemetry data: logs, traces and metrics.

Logs represent individual events.
Traces represent a structured, hierarchical view of an application’s execution flow.
Metrics are measurements of events of interest.

In this article, we will focus on logs and traces. These signals must correlate across applications: if your payment service communicates with an authorization service, you want to trace requests going from the former through the latter. This is also known as distributed tracing and is the topic of numerous academic and industry research projects, such as Xtrace, Pivot Tracing and Dapper (yes, designing and implementing a distributed tracing system is hard.)

Let’s take the concrete example of an online store with the control flow pictured below.
‍

Here are common questions you may want to answer:

Who bought items yesterday between 12 and 3 pm?
Some transactions failed. Who started them and what were their inputs?
How does payment processing time vary as a function of clients’ User-Agent?

To answer these, you need the ability to record the occurrence of events, their associated properties, and how events correlate. For example, when a checkout request enters the system, you need to record its input parameters and tie together correlated events such as payment approval or failure.

Of course, you will want to export telemetry data from your application to some datastore so you can retrieve and analyze it for debugging or auditing purposes. On top of that, it would be really cool if you could visualize traces to help you analyze your program easier.
‍

The Observability Code

First Attempt: print()

You would be surprised how popular this is. Print stuff in your app and it goes to the standard output. Redirect the standard output to a file and have some background task periodically upload the data to your observability backend.

What’s the issue with that? First, your standard output is unstructured and it will be hard to filter logs. The lack of structure will also translate into inconsistencies across developers – different levels of detail and lack of formatting standards. Second, your print statements cannot be correlated, so you cannot query for all the events associated with a specific request. Third – a consequence of the previous problem – your print statements do not have any causal relationships, so it is hard to make sense of what happened when.

Print statements alone make it very difficult to answer the observability questions above. Grep magic is always possible, I suppose. The next techniques in this section address these issues (structure, correlation and causation of telemetry data.)

Second Attempt: Structured Logging

Instead of printing “raw” messages, you can use a structured logger. This kind of utility lets you attach contextual information to log entries. For example, you might add the following log entry when a request enters the system:

logger.info("Received request!", "body", request.body);

Structured logging allows you to query log entries matching specific keys. In the example above, you could filter all the logs for entries having the key body. In addition, most loggers come with predefined keys like the timestamp of the entry.

You can see how structured logging allows you to answer questions like “Who bought items yesterday between 12 and 3 pm?” You could filter all the log entries having a body field, emitted yesterday between 12 and 3 p.m.

Structured logging on its own does not allow you to correlate log entries to form a complete picture of a request’s life cycle. Assume you want to answer the more specific question “What was the body of requests which resulted in a failed payment”. With our current system, you can query all the log entries with a body key and all the log entries reporting a payment failure, but correlating them together is hard (especially if you have many entries to sift through.)
‍

Third Attempt: Request IDs

The solution to the above limitation is simply to include a special identifier key in each log entry. This identifier should be unique per request, such that you can use the ID to query every log entry related to a specific request. The nice thing about using a unique identifier is that you can propagate it across applications, enabling a basic form of distributed tracing.

This technique is often referred to as a request ID. One early attempt at standardizing it is the x-request-id HTTP header. The way to use this header is as follows: when your app receives a request, look for the header in the request and generate a new ID if the x-request-id header is not found. Then, store the ID in such a way that all subsequent log entries dealing with that request can include the ID. (For example you could inject the ID in the request context or configure an instance of your structured logger to automatically include the ID).

In common API frameworks like express.js, this can be achieved by writing middleware implementing the following logic:

// Middleware to handle X-Request-ID
const requestIdMiddleware = (req, res, next) => {
       // Check if X-Request-ID header is present
       let requestId = req.header('X-Request-ID');
       // If not present, generate a new UUID
       if (!requestId) {
              requestId = uuidv4(); // or your favorite ID algorithm
       }
       // Contextualize the request ID
       req.requestId = requestId;
       // Add X-Request-ID to the response so it can be propagated
       res.setHeader('X-Request-ID', requestId);
       // Proceed to the next middleware
       next();
};

You are now in a position to answer moderately complex queries. Take our second observability question: “Some transactions failed. Who started them and what were their inputs?” Assuming you have request IDs for failed transactions, you can now query all log entries related to these requests. Assuming you wrote the log calls with the necessary keys, you should be able to get a better picture of why these requests failed.

One key limitation of request IDs is they do not capture the structure of your application’s execution flow, and thus, the causal relationship between log entries. This makes it much harder to analyze your application for critical path analysis or root-cause analysis. For an example of critical path analysis at scale, see this recent research paper from Uber.

Fourth Attempt 🥵: Tracing to the Rescue

Tracing aims to solve the limitations above. It is a telemetry system that not only lets you correlate events and attach contextual information, but also captures the structure of your application’s execution flow. In the tracing jargon, a trace is uniquely identified by an ID and composed of a set of spans. Each span is a node in the execution graph. The causal relationship between spans is captured by a notion of parenting, meaning spans can have parents and children.

This is where things get complicated. Implementing such a system is a lot of work, and you should rely on upcoming standards like the W3C recommendation on trace contexts. You should rely on a standard because they solve two major pain points with observability: (1) they have popular implementations like OpenTelemetry (OTel) – which happens to be the second most active CNCF project to this day – and (2) they are widely adopted by major observability backends (e.g., Datadog).

In technical terms, these standards provide a specification for the telemetry structure itself (e.g., the “spans” jargon above) and a network application layer specification (for example, which HTTP headers should be used to carry telemetry context across applications.)

Here is an example, using the OTel javascript implementation, of manually instrumenting the online store. As you can see below, it is not exactly effortless. The emboldened code is the observability code.

...
import { trace, defaultTextMapGetter, ROOT_CONTEXT } from '@opentelemetry/api';
import { tracer } from 'my/app/telemetry/module';
...

// A simplified checkout request handler
router.post('/checkout', async (ctx) => { 
       // If present, retrieve the trace context from the request
       const httpTracer = new W3CTraceContextPropagator();
       const extractedSpanContext = trace.getSpanContext(
              httpTracer.extract(ROOT_CONTEXT, ctx.request.headers, defaultTextMapGetter)       
       }

       // Create a new span
       let span: Span;
       const spanAttributes = {
              requestIP: ctx.request.ip,
              requestURL: ctx.request.url,
              requestMethod: ctx.request.method,
              requestAgent: ctx.request.header("user-agent"),
       };
       if (extractedSpanContext === undefined) {
              span = tracer.startSpan(ctx.url, spanAttributes);
       } else {
              extractedSpanContext.isRemote = true;
              span = tracer.startSpanWithContext(extractedSpanContext, ctx.url, spanAttributes);        
       }    

       // Call the payment workflow 
       const paymentSuccess = paymentWorkflow(ctx.request, span);   
       if (paymentSuccess) {
              ctx.status(200);  
       } else {
              ctx.status(500);
       }

       // Inject the trace context to the response's headers
       httpTracer.inject(
              trace.setSpanContext(ROOT_CONTEXT, span.spanContext()),
              { context: ctx },
       );
       tracer.endSpan(oc.span);
}

function paymentWorkflow(req, parentSpan) { 
       // Create a new span for the payment workflow
       const span = tracer.startSpan(      
              "paymentWorkflow",
              parentCtx?.span, // Pass the handler's context as parent
           );   
       // ... do the payment workflow ...   
       const success = ...  
       // End the span before returning 
       trace.endSpan(span); 
       return success;
}

This code:

Creates a new span when a new request arrives, potentially extracting its context from the request’s headers
Passes the span to the paymentWorkflow function.
Creates a new child span at the beginning of paymentWorkflow and closes it before returning
Attaches the trace context to the response headers
Ends the span

‍
With this in place, we can now perform powerful observability queries like analyzing the average payment processing time distribution per client’s user agent – a useful piece of information to understand your users. Assuming you can query the traces using SQL, you could simply do:

SELECT span_attribute.requestAgent, AVG(span_end_time - span_start_time) as paymentResponseTime
FROM traces
WHERE spanName='paymentWorkflow'
GROUP BY span_attribute.requestAgent

With a well-defined structure, the complexity of parsing raw telemetry data and querying is massively reduced compared to exploiting a bunch of “print” data. The query above is trivial. Furthermore, a standardized structure allows interoperability across tools.

This is great! But you are not done yet. Your telemetry signals are still in your application and cannot be queried to answer observability queries. You must now ensure these signals are collected and exported to an observability backend, a datastore you can query to answer observability questions.

The Observability Infrastructure

Beyond your application’s observability code, you need what’s called an observability pipeline: a way to (1) collect telemetry signals from the application, (2) potentially pre-process them and (3) store them in an observability backend. In this section we will cover a gradually complex implementation – to satisfy increasingly complex requirements.

As Simple as it Gets

If your application generates low volumes of telemetry data, you can implement what I would call “direct wiring” of your application observability code to the observability backend. For example, each time you generate a log entry, push that entry to the observability backend.

This will not scale with the volume of data, because the overhead associated with sending each piece of data is large in comparison to the volume of each individual piece. This is why most observability frameworks – including loggers like winston – buffer telemetry data before exporting it.

Intermediate Collection

Assuming you spend the time tuning your application observability code to handle larger volumes of data, there will quickly be a new set of issues with handling it all within the application:

You risk losing telemetry data when the application is updated or crashes
Code complexity will increase if you need to pre-process telemetry data before exporting it, for instance if you need to obfuscate Personal Identifiable Information (PII)
Resource consumption – memory footprint and CPU usage will grow, potentially increasing your costs and impacting your application’s performance

This is why most production systems use an intermediate collector, running in the hosting platform, to process application telemetry data and periodically export it to the observability backend. For example, a Splunk setup on Kubernetes typically involves configuring a Splunk daemonset on each K8s node to ingest each container's output and upload it to a Splunk backend.

While a basic setup can be put together in a matter of days, fine-tuning the daemonset to control resource consumption, for instance, will be more involved. As your system involves more applications, creating “observability lanes” with different profiles will become necessary. Treat the OTel collector architecture as a good reference to grasp the complexity involved with setting up the infra piece of observability pipelines. Also, needless to say, bolting on these lego bricks to your production system is expensive.

Summarizing your Observability Pain Points

Let’s step back for a moment to reflect on what we’ve done so far to enable the powerful observability a production-ready system needs. Presuming you started directly with OpenTelemetry (which most of us don’t), you:

Instrumented your code with one of their SDKs
Structured the instrumentation to capture your application execution flow
Added the necessary attributes to each span (e.g., request headers) ‍ If you started with structured logging and request IDs and then transitioned to OpenTelemetry, you also did all the work of adding the correct keys to your logs, handling request IDs, etc.

In addition, you:

Wrote code to collect and buffer telemetry signals in the app
Configured the app to export telemetry data outside of the app
Set up an intermediate collector on the hosting platform

Each of these steps requires far more work than should be necessary. Implementing and maintaining best-in-class observability pipeline is a never-ending effort requiring communication within and across teams. And, make no mistake, you need observability to run a world-class production system.

My teammates and I believe there is a brighter path forward; one where every cloud application is equipped with observability by default. Applications’ behavior should be immediately understandable, regardless of the complexity of the question. Application’s authors should not have to manually write observability code nor spend time managing observability pipelines.

Observability in DBOS Transact

Why aren’t cloud applications “observable by default”? I would argue:

Most “low-level” observability tooling, e.g., runtime profilers, that can capture a program’s execution flow are not designed to integrate application-layer information. (e.g., the node.js profiler has no concept of request IDs.)
Standards like OpenTelemetry are nascent and require lots of grunt work. For example, OTel’s node.js and SpringBoot implementations have automatic instrumentation for the entry point of your application, but do not cover the rest of your code: the entire call stack will be captured within a single span.

‍

DBOS Transact exposes a workflow programming model to build reliable workflows with strong transactional properties. This programming model captures the application execution flow. DBOS Transact leverages these annotations to automate all the hard things we have mentioned so far in this post. Because we chose OpenTelemetry, this also means your application’s telemetry data can be exported to any OTel-compatible backend of your choice.

‍

Observable by Design

DBOS Transact automates the manual tasks we described above. It:

Instruments your code automatically with OTel ✅
Captures your application’s execution flow ✅
Adds the necessary attributes to each span (e.g., request headers in HTTP handlers) ✅

Let’s take the previous code snippet. In DBOS Transact, it becomes massively simpler.

@PostApi('/checkout')
static async checkoutHandler(ctxt: HandlerContext): Promise {
       // Start the payment workflow in the background, return its ID
       return await ctxt.invoke(Shop).paymentWorkflow().getWorkflowUUID();
}

@Workflow()
async async paymentWorkflow(ctxt: WorkflowContext): Promise {   
       // Process the paymenta    
    return "payment processed!";
}

PostAPI and Workflow are TypeScript decorators, special functions that automatically insert glue code around a target function. DBOS Transact uses these decorators to capture the execution graph and inject the appropriate observability code. See the diagram below.

Every operation of a DBOS Transact workflow gets its own span. Spans are automatically composed into traces. Each span is set with specific attributes corresponding to the operation, e.g., request headers for HTTP handlers. Spans are stored in context objects propagated throughout the graph. Note how HTTP handlers extract trace contexts from incoming requests and inject the tracing context in response headers.

When writing the business logic of a DBOS Transact operation, you can access the current span through the operation’s context and augment the default attribute with custom information, simply using OTel span API.

Signals Collection and Export

The DBOS Transact tracer automates the tedious process of collecting and buffering telemetry data in the application (another pain point listed above ✅). DBOS Transact will export traces to an OTLP-compatible observability backend of your choice using protobuf over HTTP.

Here is the detail of a trace generated for the code above, visualized using a local Jaeger instance:
‍

In addition, DBOS Transact automatically attaches the trace ID to log entries, formats them with the OpenTelemetry log format and exports them alongside the traces. This allows you to run observability queries across both traces and logs.

Observability in DBOS Cloud

DBOS Cloud is a hosting platform for DBOS Transact applications. It does all the hard work of setting up an observability pipeline for you, i.e., it takes care of the remaining above-mentioned pain points:

It configures the app to export telemetry data ✅
It sets up telemetry collection and exports on the hosting platform ✅
It provides an observability backend and a dashboard to run observability queries ✅

‍

Configuring the DBOS Transact OTel Exporter

When you deploy an app to DBOS Cloud, it automatically configures DBOS Transact to export telemetry signals to an OpenTelemetry collector on the host. When this configuration is enabled, DBOS Transact will (1) buffer the generated telemetry signals and (2) export them to the OTel collector. Note that DBOS Transact applications are sandboxed in secure Firecracker microVMs and DBOS Cloud hosts are configured to allow OTLP traffic flowing from application VMs to the OTel collector. See our post on the anatomy of DBOS Cloud for more information.

OTel Collector

When looking for a telemetry collector, we were particularly concerned about scalability: we wanted to have a clear path to scaling the observability pipeline as DBOS Cloud scales. The OTel collector quickly became an obvious choice: it provides various ways to configure dedicated pipelines, configure filters and manage the resource an application’s telemetry data can consume on the host.

VMs’ telemetry signals are batched and pushed to a small service we wrote to format and export the data to Redshift.

OTel-to-Redshift Exporter

SQL is great for observability queries. This is why we chose Redshift as a default observability backend. DBOS Cloud runs an OTel receiver that parses batches of OTel telemetry signals, encoded with protobuf/OTLP, and inserts them to Redshift. To handle failures from Redshift – it is an external service, like any other observability backend, after all – we implemented a mechanism to cache signals we couldn’t export and retry later.

This is the query your DBOS Cloud Grafana dashboard currently uses to display application traces:

 select   
   span_start_time as "start time",   
   datediff(millisecond, span_start_time, span_end_time) as "duration (ms)",
   application_name as "app name",
   handler_request_method as method,
   operation_type as "operation type",
   operation_name as "operation",
   span_status_code as "status",
   operation_uuid as "workflow uuid",
   trace_id as "trace id"
   from app_traces as T join (select distinct application_id, application_name from app_deployment_events) as D on T.application_id = D.application_id
   where abs(datediff(day, $__timeTo(), $__timeFrom())) <= 45 and
     $__timeFilter(span_start_time) and
     ('$trace_id' = '' or '$trace_id' = T.trace_id) and
     ('$workflow_uuid' = '' or '$workflow_uuid' = operation_uuid) and
     operation_name != 'temp_workflow' and
     ('$app_name' = '[ALL]' or '$app_name' = D.application_name) and
     ('$operation_name' = '' or '$operation_name' = T.operation_name) and
     ('$operation_type' = '' or '$operation_type' = T.operation_type)
   order by span_start_time desc limit 10000
 ) order by "start time" asc

‍

Final Thoughts

Observability is hard and it shouldn’t be. Builders should be able to focus on their business logic and get observability out of the box. DBOS Transact is designed to automate the generation of most of the observability code and provides you with a default observability pipeline.

‍

We’d love feedback and comments! You can find us on Discord.

Making Serverless 15x Cheaper

Peter Kraft — Thu, 13 Jun 2024 00:41:35 +0000

We’ve often heard developers working on stateful applications say that they want to adopt serverless technology to more easily deploy to the cloud, but can’t because it’s prohibitively expensive. For clarity, by “stateful applications” we mean applications that manage persistent state stored in a database (for example, a web app built on Postgres).

In this blog post, we'll show how to make serverless 15x cheaper for stateful applications. First, we'll explain why it's too expensive right now. Then, we'll show how we architected the DBOS Cloud serverless platform to make serverless affordable for stateful applications.
‍

Improving Serverless Efficiency

The first strategy DBOS Cloud uses to make serverless efficient is sharing executors across requests to improve resource utilization. The key idea is that stateful applications are typically I/O-bound: they spend most of their time interacting with remote data stores and external services and rarely do complex computation locally.

Consider an e-commerce application that shows you the status of your latest order. When you send it a request, it calls out to a database to look up what your last order was, then waits for a response, then calls out again to look up the status of that order, then waits for a response, then finally returns the status back to you. Here’s what that might look like:

Current serverless platforms execute such applications inefficiently because they launch every request into a separate execution environment. Each execution environment sends a request to the database, then is blocked doing nothing useful until it receives the database’s response. Here’s what it looks if two people concurrently look up their latest orders in AWS Lambda:

‍

For stateful applications, current serverless platforms spend most of their time doing nothing, then charge you for the idle time.

By contrast, DBOS Cloud multiplexes concurrent requests to the same execution environment, using the time one request spends waiting for a result from the database to serve another request. To make sure execution environments are never overloaded, DBOS Cloud continuously monitors their utilization and autoscales when needed, using Firecracker to rapidly instantiate new execution environments. Here’s what it looks if multiple people concurrently look up their latest orders in DBOS:

DBOS Includes Reliable Workflow Execution

The second strategy DBOS Cloud uses is leveraging DBOS Transact’s reliable workflows to efficiently orchestrate multiple functions in a single application. Many applications consist of multiple discrete steps that all need to complete for the application to succeed; for example, the checkout flow for an online store might reserve inventory, then process payment, then ship the order, then send a confirmation email:

‍
Conventional serverless platforms require an expensive external orchestrator like AWS Step Functions to coordinate these steps, executing each in sequence and retrying them when they fail. By contrast, DBOS Cloud uses the reliable workflows built into open-source DBOS Transact to guarantee transactional execution for an application at no additional cost.

DBOS Cloud vs. AWS Lambda Cost Comparison

Now that we’ve explained how DBOS Cloud achieves its cost efficiency, let’s measure it by comparing the cost to run a stateful application workload on DBOS Cloud versus AWS Lambda. We’re referencing AWS Lambda because it’s the most popular serverless platform, but the numbers are comparable for other platforms like Azure Functions or Google Cloud Functions.

Consider an application workflow with four steps. To keep the math simple, let’s say each step takes 10 ms and the application is invoked 250M times a month (~100 application invocations/second). Since the workflow takes 40 ms total, this works out to 10M execution seconds per month. Assuming a 512MB executor is used for both DBOS Cloud and Lambda, here’s how the cost compares:

DBOS Cloud Cost

We’ll assume that in DBOS, the application is implemented as four operations in a DBOS Transact reliable workflow. Stateful, reliable workflow execution is built into DBOS, so there is no need for a separate orchestrator like AWS Step Functions. The 10M execution seconds per month falls within the $40 per-month DBOS Cloud Pro pricing tier, so the total cost is $40 per month.

AWS Lambda + AWS Step Functions Cost

In AWS Lambda, this application would likely be implemented as four functions orchestrated by an AWS Step Functions workflow. AWS Lambda charges $0.20 per 1M function invocations plus $8.33 per 1M execution seconds (assuming a 512MB executor). Additionally, AWS Step Functions charges (using Express Workflows, the cheapest option) $1.00 per 1M workflow invocations plus $8.33 per 1M operation-seconds of execution. Thus the total cost for Lambda is $200 for the 1B function invocations (four per workflow) plus $83.33 for 10M execution seconds. The total cost for Step Functions is $250 for 250M workflow invocations plus $83.33 for 10M operation-seconds. The grand total is $616.66. As we said earlier in this post, that is over 15 times more expensive than DBOS Cloud. Here's a table to summarize our cost comparison:

‍

Try it out!

To get started with DBOS, follow our quickstart to download the open-source DBOS Transact framework and start running code locally.

If you have any questions about the article or about DBOS, please ask in the comments!

How We Manage Customer Billing in <500 Lines of Typescript

Qian Li — Mon, 20 May 2024 17:17:26 +0000

When we launched DBOS Pro in April, we needed its subscription management and billing system to be secure, reliable, scalable, and easy to maintain. To achieve this, we “ate our own dog food.” We developed a subscription management and billing app with DBOS Transact and Stripe and deployed it to DBOS Cloud. The entire production application, including cloud deployment and CI/CD, requires <500 lines of code. In this blog post, we’ll explain how it works.

The entire production application is available on GitHub.

Building a Reliable Webhook

When a customer clicks “Upgrade to DBOS Pro” on the DBOS website, the subscription management app redirects them to Stripe, where they enter payment information. After they pay, the following happens:

Stripe sends an event to a webhook endpoint on the subscription management app.
The webhook retrieves the customer’s updated subscription status from Stripe.
The webhook maps the customer’s Stripe customer ID to a DBOS Cloud account ID.
The webhook updates the customer’s subscription status in DBOS Cloud.

While this may sound simple, there are some notoriously difficult challenges that any webhook must solve to be robust and scalable. We’ll explain how hard it is to solve those challenges today, and how we solved them with Stripe and DBOS.

Challenge 1: Webhook endpoints must respond quickly.

According to Stripe best practices, upon receiving an event, a webhook should immediately acknowledge it and return a successful 2XX HTTP status code, doing any complex processing asynchronously. A typical strategy is to have the webhook write to a message queue then build another service to consume messages from the queue. This works, but introduces a lot of complexity. A “simple” subscription management app now needs three services: the webhook, the message queue, and the queue consumer.

Challenge 2: Event processing must run to completion.

Each time the subscription management app receives an event, it should process the event to completion and change the user’s subscription status in DBOS Cloud, regardless of what interruptions and transient failures may occur along the way. Otherwise, a user might pay for DBOS Pro without unlocking Pro-tier features. A typical strategy is to use an orchestration service like AWS Step Functions and configure it to automatically retry each step in the workflow. This works, but also adds complexity (and Step Functions cost!) to the increasingly less simple app.

Challenge 3: Event processing must correctly handle duplicates.

According to Stripe, webhook endpoints may receive the same event more than once. However, for each event we should only upgrade or downgrade a customer once. A common practice is to make event processing idempotent, for example by recording which events have been processed in a persistent store and never processing those events again. This works, but adds yet another service, the persistent store, to the now-rather-complex app.

To sum it all up, here’s the architecture diagram you’d need to implement to build a robust and scalable webhook on AWS today following best practices. In addition to the resources we already mentioned above, we need to configure API Gateway and EventBridge to trigger Lambda or Step Functions. To put it mildly, this isn’t easy to build, test, or maintain.

Reliable Event Processing with DBOS Transact

Now, let's look at how we implement webhook event processing with DBOS. Here's the architecture diagram. As you can imagine, this is much easier to build, test, and maintain.

To ensure the event processing code always runs to completion (Challenge 2), we implement it as a DBOS Transact reliable workflow. The workflow retrieves the customer’s DBOS Cloud user ID and Stripe subscription status, then sends a request to DBOS Cloud to update the customer’s subscription status. DBOS Transact guarantees that if a workflow is interrupted for any reason, it automatically resumes from where it left off. Moreover, it automatically retries transient failures in external API calls such as those to Stripe and DBOS Cloud, following declaratively configurable policies.

Here’s the full production implementation of the workflow (at the time of this writing) as a simple Typescript function (source code).

@Workflow()
static async stripeEventWorkflow(ctxt: WorkflowContext, subscriptionID: string, customerID: string) {
  // Retrieve the updated subscription from Stripe
  const status = await ctxt.invoke(Utils).getSubscriptionStatus(subscriptionID);

  // Map the Stripe customer ID to DBOS Cloud user ID
  const dbosAuthID = await ctxt.invoke(Utils).findAuth0UserID(customerID);

  // Send a request to the DBOS Cloud admin API to change the user's subscription status
  switch (status) {
    case 'active':
    case 'trialing':
      await ctxt.invoke(Utils).updateCloudEntitlement(dbosAuthID, DBOSPlans.pro);
      break;
    case 'canceled':
    case 'unpaid':
    case 'paused':
      await ctxt.invoke(Utils).updateCloudEntitlement(dbosAuthID, DBOSPlans.free);
      break;
    default:
      ctxt.logger.info(`Do nothing for ${status} status.`);
  }
}

The webhook endpoint simply starts this workflow, then responds with a 204 HTTP status code without waiting for the workflow to finish (Challenge 1). It can do this safely because, as we mentioned earlier, DBOS workflows are reliable–once started, they always run to completion regardless of interruptions, no message queue required.

To make sure each event is processed exactly-once in the presence of duplicates (Challenge 3), the webhook invokes the workflow using Stripe’s event.id as an idempotency key. Here’s the full production implementation of the webhook (source code):

@PostApi('/stripe_webhook')
static async stripeWebhook(ctxt: HandlerContext) {
  // Verify the request is actually from Stripe
  const req = ctxt.request;
  const event = Utils.verifyStripeEvent(req.rawBody, req.headers);

  switch (event.type) {    
    // Handle events when a user subscribes, cancels, or updates their subscription
    case 'customer.subscription.created':
    case 'customer.subscription.deleted':
    case 'customer.subscription.updated': {
      const subscription = event.data.object as Stripe.Subscription;
      // Start the workflow with event.id as the idempotency key without waiting for it to finish
      await ctxt.startWorkflow(Utils, event.id).stripeEventWorkflow(subscription.id, subscription.customer as string);
      break;
    }
    // Handle the event when a user completes payment for a subscription
    case 'checkout.session.completed': {
      const checkout = event.data.object as Stripe.Checkout.Session;
      if (checkout.mode === 'subscription') {
        await ctxt.startWorkflow(Utils, event.id).stripeEventWorkflow(checkout.subscription as string, checkout.customer as string);
      }
      break;
    }
    default:
      ctxt.logger.info(`Unhandled event type ${event.type}`);
  }
}

Simple and Scalable Deployment with DBOS Cloud

We host the cloud subscription app serverlessly on DBOS Cloud. Using a serverless solution dramatically simplifies cloud deployment, as instead of provisioning fixed infrastructure we can simply deploy the app and rely on the cloud platform to scale it. Here’s the entire production deployment script (source code) (GitHub action):

#!/bin/bash

npx dbos-cloud login --with-refresh-token ${DBOS_DEPLOY_REFRESH_TOKEN}
npx dbos-cloud db status ${DBOS_APP_DB_NAME} > /dev/null
if [[ $? -ne 0 ]]; then
  npx dbos-cloud db provision ${DBOS_APP_DB_NAME} -U subscribe -W ${DBOS_DB_PASSWORD}
fi
npx dbos-cloud app status
if [[ $? -ne 0 ]]; then
  npx dbos-cloud app register -d ${DBOS_APP_DB_NAME}
fi
npx dbos-cloud app deploy

Use DBOS Transact and DBOS Cloud for free

To get started with DBOS Transact, check out the quickstart and docs. After you’ve built an application, you can serverlessly deploy it to DBOS Cloud for free. To join our community:

Check out our code and contribute on GitHub.
Join us on Discord.

The entire production subscription management app is open-source and available here.

How to Process Events Exactly-Once with Kafka and DBOS

Peter Kraft — Fri, 03 May 2024 22:00:40 +0000

Many applications built on event streaming systems like Apache Kafka need exactly-once semantics for data integrity. In other words, they need to guarantee that events are processed exactly one time and are never under- or over-counted. However, exactly-once semantics are hard to obtain in distributed systems, so many event streaming platforms default to weaker guarantees. For example, Kafka natively supports exactly-once semantics only for stream processing, not for general applications. As a result, developers who need exactly-once semantics must invent complicated architectures to obtain it.

In this post, we’ll show you how the open-source TypeScript framework DBOS Transact simplifies makes exactly-once event processing easy. We’ll demonstrate how to write, in <15 lines of code, a Kafka consumer that processes every message sent to a topic exactly one time, regardless of interruptions, crashes, or failures.

How exactly-once Kafka message processing works with DBOS Transact

Implementing exactly-once semantics in a distributed system is challenging because it’s theoretically impossible: the sender of a message cannot always verify whether an unresponsive receiver failed to process their message or is just being slow to acknowledge receipt. To get around this problem, a common strategy is to deliver a message multiple times, but design the receiver to process each event idempotently so that duplicate message deliveries have no effect. However, developing idempotent applications is also notoriously difficult!

DBOS Transact makes exactly-once event processing easy by making idempotency easy. You can process Kafka events exactly-once with two different types of operations:

Transactions, which execute a single database transaction.
Workflows, which reliably execute multiple operations.

When your consumer receives a message from Kafka, DBOS Transact constructs an idempotency key for that message from the message’s topic, partition, and offset–a combination that is guaranteed to be unique for a Kafka cluster. What happens next depends on whether you’re executing a transaction or a workflow.

Exactly-once transactions

If you’re processing the message with a single transaction, DBOS Transact synchronously executes the transaction, records its idempotency key as part of that transaction, then commits the message’s offset to Kafka. This guarantees exactly-once processing because:

If transaction execution fails, the offset isn’t committed, so Kafka retries the message later.
If a duplicate message arrives, DBOS Transact checks the idempotency key and doesn’t re-execute the transaction (the transaction logic also handles corner cases where the message arrives twice simultaneously–a subject for a future blog post).

Exactly-once workflows

If you’re processing the message with a workflow, DBOS Transact synchronously records the idempotency key, commits the message’s offset to Kafka, then asynchronously processes the workflow. This guarantees exactly-once processing because:

If recording the idempotency key fails, the offset isn’t committed, so Kafka will send the message again later.
If the workflow fails, it will always recover and resume from where it left off because DBOS workflows are reliable.
If a duplicate message arrives, DBOS Transact checks the idempotency key and doesn’t start a new workflow.

Getting Started with Kafka and DBOS

Now, let’s see what this looks like in code. The first step to exactly-once semantics is to write your event processing code in TypeScript using DBOS Transact. You can write a single database transaction or a workflow coordinating multiple operations. For now, let’s keep it simple:

import { Workflow, WorkflowContext } from '@dbos-inc/dbos-sdk';

class KafkaExample{
  @Workflow() 
  static async kafkaWorkflow(ctxt: WorkflowContext, topic: string, partition: number, message: KafkaMessage) {      
    ctxt.logger.info(`Message received: ${message.value?.toString()}`) 
  }
}

Next, annotate your method with a @KafkaConsume decorator specifying which topic to consume events from. Additionally, annotate your class with an @Kafka decorator defining which brokers to connect to. That’s all there is to it! When you start your application, DBOS Transact will invoke your method exactly-once for each message sent to the topic.

import { KafkaConfig, KafkaMessage} from "kafkajs";
import { Workflow, WorkflowContext, Kafka, KafkaConsume } from '@dbos-inc/dbos-sdk';

@Kafka({ brokers: ['localhost:9092']})
class KafkaExample{
  @KafkaConsume("example-topic")
  @Workflow()
  static async kafkaWorkflow(ctxt: WorkflowContext, topic: string, partition: number, message: KafkaMessage) {
    ctxt.logger.info(`Message received: ${message.value?.toString()}`) 
  }
}

If you need more control, you can pass KafkaJS configuration objects to both the class and method Kafka decorators. For example, you can specify a custom consumer group ID:

@KafkaConsume("example-topic", { groupId: "custom-group-id" })
@Workflow()
static async kafkaWorkflow(ctxt: WorkflowContext, topic: string, partition: number, message: KafkaMessage) { 
  ctxt.logger.info(`Message received: ${message.value?.toString()}`)
}

To get started with DBOS Transact, check out the quickstart and docs. For more information on using Kafka with DBOS Transact, check out this docs page. To join our community:

Check out our code and contribute on GitHub.
Join us on Discord.