Pulso AI

Content operations at any small team involve an enormous amount of repetitive cognitive work — reading RSS feeds, classifying articles, deciding what's worth sharing, writing summaries, formatting for different channels. Pulso automates the full pipeline from signal ingestion to multi-channel publication. The system runs as a set of composable workflow nodes orchestrated by n8n, with a Python FastAPI service handling the LLM-heavy classification and generation steps. OpenAI's structured output API ensures every model response is a predictable JSON object — no prompt parsing hacks.

80%

Time Saved

vs. manual operations

200+

Throughput

items processed / week

Channels

Slack · Newsletter · DB

Architecture: Orchestration + Inference Separation

The key architectural decision was to separate orchestration from inference. n8n handles scheduling, trigger management, retry logic, and inter-service communication. The Python FastAPI service handles only the computationally expensive LLM calls. This means the LLM layer is independently scalable and testable, and n8n provides a visual audit trail of every workflow execution without custom logging code.

PythonStructured output classification — guaranteed JSON response shape

class ContentClassification(BaseModel):
    intent: Literal["product-update", "tutorial", "opinion", "news"]
    relevance_score: float  # 0.0 – 1.0
    suggested_tags: list[str]
    publish_to: list[Literal["slack", "newsletter", "archive"]]

async def classify(raw: str) -> ContentClassification:
    response = await openai.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": CLASSIFY_PROMPT + raw}],
        response_format=ContentClassification,
    )
    return response.choices[0].message.parsed

Reliability: Idempotent Processing

Every content item is stored with a content hash before processing begins. If a workflow fails mid-execution and retries, the hash check prevents duplicate publications. Items are marked as processed atomically after all channel publishers confirm success — partial failures trigger a full retry of the publication step only, not reclassification.

Python + FastAPILLM inference service, classification, structured generation

n8nWorkflow orchestration, scheduling, retry logic, visual audit

OpenAI APIGPT-4o-mini with structured output for reliable JSON responses

SupabaseProcessed items store, deduplication, audit log

DockerContainerised deployment, compose for local dev parity

Pulso AI workflow execution log