MarkdownFlow Agent (Python)

Python backend parsing toolkit for transforming MarkdownFlow documents into personalized, AI-powered interactive content.

MarkdownFlow (also known as MDFlow or markdown-flow) extends standard Markdown with AI to create personalized, interactive pages. Its tagline is "Write Once, Deliver Personally".

🚀 Quick Start

Install

pip install markdown-flow
# or
pip install -e .  # For development

Basic Usage

from markdown_flow import MarkdownFlow, ProcessMode

# Simple content processing
document = """
Hello {{name}}! Let's explore your Python skills.

?[%{{level}} Beginner | Intermediate | Expert]

Based on your {{level}} level, here are some recommendations...
"""

mf = MarkdownFlow(document)
variables = mf.extract_variables()  # Returns: {'name', 'level'}
blocks = mf.get_all_blocks()        # Get parsed document blocks

LLM Integration

from markdown_flow import MarkdownFlow, ProcessMode
from your_llm_provider import YourLLMProvider

# Initialize with LLM provider
llm_provider = YourLLMProvider(api_key="your-key")
mf = MarkdownFlow(document, llm_provider=llm_provider)

# Process with different modes
result = mf.process(
    block_index=0,
    mode=ProcessMode.COMPLETE,
    variables={'name': 'Alice', 'level': 'Intermediate'}
)

Streaming Response

# Stream processing for real-time responses
for chunk in mf.process(
    block_index=0,
    mode=ProcessMode.STREAM,
    variables={'name': 'Bob'}
):
    print(chunk.content, end='')

Interactive Elements

# Handle user interactions
document = """
What's your preferred programming language?

?[%{{language}} Python | JavaScript | Go | Other...]

Select your skills (multi-select):

?[%{{skills}} Python||JavaScript||Go||Rust]

?[Continue | Skip]
"""

mf = MarkdownFlow(document)
blocks = mf.get_all_blocks()

for block in blocks:
    if block.block_type == BlockType.INTERACTION:
        # Process user interaction
        print(f"Interaction: {block.content}")

# Process user input
user_input = {
    'language': ['Python'],                    # Single selection
    'skills': ['Python', 'JavaScript', 'Go']  # Multi-selection
}

result = mf.process(
    block_index=1,  # Process skills interaction
    user_input=user_input,
    mode=ProcessMode.COMPLETE
)

📖 API Reference

Core Classes

MarkdownFlow

Main class for parsing and processing MarkdownFlow documents.

class MarkdownFlow:
    def __init__(
        self,
        content: str,
        llm_provider: Optional[LLMProvider] = None
    ) -> None: ...

    def get_all_blocks(self) -> List[Block]: ...
    def extract_variables(self) -> Set[str]: ...

    def process(
        self,
        block_index: int,
        mode: ProcessMode = ProcessMode.COMPLETE,
        variables: Optional[Dict[str, str]] = None,
        user_input: Optional[str] = None
    ) -> LLMResult | Generator[LLMResult, None, None]: ...

Methods:

• get_all_blocks() - Parse document into structured blocks
• extract_variables() - Extract all {{variable}} and %{{variable}} patterns
• process() - Process blocks with LLM using unified interface

Example:

mf = MarkdownFlow("""
# Welcome {{name}}!

Choose your experience: ?[%{{exp}} Beginner | Expert]

Your experience level is {{exp}}.
""")

print("Variables:", mf.extract_variables())  # {'name', 'exp'}
print("Blocks:", len(mf.get_all_blocks()))   # 3

ProcessMode

Processing mode enumeration for different use cases.

class ProcessMode(Enum):
    COMPLETE = "complete"  # Non-streaming LLM processing
    STREAM = "stream"      # Streaming LLM responses

Usage:

# Complete response
complete_result = mf.process(0, ProcessMode.COMPLETE)
print(complete_result.content)  # Full LLM response

# Streaming response
for chunk in mf.process(0, ProcessMode.STREAM):
    print(chunk.content, end='')

LLMProvider

Abstract base class for implementing LLM providers.

from abc import ABC, abstractmethod
from typing import Generator

class LLMProvider(ABC):
    @abstractmethod
    def complete(self, messages: list[dict[str, str]]) -> str: ...

    @abstractmethod
    def stream(self, messages: list[dict[str, str]]) -> Generator[str, None, None]: ...

Custom Implementation:

class OpenAIProvider(LLMProvider):
    def __init__(self, api_key: str):
        self.client = openai.OpenAI(api_key=api_key)

    def complete(self, messages: list[dict[str, str]]) -> str:
        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=messages
        )
        return response.choices[0].message.content

    def stream(self, messages: list[dict[str, str]]):
        stream = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=messages,
            stream=True
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

Block Types

BlockType

Enumeration of different block types in MarkdownFlow documents.

class BlockType(Enum):
    CONTENT = "content"                    # Regular markdown content
    INTERACTION = "interaction"            # User interaction blocks (?[...])
    PRESERVED_CONTENT = "preserved_content" # Content wrapped in === (inline) or !=== (multiline) markers

Block Structure:

# Content blocks - processed by LLM
"""
Hello {{name}}! Welcome to our platform.
"""

# Interaction blocks - user input required
"""
?[%{{choice}} Option A | Option B | Enter custom option...]
"""

# Preserved content - output as-is
"""
# Inline format (single line)
===Fixed title===

# Multiline fence with leading '!'
!===
This content is preserved exactly as written.
No LLM processing or variable replacement.
!===
"""

Interaction Types

InteractionType

Parsed interaction format types.

class InteractionType(NamedTuple):
    name: str                    # Type name
    variable: Optional[str]      # Variable to assign (%{{var}})
    buttons: List[str]          # Button options
    question: Optional[str]      # Text input question
    has_text_input: bool        # Whether text input is allowed

Supported Formats:

# TEXT_ONLY: Text input with question
"?[%{{name}} What is your name?]"

# BUTTONS_ONLY: Button selection only
"?[%{{level}} Beginner | Intermediate | Expert]"

# BUTTONS_WITH_TEXT: Buttons with fallback text input
"?[%{{preference}} Option A | Option B | Please specify...]"

# BUTTONS_MULTI_SELECT: Multi-select buttons
"?[%{{skills}} Python||JavaScript||Go||Rust]"

# BUTTONS_MULTI_WITH_TEXT: Multi-select with text fallback
"?[%{{frameworks}} React||Vue||Angular||Please specify others...]"

# NON_ASSIGNMENT_BUTTON: No-variable interaction. Supports the same shapes as
# variable interactions; the answer is not assigned to a variable but fed back
# into the conversation context (see "Non-assignment interactions" below).
"?[Continue | Cancel | Go Back]"
"?[JavaScript||TypeScript||Python]"
"?[...Anything to add?]"
"?[Option A | Option B | ...Other, please specify]"

Non-assignment interactions

Interactions without a %{{variable}} collect an answer that is not stored in any variable — it flows back into the conversation context instead:

1. Clients submit the answer under the canonical key input (DEFAULT_NON_ASSIGNMENT_INPUT_KEY); the library tolerantly merges values from any other key.

2. process() normalizes values matching a button (by display or value) to the button value and accepts anything else as free text. The normalized answer is returned in result.metadata["answer"] (list[str]).

3. Callers persist that answer and attach it to the interaction's context message via the user_answer extension field (USER_ANSWER_CONTEXT_KEY):

   context = [
       {"role": "assistant", "content": "?[Option A | Option B]", "user_answer": "Option A"},
   ]
   

   When building LLM messages the library expands this into {user: "Option A"} + {assistant: "ok"}. An empty user_answer skips the turn entirely; omitting the field keeps the legacy {user: "ok"} behavior. Extension fields are stripped before messages reach the LLM.

Utility Functions

Variable Operations

def extract_variables_from_text(text: str) -> Set[str]:
    """Extract all {{variable}} and %{{variable}} patterns."""

def replace_variables_in_text(text: str, variables: dict) -> str:
    """Replace {{variable}} patterns with values, preserve %{{variable}}."""

# Example
text = "Hello {{name}}! Choose: ?[%{{level}} Basic | Advanced]"
vars = extract_variables_from_text(text)  # {'name', 'level'}
result = replace_variables_in_text(text, {'name': 'Alice'})
# Returns: "Hello Alice! Choose: ?[%{{level}} Basic | Advanced]"

Interaction Processing

def InteractionParser.parse(content: str) -> InteractionType:
    """Parse interaction block into structured format."""

def extract_interaction_question(content: str) -> str:
    """Extract question text from interaction block."""

def generate_smart_validation_template(interaction_type: InteractionType) -> str:
    """Generate validation template for interaction."""

# Example
parser_result = InteractionParser.parse("%{{choice}} A | B | Enter custom...")
print(parser_result.name)          # "BUTTONS_WITH_TEXT"
print(parser_result.variable)      # "choice"
print(parser_result.buttons)       # ["A", "B"]
print(parser_result.question)      # "Enter custom..."

Types and Models

# Core data structures
from dataclasses import dataclass
from typing import Optional, List, Dict, Set

@dataclass
class Block:
    content: str
    block_type: BlockType
    index: int

@dataclass
class LLMResult:
    content: str
    metadata: Optional[Dict] = None

# Variable system types
Variables = Dict[str, str]  # Variable name -> value mapping

# All types are exported for use
from markdown_flow import (
    Block, LLMResult, Variables,
    BlockType, InteractionType, ProcessMode
)

🔄 Migration Guide

Parameter Format Upgrade

The new version introduces multi-select interaction support with improvements to the user_input parameter format.

Old Format

# Single string input
user_input = "Python"

# Process interaction
result = mf.process(
    block_index=1,
    user_input=user_input,
    mode=ProcessMode.COMPLETE
)

New Format

# Dictionary format with list values
user_input = {
    'language': ['Python'],                    # Single selection as list
    'skills': ['Python', 'JavaScript', 'Go']  # Multi-selection
}

# Process interaction
result = mf.process(
    block_index=1,
    user_input=user_input,
    mode=ProcessMode.COMPLETE
)

New Multi-Select Syntax

<!-- Single select (traditional) -->
?[%{{language}} Python|JavaScript|Go]

<!-- Multi-select (new) -->
?[%{{skills}} Python||JavaScript||Go||Rust]

<!-- Multi-select with text fallback -->
?[%{{frameworks}} React||Vue||Angular||Please specify others...]

Variable Types

# Variables now support both string and list values
variables = {
    'name': 'John',                           # str (traditional)
    'skills': ['Python', 'JavaScript'],      # list[str] (new)
    'experience': 'Senior'                    # str (traditional)
}

🧩 Advanced Examples

Custom LLM Provider Integration

from markdown_flow import MarkdownFlow, LLMProvider
import httpx

class CustomAPIProvider(LLMProvider):
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.api_key = api_key
        self.client = httpx.Client()

    def complete(self, messages: list[dict[str, str]]) -> str:
        # Convert messages to your API format
        prompt = "\n".join([f"{msg['role']}: {msg['content']}" for msg in messages])

        response = self.client.post(
            f"{self.base_url}/complete",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={"prompt": prompt, "max_tokens": 1000}
        )
        data = response.json()
        return data["text"]

    def stream(self, messages: list[dict[str, str]]):
        # Convert messages to your API format
        prompt = "\n".join([f"{msg['role']}: {msg['content']}" for msg in messages])

        with self.client.stream(
            "POST",
            f"{self.base_url}/stream",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={"prompt": prompt}
        ) as response:
            for chunk in response.iter_text():
                if chunk.strip():
                    yield chunk

# Usage
provider = CustomAPIProvider("https://api.example.com", "your-key")
mf = MarkdownFlow(document, llm_provider=provider)

Multi-Block Document Processing

def process_conversation():
    conversation = """
# AI Assistant

Hello {{user_name}}! I'm here to help you learn Python.

---

What's your current experience level?

?[%{{experience}} Complete Beginner | Some Experience | Experienced]

---

Based on your {{experience}} level, let me create a personalized learning plan.

This plan will include {{topics}} that match your background.

---

Would you like to start with the basics?

?[Start Learning | Customize Plan | Ask Questions]
"""

    mf = MarkdownFlow(conversation, llm_provider=your_provider)
    blocks = mf.get_all_blocks()

    variables = {
        'user_name': 'Alice',
        'experience': 'Some Experience',
        'topics': 'intermediate concepts and practical projects'
    }

    for i, block in enumerate(blocks):
        if block.block_type == BlockType.CONTENT:
            print(f"\n--- Processing Block {i} ---")
            result = mf.process(
                block_index=i,
                mode=ProcessMode.COMPLETE,
                variables=variables
            )
            print(result.content)
        elif block.block_type == BlockType.INTERACTION:
            print(f"\n--- User Interaction Block {i} ---")
            print(block.content)

Streaming with Progress Tracking

from markdown_flow import MarkdownFlow, ProcessMode

def stream_with_progress():
    document = """
Generate a comprehensive Python tutorial for {{user_name}}
focusing on {{topic}} with practical examples.

Include code samples, explanations, and practice exercises.
"""

    mf = MarkdownFlow(document, llm_provider=your_provider)

    print("Starting stream processing...")
    content = ""
    chunk_count = 0

    for chunk in mf.process(
        block_index=0,
        mode=ProcessMode.STREAM,
        variables={
            'user_name': 'developer',
            'topic': 'synchronous programming'
        }
    ):
        content += chunk.content
        chunk_count += 1

        # Show progress
        if chunk_count % 10 == 0:
            print(f"Received {chunk_count} chunks, {len(content)} characters")

        # Real-time processing
        if chunk.content.endswith('\n'):
            # Process complete line
            lines = content.strip().split('\n')
            if lines:
                latest_line = lines[-1]
                # Do something with complete line
                pass

    print(f"\nStreaming complete! Total: {chunk_count} chunks, {len(content)} characters")
    return content

Interactive Document Builder

from markdown_flow import MarkdownFlow, BlockType, InteractionType

class InteractiveDocumentBuilder:
    def __init__(self, template: str, llm_provider):
        self.mf = MarkdownFlow(template, llm_provider)
        self.user_responses = {}
        self.current_block = 0

    def start_interaction(self):
        blocks = self.mf.get_all_blocks()

        for i, block in enumerate(blocks):
            if block.block_type == BlockType.CONTENT:
                # Process content block with current variables
                result = self.mf.process(
                    block_index=i,
                    mode=ProcessMode.COMPLETE,
                    variables=self.user_responses
                )
                print(f"\nContent: {result.content}")

            elif block.block_type == BlockType.INTERACTION:
                # Handle user interaction
                response = self.handle_interaction(block.content)
                if response:
                    self.user_responses.update(response)

    def handle_interaction(self, interaction_content: str):
        from markdown_flow.parser import InteractionParser

        interaction = InteractionParser().parse(interaction_content)
        print(f"\n{interaction_content}")

        if interaction.name == "BUTTONS_ONLY":
            print("Choose an option:")
            for i, button in enumerate(interaction.buttons, 1):
                print(f"{i}. {button}")

            choice = input("Enter choice number: ")
            try:
                selected = interaction.buttons[int(choice) - 1]
                return {interaction.variable: selected}
            except (ValueError, IndexError):
                print("Invalid choice")
                return self.handle_interaction(interaction_content)

        elif interaction.name == "TEXT_ONLY":
            response = input(f"{interaction.question}: ")
            return {interaction.variable: response}

        return {}

# Usage
template = """
Welcome! Let's create a personalized learning plan.

What's your name?
?[%{{name}} Enter your name]

Hi {{name}}! What would you like to learn?
?[%{{subject}} Python | JavaScript | Data Science | Machine Learning]

Great choice, {{name}}! {{subject}} is an excellent field to study.
"""

builder = InteractiveDocumentBuilder(template, your_llm_provider)
builder.start_interaction()

Variable System Deep Dive

Text variable names

Names are opaque Unicode text. Python str.strip() removes surrounding whitespace; the remaining name must be nonempty. Internal spaces, punctuation, emoji, single braces, brackets, |, //, and ... stay exactly as written. Internal control characters (U+0000–001F and U+007F–009F) and U+2028/U+2029 are invalid. Case, repeated spaces, and Unicode sequences are never normalized; U+FEFF remains part of the name.

The first literal }} closes a marker. A name cannot contain that pair. Quotes and backslashes are literal characters with no decoding rule, and nested {{ inside a name is not another marker. Every complete {{ ... }} or %{{ ... }} is read before interaction delimiters, so internal ], |, //, and ... stay literal. Only a leading assignment marker sets the interaction's variable; markers elsewhere in its body remain literal option text. An unterminated marker keeps the remaining text opaque, without restarting at an inner ?[.

At document level, complete inline code, inline math, link resources, URI/email angle autolinks, indented code, and raw-text HTML bodies (script, style, pre, textarea) bound local variable examples. Quote/list-contained fences and ordinary HTML blocks also keep unfinished examples local. Reference-definition destinations and titles retain their complete variable examples. Physical CRLF, LF and CR boundaries preserve original offsets and marker text. Complete variable examples there still support collection and replacement, but an unclosed example cannot consume text beyond that markup span. Escaped document openers stay literal. A true raw marker owns its entire name before any code or HTML inside that name is considered; interaction bodies keep the raw marker grammar.

Use format_variable_reference(name) and format_variable_assignment(name) from markdown_flow.variable_names to emit padded markers such as {{ What is your gender? }} and %{{ a} }}. Padding lets a name end in a single brace. Extraction and replacement use the same trimmed key; existing stored keys are not renamed or migrated.

The shared conformance corpus is scripts/variable-name-contract.json. Run its offline parser, document, replacement, translation, and answer round trips with python -m pytest scripts/test_variable_names.py -q.

from markdown_flow import extract_variables_from_text, replace_variables_in_text

def demonstrate_variable_system():
    # Complex document with both variable types
    document = """
    Welcome {{user_name}} to the {{course_title}} course!

    Please rate your experience: ?[%{{rating}} 1 | 2 | 3 | 4 | 5]

    Current progress: {{progress_percent}}%
    Assignment due: {{due_date}}

    Your rating of %{{rating}} helps us improve the course content.
    """

    # Extract all variables
    all_vars = extract_variables_from_text(document)
    print(f"All variables found: {all_vars}")
    # Output: {'user_name', 'course_title', 'rating', 'progress_percent', 'due_date'}

    # Replace only {{variable}} patterns, preserve %{{variable}}
    replacements = {
        'user_name': 'Alice',
        'course_title': 'Python Advanced',
        'progress_percent': '75',
        'due_date': '2024-12-15',
        'rating': '4'  # This won't be replaced due to %{{}} format
    }

    result = replace_variables_in_text(document, replacements)
    print("\nAfter replacement:")
    print(result)

    # The %{{rating}} remains unchanged for LLM processing,
    # while {{user_name}}, {{course_title}}, etc. are replaced

demonstrate_variable_system()

🌐 MarkdownFlow Ecosystem

markdown-flow-agent-py is part of the MarkdownFlow ecosystem for creating personalized, AI-driven interactive documents:

• markdown-flow - Main repository with homepage, documentation, and interactive playground
• markdown-flow-ui - React component library for rendering interactive MarkdownFlow documents
• markdown-it-flow - markdown-it plugin to parse and render MarkdownFlow syntax
• remark-flow - Remark plugin to parse and process MarkdownFlow syntax in React applications

💖 Sponsors

AI-Shifu.com

AI-powered personalized learning platform

📄 License

MIT License - see LICENSE file for details.

🙏 Acknowledgments

• Python for the robust programming language
• Ruff for lightning-fast Python linting and formatting
• MyPy for static type checking
• Commitizen for standardized commit messages
• Lefthook for automated code quality checks (git hooks + CI)

📞 Support

• 📖 Documentation
• 🐛 Issue Tracker
• 💬 Discussions