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".

English | 简体中文

🚀 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 = await mf.process(
    block_index=0,
    mode=ProcessMode.COMPLETE,
    variables={'name': 'Alice', 'level': 'Intermediate'}
)

Streaming Response

# Stream processing for real-time responses
async 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 = await 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]: ...

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

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):
    PROMPT_ONLY = "prompt_only"  # Generate prompts without LLM calls
    COMPLETE = "complete"        # Non-streaming LLM processing
    STREAM = "stream"           # Streaming LLM responses

Usage:

# Generate prompt only
prompt_result = await mf.process(0, ProcessMode.PROMPT_ONLY)
print(prompt_result.content)  # Raw prompt text

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

# Streaming response
async 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 AsyncGenerator

class LLMProvider(ABC):
    @abstractmethod
    async def complete(self, prompt: str) -> LLMResult: ...

    @abstractmethod
    async def stream(self, prompt: str) -> AsyncGenerator[str, None]: ...

Custom Implementation:

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

    async def complete(self, prompt: str) -> LLMResult:
        response = await self.client.completions.create(
            model="gpt-3.5-turbo",
            prompt=prompt,
            max_tokens=500
        )
        return LLMResult(content=response.choices[0].text.strip())

    async def stream(self, prompt: str):
        stream = await self.client.completions.create(
            model="gpt-3.5-turbo",
            prompt=prompt,
            stream=True
        )
        async for chunk in stream:
            if chunk.choices[0].text:
                yield chunk.choices[0].text

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: Display buttons without variable assignment
"?[Continue | Cancel | Go Back]"

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 = await 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 = await 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, LLMResult
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.AsyncClient()

    async def complete(self, prompt: str) -> LLMResult:
        response = await 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 LLMResult(content=data["text"])

    async def stream(self, prompt: str):
        async with self.client.stream(
            "POST",
            f"{self.base_url}/stream",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={"prompt": prompt}
        ) as response:
            async for chunk in response.aiter_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

async 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 = await 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
import asyncio

async 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

    async for chunk in mf.process(
        block_index=0,
        mode=ProcessMode.STREAM,
        variables={
            'user_name': 'developer',
            'topic': 'async 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

    async 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 = await 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 = await self.handle_interaction(block.content)
                if response:
                    self.user_responses.update(response)

    async def handle_interaction(self, interaction_content: str):
        from markdown_flow.utils 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 await 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)
await builder.start_interaction()

Variable System Deep Dive

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
• Pre-commit for automated code quality checks

📞 Support

• 📖 Documentation
• 🐛 Issue Tracker
• 💬 Discussions