DEV Community: Qian Li

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 😊

How to Deploy an AI App (w/ Large Deps) to AWS Lambda

Qian Li — Wed, 09 Oct 2024 16:40:53 +0000

I recently spent two hours getting a simple LlamaIndex app to run on AWS Lambda. While the function itself consists of just a few lines of Python code (as shown below), managing dependencies and deployment can be tricky.

import json
from llama_index.llms.openai import OpenAI

llm=OpenAI(model="gpt-4o-mini")

def lambda_handler(event, context):
    response = llm.complete("What public transportation might be available in a city?")
    return {
        'statusCode': 200,
        'body': str(response),
    }

Here are some key tips that helped me:

First, install packages for the correct platform. It's important to install all packages for the "manylinux2014_x86_64" target platform. Otherwise, certain packages might be incompatible with the AWS Lambda runtime. To ensure compatibility, install dependencies with the following command:

pip install -r requirements.txt --platform manylinux2014_x86_64 --target ./deps --only-binary=:all:

Second, Lambda has a 250MB limit for the total code package size, which can quickly become an issue if you're using LlamaIndex or other large dependencies. If your package exceeds this limit, check the largest files in your dependencies folder:

du -h -d 2 | sort -hr | head -n20

In my case, I found that the pandas/tests directory was taking up about 35MB of space, which was unnecessary for my function, so I removed it to bring the package size back within the limit.

rm -r deps/pandas/tests

Then, zip everything up and upload via S3. After trimming unnecessary files, create a zip archive containing both your code and dependencies. Since Lambda's web console has a 50MB upload limit, you'll need to upload larger zip files to an S3 bucket and use the S3 URI to deploy the function.

zip -r test_lambda.zip data/ lambda_function.py
cd deps/
zip -r ../test_lambda.zip .

Finally, adjust your Lambda settings before deploying. By default, Lambda functions are allocated only 128MB of memory and a 3-second timeout, which is insufficient for many AI applications involving large dependencies and calling LLMs. I'd increase the memory to 512MB and extend the timeout to 30 seconds. Additionally, don't forget to set essential environment variables like your OpenAI API Key.

It took me quite a few tries to figure out the correct way to install packages and bundle everything together for Lambda. AWS Lambda is user-friendly for basic scripts, but once you add larger dependencies, things get more complicated.

Here's the final sequence of steps:

# Install dependencies
pip install -r requirements.txt --platform manylinux2014_x86_64 --target ./deps --only-binary=:all:

# Create a zip file for code and data
zip -r test_lambda.zip data/ lambda_function.py

# Include dependencies in the zip file, while removing large unused files
cd deps/
rm -r pandas/tests
zip -r ../test_lambda.zip .

p.s., I also tried deploying a similar function on DBOS Cloud, and it only took a single command:

dbos-cloud app deploy

In DBOS, the dependency management was handled automatically via the requirements.txt file, and environment variables were set in dbos-config.yaml. I might be biased, but I enjoy the simplicity of DBOS Cloud's deployment process.

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 😊

Build reliable Slack apps

Qian Li — Fri, 13 Sep 2024 22:55:17 +0000

Building a Slack app is fun! But is your app reliable?
While building one myself, I noticed two common issues in popular open-source Slack apps:

Many apps process events synchronously, which can lead to timeouts. Slack expects responses within 3 seconds, but if your app triggers AI/RAG pipelines, then AI models may take longer to generate a reply (e.g., the new o1 model can take ~10 seconds to "think"). Slack’s best practices recommend queuing events and processing them asynchronously.
Many apps don't handle duplicate events. If your app fails to respond, Slack retries the event three times. Without proper handling, retries can cause duplicate or inconsistent responses from the app. This leads to bad user experiences.

Here's how I solve them with DBOS Python, an open-source lightweight durable execution library. I started from an off-the-shelf AI/RAG-based Slack app demo (llamabot from LlamaIndex), lightly modified and annotated functions so each incoming message starts a DBOS workflow.

The message dispatch code is simple:

@slackapp.message()
def handle_message(request: BoltRequest) -> None:
    DBOS.logger.info(f"Received message: {request.body}")
    event_id = request.body["event_id"]
    # Use the unique event_id as an idempotency key to guarantee each message is processed exactly-once
    with SetWorkflowID(event_id):
        # Start the event processing workflow in the background then respond to Slack.
        # We can't wait for the workflow to finish because Slack expects the
        # endpoint to reply within 3 seconds.
        DBOS.start_workflow(message_workflow, request.body["event"])

The workflow is initiated in the background, allowing my app to respond to Slack quickly. DBOS workflows always run to completion once started (even running asynchronously). Thus, messages are always processed reliably.
I use the message’s event ID as the workflow's idempotency key, so DBOS uses it to ensure each message is processed exactly once.

You can find more details about the AI-powered Slack app I built in this GitHub repo: https://github.com/dbos-inc/dbos-demo-apps/tree/main/python/llamabot

The README contains detailed instructions on how you could directly use this app in your Slack workspace.

How do you usually build reliable applications? Do you have any feedback for this app? Please let me know!

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.