AI Workflow in Python: Build Steps You Can Test

A Python AI workflow is a sequence of steps where each step calls a model, a tool, or both, and the output of one step becomes the input of the next. Python is a practical choice for this because the same language handles orchestration, data processing, and testing without switching contexts.

The design goal is testability: a workflow where each step has a validated input and a checkable output is one you can trust before it touches production. Tools like fenbs hold the shared task context and lessons alongside the code — not inside the workflow runtime, but in a durable record every connected assistant can read before similar work. Python-specific AI tools and LangChain workflows are covered on their own pages.

Structure the workflow as a sequence of typed steps

Each step is a function that takes a typed input and returns a typed output. Use Pydantic or Python dataclasses to define the types. A step that takes and returns untyped dictionaries is a step that fails at runtime instead of at the type checker.

The step signature documents the contract: what it needs, what it produces, and what it is allowed to call. A step that only needs a query string and a context block should not receive the entire workflow state. Narrow inputs make failures easier to isolate.

from pydantic import BaseModel
from typing import Literal

class ExtractInput(BaseModel):
    text: str
    task_id: str

class ExtractOutput(BaseModel):
    entities: list[str]
    confidence: float
    status: Literal["ok", "low_confidence"]

def extract_entities(inp: ExtractInput) -> ExtractOutput:
    # call the model here
    ...

Validated inputs before calling the model

Validate inputs at the step boundary, before the model call. A missing required field or a value outside the expected range is caught immediately, not buried inside a traceback from inside the model call.

Log the validated input before calling the model. When a step produces an unexpected output, you want to replay the same input exactly. An input logged as a JSON record is replayable; an input reconstructed from memory is not.

Recorded outputs and approval boundaries

Record the output of each step before passing it to the next one. A dictionary in memory is not a record. Write it to a file, a database row, or a structured log entry — something that survives a crash.

After recording, check whether the step requires human approval before continuing. An approval boundary is a state field: the workflow writes the proposed action and sets a status of "pending." It does not continue until a person sets the status to "approved."

Implement the check as a simple poll or a webhook. The workflow writes the pending state to a store, then either polls at intervals or listens for a webhook callback. When the approval arrives, it reads the approved state and continues.

Testing the workflow

Test each step as a function: give it a fixed input, call it, and assert that the output matches the expected schema and values. These tests run without making real model calls when you use a mock or a recorded response.

Build a small fixture set: two or three real inputs with their expected outputs and the reason each was chosen. Run the fixture set whenever the workflow changes. An output that changed unexpectedly is a regression until you understand why.

For the approval boundary, test that the workflow stops and waits when a step produces a "pending" state. Test that it continues correctly after the approval is recorded. The AI agent approval workflow guide has the full pattern.

Lessons from the workflow

A workflow run that fails in an unexpected way reveals an assumption the design made. A step that passes the schema check but produces a result a person would not use reveals a gap in the evaluation set. Both are worth recording.

Record discoveries where the next developer will read them. A lesson that explains why a particular validation rule was added, or why an approval boundary was placed before a specific step, saves the next person from repeating the same experiment.

How to do this with fenbs

1. Create a free account and add your Python workflow as a project in fenbs. List each step as a task with its input type, output type, and approval requirement.
2. Write the approved rules in Decisions and rules: which steps need sign-off, what the input validation rules are, which models are authorized. Connect your AI assistant from Integrations so it reads these before each run.
3. At the start of each run, the workflow reads the project context from fenbs: rules first, then the active task and its plan, then Where we left off.
4. After each run, update the task in fenbs: what was completed, what passed validation, and where the next run starts. Record any approval decisions made.
5. When a run reveals a gap, record it as a lesson in fenbs. The next assistant doing similar Python workflow work reads it before starting.

fenbs does not manage the workflow's runtime state or the model API calls. It holds the human decisions and task record that give the workflow its context before it starts and its history after it ends. That record is available to any connected assistant, in any session, regardless of which model runs the next step.