5 Patterns. Infinite Apps.

Template

# Multi-Agent Pipeline Template

class MultiAgentPipeline:
    def __init__(self, agents: List[Agent], orchestrator: Orchestrator):
        self.agents = agents
        self.orchestrator = orchestrator
    
    def run(self, input: RawInput) -> StructuredOutput:
        context = Context(input=input)
        
        for agent in self.agents:
            result = agent.execute(context)
            context.add_result(agent.name, result)
            
        return self.orchestrator.format(context)

# Agent Definition
class Agent:
    name: str
    prompt_template: str
    output_schema: Schema
    tools: List[Tool]
    
    def execute(self, context: Context) -> Result:
        # 1. Build prompt from template + context
        # 2. Call LLM with structured output enforcement
        # 3. Validate against schema
        # 4. Return typed result
        pass

# Orchestrator
class Orchestrator:
    def format(self, context: Context) -> StructuredOutput:
        # Combine all agent outputs into final format
        # Generate Mermaid diagrams
        # Export: Markdown, Confluence, Notion, JSON
        pass

BA Assistant Agent Config

Key Implementation Details

• Context passing: Each agent receives full context from previous agents
• Structured output: Pydantic/JSON Schema enforcement on every agent
• Mermaid validation: Syntax check + render test before output
• Human-in-loop: Checkpoint after Agent 1 & 3 for review
• Streaming: Token-by-token UI updates via Streamlit callbacks

Template

# Vision-to-Structured Template

class VisionToStructured:
    def __init__(self, vision_model: VisionLLM, parser: StructuredParser):
        self.vision = vision_model
        self.parser = parser
    
    def process(self, image: Image) -> StructuredSpecs:
        # 1. Vision analysis
        analysis = self.vision.analyze(image, prompt=VISION_PROMPT)
        
        # 2. Structured extraction
        specs = self.parser.extract(analysis)
        
        # 3. Diagram generation
        diagrams = self.generate_diagrams(specs)
        
        return SpecsOutput(requirements=specs, diagrams=diagrams)

VISION_PROMPT = """
Analyze this image (whiteboard/sketch/diagram/screenshot) and extract:
1. TEXT: All readable text (OCR)
2. SHAPES: Boxes, arrows, circles, lines, connections
3. LAYOUT: Hierarchy, grouping, flow direction
4. INTENT: What type of diagram? (process flow, architecture, UI, data model)

Return structured JSON with: elements[], relationships[], metadata{}
"""

Supported Input Types

• Whiteboard photos (messy handwriting, shadows, angles)
• Process flow diagrams (BPMN-ish, flowcharts)
• Architecture sketches (boxes + arrows)
• UI wireframes / mockups
• Data model diagrams (ER-ish)

Template

# Streaming Wrapper Template

class StreamingAPIWrapper:
    def __init__(self, client: APIClient, config: StreamingConfig):
        self.client = client
        self.config = config
        self.buffer = ByteBuffer()
    
    async def stream(self, request: Request) -> AsyncGenerator[Chunk, None]:
        async for chunk in self.client.stream(request):
            # Transform / validate / enrich
            processed = self.process_chunk(chunk)
            yield processed
            
            # Update UI in real-time
            self.update_ui(processed)
    
    def process_chunk(self, chunk: Chunk) -> Chunk:
        # Add metadata, check errors, transform format
        return chunk

# Streamlit Integration
def streamlit_streaming_ui(wrapper: StreamingAPIWrapper, request: Request):
    placeholder = st.empty()
    audio_buffer = BytesIO()
    
    async for chunk in wrapper.stream(request):
        audio_buffer.write(chunk.data)
        # Play progressively
        placeholder.audio(audio_buffer.getvalue(), format='audio/wav')
    
    # Final: offer download
    st.download_button("Download", audio_buffer.getvalue(), "output.wav")

Sarvam TTS Config

Template

# Graph-to-Mermaid Template

class GraphToMermaid:
    def __init__(self, extractor: EntityExtractor, mapper: RelationshipMapper):
        self.extractor = extractor
        self.mapper = mapper
    
    def build(self, source: Source) -> MermaidGraph:
        # 1. Extract entities
        entities = self.extractor.extract(source)
        
        # 2. Map relationships
        relationships = self.mapper.map(entities, source)
        
        # 3. Build graph
        graph = nx.DiGraph()
        graph.add_nodes_from((e.id, e) for e in entities)
        graph.add_edges_from((r.source, r.target, r) for r in relationships)
        
        # 4. Generate Mermaid
        mermaid = self.to_mermaid(graph)
        return MermaidGraph(graph=graph, mermaid=mermaid, entities=entities)
    
    def to_mermaid(self, graph: nx.DiGraph) -> str:
        lines = ["graph TD"]
        for node, data in graph.nodes(data=True):
            label = data['label'].replace('"', '\\"')
            lines.append(f'  {node}["{label}"]')
        for u, v, data in graph.edges(data=True):
            rel = data.get('type', 'relates_to')
            lines.append(f'  {u} -->|{rel}| {v}')
        return '\n'.join(lines)

Relationship Types Supported

• Family: parent, child, sibling, spouse
• Romance: dating, married, ex, unrequited
• Conflict: rivalry, enemy, betrayal
• Alliance: friend, ally, mentor, teammate
• Professional: boss, subordinate, colleague, partner

Template

# Multi-API Aggregation Template

class APIAggregator:
    def __init__(self, apis: List[APIClient], normalizer: Normalizer, cache: Cache):
        self.apis = apis
        self.normalizer = normalizer
        self.cache = cache
    
    async def search(self, query: SearchQuery) -> AggregatedResults:
        # 1. Check cache
        cached = await self.cache.get(query.key)
        if cached: return cached
        
        # 2. Parallel fetch
        tasks = [api.search(query) for api in self.apis]
        raw_results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # 3. Normalize
        normalized = []
        for api, results in zip(self.apis, raw_results):
            if isinstance(results, Exception): continue
            normalized.extend(self.normalizer.normalize(api.name, results))
        
        # 4. Rank & dedupe
        ranked = self.rank(normalized, query)
        deduped = self.dedupe(ranked)
        
        # 5. Cache & return
        await self.cache.set(query.key, deduped)
        return deduped
    
    def rank(self, results: List[Result], query: SearchQuery) -> List[Result]:
        # Score: relevance * freshness * source_reliability * user_preference
        return sorted(results, key=lambda r: r.score, reverse=True)

Streaming Finder Sources