Use Cases & Examples¶
126 production-ready YAML configurations — from agentic AI orchestration and advanced RAG pipelines to enterprise integration patterns. Deploy intelligent agents, knowledge-powered workflows, and end-to-end AI applications.
Table of Contents¶
- Enterprise Integration Patterns
- Data Transformation & ETL
- Agentic AI — Single Agents
- Agentic AI — Multi-Agent Orchestration
- Agentic AI — Agent Flows
- Agentic AI — Memory Systems
- Agentic AI — Tool Integration
- RAG — Document Ingestion Pipelines
- RAG — Chunking Strategies
- RAG — Embedding & Storage
- RAG — Retrieval Strategies
- RAG — Postprocessing & Reranking
- RAG — Synthesis & Generation
- RAG — Advanced Pipelines
- RAG — Knowledge Graphs
- RAG — Evaluation & Quality
- Multimodal AI
- Protocol Interop — MCP, A2A, AG-UI, A2UI
- Security Patterns
- Production & Enterprise
- End-to-End Application Patterns
- Advanced AI Agent & RAG Patterns
- memx Vector Database Integration
- Code Agent with Built-in Functions
- Browser Automation Agent
1. Enterprise Integration Patterns¶
1.1 Content-Based Router¶
Route messages to different backends based on content.
workflows:
- id: order-router
description: Route orders by type
trigger:
uri: kafka://orders?brokers=kafka:9092&groupId=router
steps:
- choice:
when:
- starlang: 'headers["OrderType"] == "digital"'
steps:
- target: http://digital-fulfillment:8080/process
- starlang: 'headers["OrderType"] == "physical"'
steps:
- target: http://warehouse-api:8080/ship
- starlang: 'headers["OrderType"] == "subscription"'
steps:
- target: http://billing-service:8080/renew
otherwise:
steps:
- target: log://dead-letter?level=WARN
- target: kafka://dead-letter-orders?brokers=kafka:9092
1.2 Scatter-Gather with Aggregation¶
Fan out to multiple services, aggregate responses.
workflows:
- id: price-aggregator
description: Get best price from multiple suppliers
trigger:
uri: http-server://0.0.0.0:8080/api/price-check
steps:
- multicast:
parallel: true
uris:
- http://supplier-a:8080/quote
- http://supplier-b:8080/quote
- http://supplier-c:8080/quote
- aggregate:
correlation: 'headers["RequestId"]'
strategy: starlang
expression: |
results = [read(msg, "application/json") for msg in messages]
best = min(results, key=lambda r: r["price"])
write(best, "application/json")
completion-size: 3
completion-timeout: 5s
1.3 Saga Pattern for Distributed Transactions¶
Coordinate multi-step operations with compensation.
workflows:
- id: booking-saga
description: Book flight + hotel with compensation
trigger:
uri: http-server://0.0.0.0:8080/api/book
steps:
- saga:
propagation: required
steps:
- target: http://flight-service:8080/book
- target: http://hotel-service:8080/reserve
- target: http://payment-service:8080/charge
compensate:
- target: http://flight-service:8080/cancel
- target: http://hotel-service:8080/cancel
- target: http://payment-service:8080/refund
1.4 Circuit Breaker with Fallback¶
Protect against downstream service failures.
workflows:
- id: resilient-api
description: Call external API with circuit breaker
trigger:
uri: http-server://0.0.0.0:8080/api/data
steps:
- circuit-breaker:
threshold: 5
half-open-after: 30s
steps:
- target: http://external-api:8080/data
fallback:
steps:
- set-body:
constant: '{"status":"degraded","source":"cache"}'
- target: redis://cache-server:6379?command=GET&key=${header.CacheKey}
1.5 Idempotent Consumer¶
Prevent duplicate message processing.
workflows:
- id: payment-processor
description: Process payments exactly once
trigger:
uri: kafka://payments?brokers=kafka:9092&groupId=processor
steps:
- idempotent-consumer:
expression:
starlang: 'headers["PaymentId"]'
repository: redis
redis-url: redis://localhost:6379
steps:
- target: http://payment-gateway:8080/process
- target: postgres://db:5432/payments?operation=insert
1.6 Dead Letter Channel with Retry¶
Automatic retries with dead letter queue.
workflows:
- id: reliable-processor
description: Process with retry and DLQ
trigger:
uri: amqp://orders?queue=incoming
error-handler:
type: dead-letter-channel
dead-letter-uri: amqp://orders?queue=dlq
redelivery-policy:
maximum-redeliveries: 3
redelivery-delay: 1s
back-off-multiplier: 2.0
maximum-redelivery-delay: 30s
retry-attempted-log-level: WARN
steps:
- target: http://order-service:8080/process
1.7 Throttle and Rate Limiting¶
Control message throughput.
workflows:
- id: api-gateway
description: Rate-limited API gateway
trigger:
uri: http-server://0.0.0.0:8080/api/v1
steps:
- throttle:
rate: 100
period: 1s
reject-execution: true
- target: http://backend:8080/api/v1
1.8 Wire Tap for Audit¶
Non-blocking audit copy of messages.
workflows:
- id: order-pipeline
description: Process orders with audit trail
trigger:
uri: kafka://orders?brokers=kafka:9092&groupId=main
steps:
- wiretap: kafka://audit-log?brokers=kafka:9092
- target: http://order-service:8080/process
2. Data Transformation & ETL¶
2.1 JSON-to-JSON Transform with Starlang¶
workflows:
- id: transform-orders
description: Normalize order format
trigger:
uri: kafka://raw-orders?brokers=kafka:9092&groupId=etl
steps:
- transform:
starlang: |
order = read(payload, "application/json")
normalized = {
"id": order["order_id"],
"customer": {
"name": order["cust_name"],
"email": order["cust_email"],
},
"items": [
{"sku": item["product_code"], "qty": item["quantity"], "price": float(item["unit_price"])}
for item in order.get("line_items", [])
],
"total": sum(float(i["unit_price"]) * i["quantity"] for i in order.get("line_items", [])),
"currency": order.get("currency", "USD"),
"created_at": order["timestamp"],
}
write(normalized, "application/json")
- target: kafka://normalized-orders?brokers=kafka:9092
2.2 CSV to JSON Conversion¶
workflows:
- id: csv-to-json
description: Convert CSV files to JSON documents
trigger:
uri: file:///data/inbox?include=*.csv&move=../processed
steps:
- unmarshal:
format: csv
- transform:
starlang: |
rows = read(payload, "text/csv")
records = [{"name": r[0], "email": r[1], "department": r[2]} for r in rows[1:]]
write(records, "application/json")
- split:
starlang: 'payload'
steps:
- target: postgres://db:5432/employees?operation=insert&table=employees
2.3 Multi-Format ETL Pipeline¶
workflows:
- id: etl-pipeline
description: Extract from multiple sources, transform, load
trigger:
uri: cron://etl?schedule=0 */6 * * *
steps:
# Extract from multiple sources in parallel
- multicast:
parallel: true
uris:
- http://crm-api:8080/contacts?since=${header.LastRun}
- postgres://warehouse:5432/sales?operation=query&query=SELECT * FROM recent_sales
- s3://data-lake/reports/daily.json?region=us-east-1
# Aggregate all sources
- aggregate:
correlation: "etl-batch"
strategy: starlang
expression: |
combined = {"contacts": [], "sales": [], "reports": []}
for msg in messages:
data = read(msg, "application/json")
if "email" in data:
combined["contacts"].append(data)
elif "revenue" in data:
combined["sales"].append(data)
else:
combined["reports"].append(data)
write(combined, "application/json")
completion-size: 3
# Transform and load
- transform:
starlang: |
batch = read(payload, "application/json")
summary = {
"total_contacts": sizeOf(batch["contacts"]),
"total_revenue": sum(s["revenue"] for s in batch["sales"]),
"report_count": sizeOf(batch["reports"]),
"generated_at": time.now().format("2006-01-02T15:04:05Z"),
}
write(summary, "application/json")
- target: elasticsearch://analytics:9200?index=daily-summary&operation=index
2.4 XML to JSON with Starlang¶
workflows:
- id: xml-transform
description: Convert SOAP responses to REST JSON
trigger:
uri: http-server://0.0.0.0:8080/api/legacy-bridge
steps:
- target: http://legacy-soap:8080/service
- transform:
starlang: |
data = read(payload, "application/xml")
result = {
"status": data["Envelope"]["Body"]["Response"]["Status"],
"items": [
{"id": item["ID"], "name": item["Name"]}
for item in data["Envelope"]["Body"]["Response"]["Items"]["Item"]
],
}
write(result, "application/json")
3. Agentic AI — Single Agents¶
3.1 ReAct Agent — Customer Support¶
The ReAct agent reasons step-by-step and uses tools to resolve customer queries.
workflows:
- id: customer-support
description: AI customer support agent
trigger:
uri: http-server://0.0.0.0:8080/api/support
steps:
- agent:
id: support-agent
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are a helpful customer support agent for TechCo.
Guidelines:
- Always greet the customer warmly
- Look up their account before answering billing questions
- Check order status using the tracking tool
- Escalate to human agents for refund requests over $500
- Be concise but thorough
max_iterations: 10
memory:
enabled: true
type: sliding-window
window_size: 20
session_id_header: "X-Session-ID"
tools:
- name: lookup_account
type: function
description: "Look up customer account by email or ID"
parameters:
type: object
properties:
email:
type: string
description: Customer email address
customer_id:
type: string
description: Customer ID
required: []
- name: check_order_status
type: function
description: "Check the status of an order"
parameters:
type: object
properties:
order_id:
type: string
required: ["order_id"]
- name: create_ticket
type: function
description: "Create a support ticket for escalation"
parameters:
type: object
properties:
subject:
type: string
priority:
type: string
enum: ["low", "medium", "high", "critical"]
description:
type: string
required: ["subject", "priority", "description"]
3.2 Plan-Execute Agent — Research Analyst¶
Two-phase agent that creates a plan, then executes each step.
workflows:
- id: research-analyst
description: AI research analyst with planning
trigger:
uri: http-server://0.0.0.0:8080/api/research
steps:
- agent:
id: analyst
type: plan-execute
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.3
system_prompt: |
You are a senior research analyst. When given a research topic:
1. Create a detailed research plan with specific steps
2. Execute each step using available tools
3. Synthesize findings into a comprehensive report
Always cite your sources and flag areas of uncertainty.
max_iterations: 20
tools:
- name: search_web
type: function
description: "Search the web for information"
parameters:
type: object
properties:
query:
type: string
required: ["query"]
- name: search_arxiv
type: function
description: "Search academic papers on arXiv"
parameters:
type: object
properties:
query:
type: string
max_results:
type: integer
default: 5
required: ["query"]
- name: analyze_data
type: function
description: "Analyze structured data and produce insights"
parameters:
type: object
properties:
data:
type: string
description: JSON data to analyze
analysis_type:
type: string
enum: ["trend", "comparison", "statistical"]
required: ["data", "analysis_type"]
3.3 Deep Agent — Complex Problem Solver¶
Recursive decomposition for complex multi-faceted problems.
workflows:
- id: architecture-advisor
description: Deep agent for software architecture decisions
trigger:
uri: http-server://0.0.0.0:8080/api/architecture-review
steps:
- agent:
id: architect
type: deep
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.2
system_prompt: |
You are a senior software architect. Analyze architecture decisions by:
1. Breaking complex questions into sub-problems
2. Analyzing each sub-problem independently
3. Considering trade-offs: performance, scalability, maintainability, cost
4. Synthesizing recommendations with clear reasoning
If a sub-problem is still complex, break it further.
max_iterations: 15
deep_max_depth: 3
tools:
- name: check_compatibility
type: function
description: "Check technology compatibility between two systems"
parameters:
type: object
properties:
system_a:
type: string
system_b:
type: string
required: ["system_a", "system_b"]
- name: estimate_load
type: function
description: "Estimate system load for given traffic patterns"
parameters:
type: object
properties:
requests_per_second:
type: integer
payload_size_kb:
type: integer
pattern:
type: string
enum: ["steady", "bursty", "seasonal"]
required: ["requests_per_second"]
3.4 Loop Agent — Code Reviewer¶
Iterative refinement until quality threshold met.
workflows:
- id: code-reviewer
description: Iterative code review agent
trigger:
uri: http-server://0.0.0.0:8080/api/review
steps:
- agent:
id: reviewer
type: loop
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
You are a meticulous code reviewer. Review the provided code for:
1. Security vulnerabilities (OWASP Top 10)
2. Performance issues
3. Code quality and maintainability
4. Test coverage gaps
5. Best practice violations
After each review pass, check if you've covered all areas.
When your review is comprehensive and all issues are documented,
respond with "COMPLETE" followed by the full review report.
If you haven't finished, indicate what areas still need review.
max_iterations: 5
4. Agentic AI — Multi-Agent Orchestration¶
4.1 Supervisor Agent — Customer Service Team¶
A supervisor routes customer queries to specialist agents.
workflows:
- id: customer-service-team
description: Supervisor routing to specialist agents
trigger:
uri: http-server://0.0.0.0:8080/api/help
steps:
- agent:
id: supervisor
type: supervisor
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are a customer service supervisor. Analyze the customer's request
and route it to the most appropriate specialist:
- billing-agent: Payment issues, invoices, refunds, subscription changes
- technical-agent: Product bugs, setup help, API questions, integrations
- sales-agent: Pricing questions, upgrades, feature requests, demos
Choose the best agent based on the customer's primary need.
sub_agents:
- id: billing-agent
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are a billing specialist. Help with payments, invoices,
refunds, and subscription management.
tools:
- name: lookup_invoice
type: function
description: "Look up invoice by ID or customer email"
parameters:
type: object
properties:
invoice_id: { type: string }
email: { type: string }
- name: process_refund
type: function
description: "Process a refund"
parameters:
type: object
properties:
invoice_id: { type: string }
amount: { type: number }
reason: { type: string }
required: ["invoice_id", "amount", "reason"]
- id: technical-agent
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
You are a technical support engineer. Help with product bugs,
setup, API questions, and integration issues.
tools:
- name: search_docs
type: rag-query
description: "Search the product documentation"
rag:
store: pgvector
collection: product-docs
top_k: 5
- name: check_service_status
type: function
description: "Check service health status"
parameters:
type: object
properties:
service: { type: string }
required: ["service"]
- id: sales-agent
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are a sales specialist. Help with pricing, upgrades,
feature requests, and demo scheduling.
tools:
- name: get_pricing
type: function
description: "Get pricing for a plan"
parameters:
type: object
properties:
plan: { type: string, enum: ["starter", "pro", "enterprise"] }
required: ["plan"]
- name: schedule_demo
type: function
description: "Schedule a product demo"
parameters:
type: object
properties:
name: { type: string }
email: { type: string }
preferred_time: { type: string }
required: ["name", "email"]
4.2 Sequential Agent — Content Pipeline¶
Agents refine output in sequence: draft → review → polish.
workflows:
- id: content-pipeline
description: Multi-agent content creation pipeline
trigger:
uri: http-server://0.0.0.0:8080/api/create-content
steps:
- agent:
id: content-pipeline
type: sequential
sub_agents:
- id: researcher
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
You are a research specialist. Given a topic, research it
thoroughly and produce a detailed outline with key facts,
statistics, and talking points. Include source references.
tools:
- name: search_web
type: function
description: "Search the web"
parameters:
type: object
properties:
query: { type: string }
required: ["query"]
- id: writer
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are an expert content writer. Take the research outline
provided and write a compelling, well-structured article.
Use engaging language, clear headings, and concrete examples.
- id: editor
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are a senior editor. Review the article for:
- Grammar and clarity
- Logical flow and structure
- Factual accuracy
- Tone consistency
- SEO optimization
Return the polished final version with your edits applied.
4.3 Parallel Agent — Multi-Perspective Analysis¶
Agents analyze independently, results aggregated.
workflows:
- id: investment-analysis
description: Parallel analysis from multiple expert perspectives
trigger:
uri: http-server://0.0.0.0:8080/api/analyze-investment
steps:
- agent:
id: investment-team
type: parallel
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Synthesize the analyses from all perspectives into a final
investment recommendation with confidence levels.
sub_agents:
- id: financial-analyst
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
You are a financial analyst. Analyze the investment from a
financial perspective: revenue, margins, cash flow, valuation
multiples, and growth trajectory.
- id: market-analyst
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are a market analyst. Analyze the competitive landscape,
market size, market share trends, and industry dynamics.
- id: risk-analyst
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
You are a risk analyst. Identify and assess key risks:
regulatory, competitive, technological, macroeconomic,
and operational. Rate each risk as low/medium/high.
- id: technical-analyst
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
You are a technology analyst. Evaluate the company's
technology stack, IP portfolio, R&D pipeline, and
technical moat vs competitors.
4.4 Nested Supervisor — Enterprise Help Desk¶
Hierarchical routing with sub-supervisors.
workflows:
- id: enterprise-helpdesk
description: Two-tier supervisor with sub-teams
trigger:
uri: http-server://0.0.0.0:8080/api/helpdesk
steps:
- agent:
id: tier1-supervisor
type: supervisor
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are the Tier 1 help desk supervisor. Route requests to:
- it-team: Hardware, software, network, access issues
- hr-team: Benefits, payroll, time-off, policies
- facilities-team: Office space, equipment, maintenance
sub_agents:
- id: it-team
type: supervisor
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
You supervise the IT team. Route to specialists:
- network-specialist: WiFi, VPN, firewall, connectivity
- software-specialist: App installs, licenses, updates, bugs
- security-specialist: Access requests, password resets, MFA
sub_agents:
- id: network-specialist
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: "You are a network specialist. Diagnose and resolve connectivity issues."
- id: software-specialist
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: "You are a software specialist. Help with application issues."
- id: security-specialist
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: "You are a security specialist. Handle access and authentication."
- id: hr-team
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: "You are an HR specialist. Help with benefits, payroll, and policies."
- id: facilities-team
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: "You handle facilities requests: office space, equipment, maintenance."
5. Agentic AI — Agent Flows¶
5.1 Sequential Flow — Document Processing Pipeline¶
workflows:
- id: doc-processing
description: Sequential processing with multiple AI stages
trigger:
uri: http-server://0.0.0.0:8080/api/process-document
steps:
# Stage 1: Extract key information
- agent:
id: extractor
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Extract the following from the document:
- Document type (contract, invoice, report, letter)
- Key dates
- Named entities (people, companies, locations)
- Financial amounts
Return as structured JSON.
# Stage 2: Classify and route
- choice:
when:
- starlang: 'payload["document_type"] == "contract"'
steps:
- agent:
id: contract-analyzer
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Analyze the contract for:
- Key obligations for each party
- Important deadlines and milestones
- Risk clauses (liability, termination, penalties)
- Compliance issues
- starlang: 'payload["document_type"] == "invoice"'
steps:
- agent:
id: invoice-processor
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Validate the invoice:
- Match line items to purchase orders
- Verify tax calculations
- Flag discrepancies
- Prepare for payment approval
5.2 Conditional Flow — Quality Gate¶
workflows:
- id: quality-gate
description: AI-powered quality gate with conditional routing
trigger:
uri: kafka://submissions?brokers=kafka:9092&groupId=qa
steps:
# AI quality assessment
- agent:
id: qa-assessor
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Assess the quality of this submission. Score from 0-100.
Return JSON: {"score": N, "issues": [...], "recommendation": "approve|revise|reject"}
# Route based on score
- choice:
when:
- starlang: 'payload["recommendation"] == "approve"'
steps:
- target: kafka://approved?brokers=kafka:9092
- starlang: 'payload["recommendation"] == "revise"'
steps:
# Send back through revision agent
- agent:
id: revision-agent
type: loop
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Revise this submission based on the QA feedback.
Address each issue identified. When all issues are
resolved, respond with "COMPLETE" followed by the
revised submission.
max_iterations: 3
- target: kafka://approved?brokers=kafka:9092
otherwise:
steps:
- target: kafka://rejected?brokers=kafka:9092
5.3 Parallel Flow — Multi-Model Consensus¶
workflows:
- id: consensus-analysis
description: Get analysis from multiple LLM providers and synthesize
trigger:
uri: http-server://0.0.0.0:8080/api/consensus
steps:
# Run same prompt through different models in parallel
- multicast:
parallel: true
uris:
- agent://openai-analyst
- agent://anthropic-analyst
- agent://groq-analyst
# Synthesize all perspectives
- agent:
id: synthesizer
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You received analyses from three different AI models.
Synthesize them into a consensus report:
- Where do all models agree? (high confidence)
- Where do they disagree? (flag for human review)
- What unique insights does each provide?
6. Agentic AI — Memory Systems¶
6.1 Sliding Window Memory — Chat Bot¶
Best for bounded-context conversations.
workflows:
- id: chatbot
description: Conversational chatbot with sliding window memory
trigger:
uri: http-server://0.0.0.0:8080/api/chat
steps:
- agent:
id: chatbot
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: "You are a friendly assistant."
memory:
enabled: true
type: sliding-window
window_size: 30
session_id_header: "X-Session-ID"
6.2 Summary Memory — Long Conversations¶
Auto-summarizes old messages to maintain context without exceeding token limits.
workflows:
- id: therapy-bot
description: Long-running conversation with summary memory
trigger:
uri: http-server://0.0.0.0:8080/api/session
steps:
- agent:
id: therapist
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are an empathetic conversational partner. Remember key
details from previous conversations and refer to them naturally.
memory:
enabled: true
type: summary
max_messages: 50
session_id_header: "X-Session-ID"
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
6.3 Token Window Memory — Cost-Controlled Agent¶
Strict token budget for LLM calls.
workflows:
- id: cost-controlled-agent
description: Agent with strict token budget
trigger:
uri: http-server://0.0.0.0:8080/api/assist
steps:
- agent:
id: assistant
type: react
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
max_tokens: 1000
system_prompt: "Be concise and helpful."
memory:
enabled: true
type: token-window
max_tokens: 4000
session_id_header: "X-Session-ID"
6.4 SQL Long-Term Memory — Persistent Sessions¶
Sessions persisted across server restarts.
workflows:
- id: persistent-assistant
description: Assistant with PostgreSQL-backed session memory
trigger:
uri: http-server://0.0.0.0:8080/api/assistant
steps:
- agent:
id: persistent-agent
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: "You are a personal assistant that remembers everything."
memory:
enabled: true
type: sql-long-term
dsn: "postgres://user:pass@db:5432/helix?sslmode=disable"
session_id_header: "X-Session-ID"
6.5 Vector Memory — Semantic Recall¶
Find relevant past conversations by meaning, not just recency.
workflows:
- id: knowledge-agent
description: Agent with semantic memory recall
trigger:
uri: http-server://0.0.0.0:8080/api/knowledge
steps:
- agent:
id: knowledge-agent
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
You are a knowledge management agent. You remember all past
interactions and can recall relevant information from any
previous conversation based on semantic similarity.
memory:
enabled: true
type: vector
top_k: 10
embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
session_id_header: "X-Session-ID"
6.6 Composite Memory — Best of All Worlds¶
Combine sliding window (recent) + vector (semantic) + long-term (persistent).
workflows:
- id: advanced-assistant
description: Agent with composite memory system
trigger:
uri: http-server://0.0.0.0:8080/api/advanced
steps:
- agent:
id: super-agent
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are an advanced assistant with perfect memory. You can recall:
- Recent conversation context (last 20 messages)
- Semantically relevant past conversations
- Long-term persistent facts and preferences
memory:
enabled: true
type: composite
session_id_header: "X-Session-ID"
backends:
- type: sliding-window
window_size: 20
- type: vector
top_k: 5
embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- type: sql-long-term
dsn: "postgres://user:pass@db:5432/helix?sslmode=disable"
7. Agentic AI — Tool Integration¶
7.1 RAG Query Tool¶
Give agents access to your knowledge base.
workflows:
- id: docs-assistant
description: Agent with RAG-powered documentation search
trigger:
uri: http-server://0.0.0.0:8080/api/docs-qa
steps:
- agent:
id: docs-agent
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are a documentation assistant. Use the search_docs tool
to find relevant information before answering questions.
Always cite the source document.
tools:
- name: search_docs
type: rag-query
description: "Search the product documentation knowledge base"
rag:
store: pgvector
connection: "postgres://user:pass@db:5432/vectors"
collection: product-docs
embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
top_k: 5
strategy: hybrid
alpha: 0.7
7.2 MCP Tool Integration¶
Connect agents to external MCP servers (Claude Desktop tools, etc.).
workflows:
- id: mcp-agent
description: Agent with access to external MCP tools
trigger:
uri: http-server://0.0.0.0:8080/api/mcp-agent
steps:
- agent:
id: mcp-powered-agent
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You have access to tools from multiple MCP servers:
- GitHub tools for repository management
- Filesystem tools for file operations
- Database tools for SQL queries
tools:
- name: github_tools
type: mcp
mcp_server:
url: http://localhost:3001/mcp
transport: http
- name: fs_tools
type: mcp
mcp_server:
command: npx
args: ["-y", "@modelcontextprotocol/server-filesystem", "/data"]
transport: stdio
- name: db_tools
type: mcp
mcp_server:
url: http://localhost:3002/mcp
transport: sse
7.3 HTTP API Tool¶
Give agents access to external REST APIs.
workflows:
- id: api-agent
description: Agent that calls external APIs
trigger:
uri: http-server://0.0.0.0:8080/api/smart-agent
steps:
- agent:
id: api-caller
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: "You can call external APIs to get real-time data."
tools:
- name: get_weather
type: connector
description: "Get current weather for a city"
http:
method: GET
url: "https://api.weather.com/v1/current"
headers:
X-API-Key: "${env:WEATHER_API_KEY}"
query_params:
city: "{{city}}"
parameters:
type: object
properties:
city: { type: string }
required: ["city"]
- name: create_jira_ticket
type: connector
description: "Create a JIRA issue"
http:
method: POST
url: "https://myorg.atlassian.net/rest/api/3/issue"
headers:
Authorization: "Basic ${env:JIRA_TOKEN}"
Content-Type: "application/json"
parameters:
type: object
properties:
summary: { type: string }
description: { type: string }
priority: { type: string, enum: ["High", "Medium", "Low"] }
required: ["summary", "description"]
8. RAG — Document Ingestion Pipelines¶
8.1 Multi-Format Directory Ingestion¶
Ingest an entire directory of mixed documents.
workflows:
- id: ingest-docs
description: Ingest all documents from a directory
trigger:
uri: cron://ingest?schedule=0 2 * * *
steps:
- rag-ingest:
reader: directory
path: /data/documents
recursive: true
# Auto-detects: PDF, DOCX, MD, HTML, CSV, JSON, XML, TXT
- rag-chunk:
strategy: recursive
chunk_size: 1000
overlap: 200
separators: ["\n\n", "\n", " "]
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
batch_size: 100
- rag-store:
store: pgvector
connection: "postgres://user:pass@db:5432/vectors"
collection: company-docs
8.2 PDF with OCR Ingestion¶
Process scanned PDFs using OCR before chunking.
workflows:
- id: ingest-scanned-pdfs
description: OCR + ingest for scanned documents
trigger:
uri: file:///data/scanned?include=*.pdf&move=../processed
steps:
- rag-ingest:
reader: ocr
backend: docling
endpoint: http://docling:8080
output_format: markdown
- rag-chunk:
strategy: markdown
chunk_size: 800
retain_headers: true
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: qdrant
url: http://qdrant:6333
collection: scanned-docs
8.3 Large File Streaming OCR¶
Handle multi-GB files with streaming and parallel page processing.
workflows:
- id: ingest-large-pdfs
description: Stream large PDFs with page-level parallelism
trigger:
uri: s3://documents/large?region=us-east-1
steps:
- rag-ingest:
reader: ocr-streaming
backend: docling
endpoint: http://docling:8080
max_memory_bytes: 52428800 # 50MB threshold
max_file_size: 2147483648 # 2GB max
page_split: true
concurrency: 4
output_format: markdown
- rag-chunk:
strategy: semantic
threshold: 0.5
embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "postgres://user:pass@db:5432/vectors"
collection: large-docs
8.4 Code Repository Ingestion¶
Index source code with language-aware chunking.
workflows:
- id: ingest-codebase
description: Index source code for code search
trigger:
uri: cron://code-sync?schedule=0 */4 * * *
steps:
- rag-ingest:
reader: directory
path: /repos/my-project
recursive: true
include: ["*.go", "*.py", "*.js", "*.ts", "*.java"]
exclude: ["*_test.go", "*.test.js", "vendor/**", "node_modules/**"]
- rag-chunk:
strategy: code
languages: ["go", "python", "javascript", "java"]
max_chunk_size: 2000
- rag-extract:
fields: ["title", "summary", "keywords"]
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
- rag-embed:
provider: openai
model: text-embedding-3-large
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: elasticsearch
url: http://es:9200
index: code-search
8.5 Web Content Ingestion¶
Ingest HTML pages with table extraction.
workflows:
- id: ingest-web-pages
description: Crawl and ingest web documentation
trigger:
uri: http-server://0.0.0.0:8080/api/ingest-url
steps:
# Fetch the page
- target: http://${header.TargetUrl}
- rag-ingest:
reader: html
extract_tables: true
strip_scripts: true
strip_styles: true
- rag-chunk:
strategy: html
chunk_size: 500
retain_structure: true
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "postgres://user:pass@db:5432/vectors"
collection: web-docs
9. RAG — Chunking Strategies¶
9.1 Semantic Chunking for Technical Docs¶
- rag-chunk:
strategy: semantic
threshold: 0.5 # Lower = more splits, higher = fewer
max_chunk_size: 1500
embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
9.2 Hierarchical Chunking for Auto-Merging Retrieval¶
- rag-chunk:
strategy: hierarchical
levels:
- chunk_size: 2000 # Parent (Level 0)
- chunk_size: 500 # Child (Level 1)
- chunk_size: 128 # Leaf (Level 2)
9.3 Sliding Window for Dense Retrieval¶
9.4 Sentence-Based for Conversational Data¶
9.5 Token-Aware Chunking¶
- rag-chunk:
strategy: token
chunk_size: 512 # Tokens, not characters
overlap: 64
tokenizer: cl100k_base # GPT-4 tokenizer
10. RAG — Embedding & Storage¶
10.1 OpenAI Embeddings + pgvector¶
- rag-embed:
provider: openai
model: text-embedding-3-small # 1536 dims, cheapest
api_key: "${env:OPENAI_API_KEY}"
batch_size: 100
- rag-store:
store: pgvector
connection: "postgres://user:pass@db:5432/vectors?sslmode=require"
collection: my-docs
dimensions: 1536
10.2 Local Ollama Embeddings + Milvus¶
- rag-embed:
provider: ollama
model: nomic-embed-text
base_url: http://ollama:11434
batch_size: 32
- rag-store:
store: milvus
url: http://milvus:19530
collection: local-docs
dimensions: 768
10.3 Multimodal CLIP Embeddings¶
- rag-embed:
provider: clip
model: clip-vit-large-patch14
base_url: http://clip-server:8080
- rag-store:
store: qdrant
url: http://qdrant:6333
collection: multimodal-docs
dimensions: 768
10.4 Jina Embeddings + Redis Vector¶
- rag-embed:
provider: jina
model: jina-embeddings-v3
api_key: "${env:JINA_API_KEY}"
- rag-store:
store: redis
url: redis://redis:6379
collection: jina-docs
dimensions: 1024
11. RAG — Retrieval Strategies¶
11.1 Simple Vector Search¶
11.2 Hybrid Search (Vector + BM25)¶
Best for most production use cases — combines semantic understanding with keyword precision.
11.3 MMR (Maximal Marginal Relevance)¶
Diverse results — avoids returning near-duplicate chunks.
11.4 HyDE — Hypothetical Document Embeddings¶
Generates a hypothetical answer, then searches using its embedding.
- rag-retrieve:
strategy: hyde
top_k: 10
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
num_hypothetical: 3 # Generate 3 hypothetical docs
11.5 MultiQuery — Query Reformulation¶
Generates multiple phrasings of the query for broader recall.
- rag-retrieve:
strategy: multi_query
top_k: 10
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
num_queries: 4 # Generate 4 query variants
11.6 StepBack Prompting¶
Generates an abstract version of the query for broader concept matching.
- rag-retrieve:
strategy: step_back
top_k: 10
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
11.7 SubQuestion Decomposition¶
Breaks complex queries into sub-questions, retrieves for each.
- rag-retrieve:
strategy: sub_question
top_k: 5
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
max_sub_questions: 4
11.8 Auto-Merging Retrieval¶
Retrieves leaf chunks, auto-merges into parent when threshold met.
# Requires hierarchical chunking (see 9.2)
- rag-retrieve:
strategy: auto_merging
top_k: 10
merge_threshold: 0.6 # Merge parent if 60%+ children retrieved
11.9 Fusion Retrieval (Reciprocal Rank Fusion)¶
Combines multiple retriever results using RRF scoring.
- rag-retrieve:
strategy: fusion
top_k: 10
sub_retrievers:
- strategy: simple
weight: 1.0
- strategy: bm25
weight: 0.8
rrf_k: 60 # RRF constant
11.10 Router Retrieval¶
LLM decides which retriever to use based on query type.
- rag-retrieve:
strategy: router
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
sub_retrievers:
- name: code-search
description: "For code-related queries"
strategy: simple
collection: code-index
- name: docs-search
description: "For documentation queries"
strategy: hybrid
collection: docs-index
- name: api-search
description: "For API reference queries"
strategy: bm25
collection: api-index
12. RAG — Postprocessing & Reranking¶
12.1 Cohere Reranking¶
Cross-encoder reranking for much better relevance.
- rag-postprocess:
steps:
- type: cohere-rerank
params:
model: rerank-english-v3.0
api_key: "${env:COHERE_API_KEY}"
top_n: 5
12.2 Full Postprocessing Pipeline¶
- rag-postprocess:
steps:
# 1. Remove low-quality results
- type: similarity-filter
params:
min_score: 0.3
# 2. Remove duplicates
- type: dedup
# 3. Cross-encoder reranking
- type: cohere-rerank
params:
model: rerank-english-v3.0
api_key: "${env:COHERE_API_KEY}"
top_n: 10
# 4. Diversity filtering
- type: diversity-filter
params:
threshold: 0.85
# 5. Reorder for LLM attention patterns
- type: long-context-reorder
# 6. Final top-K
- type: top-k
params:
k: 5
12.3 Metadata Filtering¶
- rag-postprocess:
steps:
- type: metadata-filter
params:
conditions:
- key: department
operator: eq
value: engineering
- key: created_at
operator: gt
value: "2025-01-01"
12.4 PII Masking¶
- rag-postprocess:
steps:
- type: pii-mask
params:
patterns: ["email", "ssn", "phone", "credit_card"]
action: redact # or "hash"
12.5 Keyword Boosting¶
- rag-postprocess:
steps:
- type: keyword-boost
params:
keywords: ["security", "compliance", "audit"]
boost_factor: 1.5
13. RAG — Synthesis & Generation¶
13.1 Compact Synthesis (Default)¶
All context in single prompt — cheapest, fastest.
- rag-synthesize:
mode: compact
max_tokens: 2000
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Answer the question using ONLY the provided context.
If the context doesn't contain the answer, say so.
Cite sources using [Source: document_name].
13.2 Refine Synthesis¶
Iteratively refines answer with each chunk — best quality for complex questions.
- rag-synthesize:
mode: refine
max_chunks: 10
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Refine your previous answer using this additional context.
Maintain accuracy and add new relevant information.
13.3 Tree Summarize¶
Hierarchical summarization — good for many documents.
- rag-synthesize:
mode: tree_summarize
chunks_per_level: 3
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
13.4 Accumulate Synthesis (Parallel)¶
Generate per-chunk answers in parallel, then combine.
- rag-synthesize:
mode: accumulate
parallel: true
model:
provider: groq
model: llama-3.3-70b-versatile
api_key: "${env:GROQ_API_KEY}"
14. RAG — Advanced Pipelines¶
14.1 Full Production RAG Pipeline¶
Complete pipeline with all stages optimized for production.
workflows:
# Ingestion pipeline (run nightly)
- id: rag-ingest-pipeline
description: Full document ingestion with metadata extraction
trigger:
uri: cron://ingest?schedule=0 1 * * *
steps:
- rag-ingest:
reader: directory
path: /data/documents
recursive: true
- rag-extract:
fields: ["title", "summary", "keywords", "entities", "language"]
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
- rag-chunk:
strategy: recursive
chunk_size: 800
overlap: 150
separators: ["\n\n", "\n", ". ", " "]
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
batch_size: 100
- rag-cache:
enabled: true
backend: redis
url: redis://redis:6379
max_size: 50000
- rag-store:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: production-docs
# Query pipeline (real-time)
- id: rag-query-pipeline
description: Optimized query pipeline with reranking
trigger:
uri: http-server://0.0.0.0:8080/api/query
steps:
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-retrieve:
strategy: hybrid
top_k: 20
alpha: 0.7
- rag-postprocess:
steps:
- type: similarity-filter
params:
min_score: 0.25
- type: cohere-rerank
params:
model: rerank-english-v3.0
api_key: "${env:COHERE_API_KEY}"
top_n: 5
- type: long-context-reorder
- rag-synthesize:
mode: compact
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Answer the question using the provided context.
Be accurate and concise. Cite sources with [Source: filename].
If unsure, indicate your confidence level.
14.2 Multi-Collection RAG¶
Query across multiple knowledge bases.
workflows:
- id: multi-collection-rag
description: Search across engineering + legal + product docs
trigger:
uri: http-server://0.0.0.0:8080/api/unified-search
steps:
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
# Search all collections in parallel
- multicast:
parallel: true
uris:
- rag-retrieve://engineering-docs?strategy=simple&top_k=5
- rag-retrieve://legal-docs?strategy=simple&top_k=5
- rag-retrieve://product-docs?strategy=simple&top_k=5
# Merge and rerank
- rag-postprocess:
steps:
- type: cohere-rerank
params:
top_n: 8
api_key: "${env:COHERE_API_KEY}"
- rag-synthesize:
mode: compact
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
14.3 Agentic RAG — Agent-Driven Retrieval¶
Agent decides when and what to retrieve.
workflows:
- id: agentic-rag
description: Agent with RAG tool for intelligent retrieval
trigger:
uri: http-server://0.0.0.0:8080/api/smart-qa
steps:
- agent:
id: rag-agent
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are an expert assistant with access to a knowledge base.
Strategy:
1. Analyze the user's question to identify what information you need
2. Search the knowledge base with targeted queries
3. If results are insufficient, reformulate and search again
4. Synthesize a comprehensive answer with citations
You can make multiple searches with different queries.
max_iterations: 10
tools:
- name: search_knowledge_base
type: rag-query
description: "Search the knowledge base. Use specific, targeted queries."
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: all-docs
embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
top_k: 5
strategy: hybrid
postprocess:
- type: cohere-rerank
params:
top_n: 3
api_key: "${env:COHERE_API_KEY}"
- name: search_code
type: rag-query
description: "Search source code examples and API references"
rag:
store: elasticsearch
url: http://es:9200
collection: code-index
top_k: 5
strategy: bm25
14.4 Corrective RAG — Self-Correcting Pipeline¶
Evaluates retrieval quality and retries with different strategies.
workflows:
- id: corrective-rag
description: Self-correcting RAG with quality evaluation
trigger:
uri: http-server://0.0.0.0:8080/api/accurate-qa
steps:
# First attempt: simple retrieval
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-retrieve:
strategy: simple
top_k: 10
# Evaluate retrieval quality
- rag-eval:
metrics: ["context_relevancy"]
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
threshold: 0.6
# If quality is low, retry with HyDE
- choice:
when:
- starlang: 'headers["RagEvalScore"] < 0.6'
steps:
- rag-retrieve:
strategy: hyde
top_k: 10
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
- rag-postprocess:
steps:
- type: cohere-rerank
params:
top_n: 5
api_key: "${env:COHERE_API_KEY}"
- rag-synthesize:
mode: refine
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
15. RAG — Knowledge Graphs¶
15.1 Build Knowledge Graph from Documents¶
workflows:
- id: build-knowledge-graph
description: Extract entities and relationships into a knowledge graph
trigger:
uri: cron://kg-build?schedule=0 3 * * *
steps:
- rag-ingest:
reader: directory
path: /data/documents
recursive: true
- rag-chunk:
strategy: sentence
max_chunk_size: 500
- rag-graph-build:
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
entity_types: ["Person", "Organization", "Technology", "Product", "Event"]
relationship_types: ["works_for", "uses", "competes_with", "acquired", "founded"]
store:
backend: neo4j
url: bolt://neo4j:7687
username: neo4j
password: "${env:NEO4J_PASSWORD}"
15.2 Graph-Enhanced RAG Retrieval¶
workflows:
- id: graph-rag-query
description: Combine vector search with knowledge graph traversal
trigger:
uri: http-server://0.0.0.0:8080/api/graph-qa
steps:
# Vector retrieval
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-retrieve:
strategy: simple
top_k: 5
# Knowledge graph query for related entities
- rag-graph-query:
store:
backend: neo4j
url: bolt://neo4j:7687
max_hops: 2
max_results: 10
# Synthesize with both vector and graph context
- rag-synthesize:
mode: compact
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Answer using both the document excerpts and knowledge graph relationships.
The knowledge graph provides entity connections and relationships.
Use both sources for a comprehensive answer.
16. RAG — Evaluation & Quality¶
16.1 RAG Quality Evaluation Pipeline¶
workflows:
- id: rag-eval-pipeline
description: Evaluate RAG pipeline quality
trigger:
uri: http-server://0.0.0.0:8080/api/eval
steps:
# Run the RAG pipeline
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-retrieve:
strategy: hybrid
top_k: 10
- rag-synthesize:
mode: compact
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
# Evaluate quality
- rag-eval:
metrics:
- faithfulness # Is the answer grounded in context?
- answer_relevancy # Is the answer relevant to the query?
- context_relevancy # Is the retrieved context relevant?
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
# Log metrics
- transform:
starlang: |
result = read(payload, "application/json")
result["metrics"] = {
"faithfulness": float(headers["RagEvalFaithfulness"]),
"answer_relevancy": float(headers["RagEvalAnswerRelevancy"]),
"context_relevancy": float(headers["RagEvalContextRelevancy"]),
}
write(result, "application/json")
- target: elasticsearch://metrics:9200?index=rag-quality&operation=index
16.2 Retrieval Metrics (Offline Evaluation)¶
workflows:
- id: retrieval-eval
description: Evaluate retrieval quality against ground truth
trigger:
uri: file:///data/eval/test-queries.json
steps:
- split:
starlang: 'payload'
steps:
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-retrieve:
strategy: hybrid
top_k: 10
- rag-eval:
metrics:
- hit_rate # Did we retrieve any relevant doc?
- mrr # Mean Reciprocal Rank
- ndcg # Normalized Discounted Cumulative Gain
ground_truth_key: "expected_doc_ids"
- aggregate:
correlation: "eval-batch"
strategy: starlang
expression: |
metrics = [read(m, "application/json") for m in messages]
avg_hit_rate = sum(m["hit_rate"] for m in metrics) / sizeOf(metrics)
avg_mrr = sum(m["mrr"] for m in metrics) / sizeOf(metrics)
avg_ndcg = sum(m["ndcg"] for m in metrics) / sizeOf(metrics)
write({"avg_hit_rate": avg_hit_rate, "avg_mrr": avg_mrr, "avg_ndcg": avg_ndcg}, "application/json")
17. Multimodal AI¶
17.1 Vision Agent — Image Analysis¶
workflows:
- id: image-analyzer
description: Analyze images using vision models
trigger:
uri: http-server://0.0.0.0:8080/api/analyze-image
steps:
- agent:
id: vision-agent
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
You are a visual analysis expert. Analyze images for:
- Object identification
- Text extraction (OCR)
- Scene description
- Quality assessment
17.2 Audio Transcription + Analysis¶
workflows:
- id: audio-pipeline
description: Transcribe audio and extract insights
trigger:
uri: s3://recordings/incoming?region=us-east-1
steps:
# Transcribe audio
- rag-ingest:
reader: audio
model:
provider: openai
model: whisper-1
api_key: "${env:OPENAI_API_KEY}"
# Analyze transcript
- agent:
id: transcript-analyzer
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Analyze this meeting transcript. Extract:
- Key decisions made
- Action items with owners
- Follow-up questions
- Sentiment analysis
Return structured JSON.
- target: postgres://db:5432/meetings?operation=insert
17.3 Multimodal RAG — Images + Text¶
workflows:
- id: multimodal-rag-ingest
description: Ingest images and text together
trigger:
uri: cron://mm-ingest?schedule=0 4 * * *
steps:
- rag-ingest:
reader: directory
path: /data/mixed-content
recursive: true
- rag-chunk:
strategy: recursive
chunk_size: 500
- rag-embed:
provider: clip
model: clip-vit-large-patch14
base_url: http://clip-server:8080
- rag-store:
store: qdrant
url: http://qdrant:6333
collection: multimodal-docs
18. Protocol Interop — MCP, A2A, AG-UI, A2UI¶
18.1 MCP Server — Expose Routes as Tools¶
Let Claude Desktop and other MCP clients call your workflows.
helix:
mcp:
server:
enabled: true
transport: http
addr: "0.0.0.0:3000"
name: "my-helix-tools"
version: "1.0.0"
workflows:
- id: process-order
description: "Process a customer order with validation and fulfillment"
trigger:
uri: direct://process-order
steps:
- transform:
starlang: |
order = read(payload, "application/json")
order["status"] = "processing"
order["processed_at"] = time.now().format("2006-01-02T15:04:05Z")
write(order, "application/json")
- target: http://fulfillment:8080/orders
- id: search-knowledge
description: "Search the internal knowledge base for information"
trigger:
uri: direct://search-knowledge
steps:
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-retrieve:
strategy: hybrid
top_k: 5
- rag-synthesize:
mode: compact
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
18.2 MCP Client — Call External MCP Servers¶
workflows:
- id: mcp-bridge
description: Call external MCP tools from workflows
trigger:
uri: http-server://0.0.0.0:8080/api/github
steps:
- target: mcp://localhost:3001/create_issue?transport=http
18.3 A2A Server — Agent-to-Agent Protocol¶
Expose agents for cross-platform discovery and invocation.
helix:
a2a:
servers:
- enabled: true
addr: "0.0.0.0:4000"
name: "my-agent-server"
description: "Enterprise AI agents"
version: "1.0.0"
workflows:
- id: analyst-agent
description: "Financial analysis agent"
trigger:
uri: direct://analyst
steps:
- agent:
id: financial-analyst
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: "You are a financial analyst..."
18.4 AG-UI Server — Streaming Chat Frontend¶
Expose agents to CopilotKit and other AG-UI compatible frontends.
helix:
agui:
servers:
- enabled: true
addr: "0.0.0.0:5000"
path: /api/copilot
name: "my-copilot"
workflows:
- id: copilot-agent
trigger:
uri: direct://copilot
steps:
- agent:
id: copilot
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: "You are an embedded copilot assistant."
memory:
enabled: true
type: sliding-window
window_size: 20
18.5 A2UI Server — Rich Declarative UI¶
Expose agents with declarative UI surfaces (Google A2UI protocol).
helix:
a2ui:
servers:
- enabled: true
addr: "0.0.0.0:6000"
path: /api/ui
name: "my-ui-agent"
workflows:
- id: dashboard-agent
trigger:
uri: direct://dashboard
steps:
- agent:
id: dashboard
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
You generate dashboard UI surfaces. Return A2UI components
like cards, charts, tables, and forms.
19. Security Patterns¶
19.1 JWT Authentication¶
workflows:
- id: secure-api
trigger:
uri: http-server://0.0.0.0:8080/api/secure
auth:
type: jwt
jwt:
secret: "${env:JWT_SECRET}"
algorithm: HS256
claims:
iss: "my-auth-server"
aud: "my-api"
steps:
- target: direct://protected-handler
19.2 OAuth2 Token Introspection¶
workflows:
- id: oauth2-api
trigger:
uri: http-server://0.0.0.0:8080/api/oauth
auth:
type: oauth2
oauth2:
introspection_url: "https://auth.example.com/oauth/introspect"
client_id: "${env:OAUTH_CLIENT_ID}"
client_secret: "${env:OAUTH_CLIENT_SECRET}"
required_scopes: ["read", "write"]
steps:
- target: direct://oauth-handler
19.3 API Key with Rate Limiting¶
workflows:
- id: rate-limited-api
trigger:
uri: http-server://0.0.0.0:8080/api/public
auth:
type: apikey
apikey:
header: X-API-Key
keys:
- "${env:API_KEY_1}"
- "${env:API_KEY_2}"
steps:
- throttle:
rate: 60
period: 1m
- target: direct://api-handler
19.4 Mutual TLS¶
helix:
server:
tls:
enabled: true
cert_file: /certs/server.crt
key_file: /certs/server.key
ca_file: /certs/ca.crt
client_auth: require # Mutual TLS
20. Production & Enterprise¶
20.1 Multi-Tenant Configuration¶
helix:
tenant:
enabled: true
header: X-Tenant-ID
required: true
allowlist:
- tenant-a
- tenant-b
- tenant-c
20.2 Audit Logging¶
20.3 Graceful Shutdown¶
20.4 Clustering with Leader Election¶
helix:
cluster:
enabled: true
backend: redis
redis_url: "redis://redis:6379"
node_id: "${env:HOSTNAME}"
heartbeat_interval: 5s
election_timeout: 15s
20.5 Observability Stack¶
helix:
metrics:
enabled: true
prometheus:
enabled: true
addr: "0.0.0.0:9090"
path: /metrics
tracing:
enabled: true
exporter: otlp
endpoint: "otel-collector:4317"
sample_rate: 0.1
health:
enabled: true
addr: "0.0.0.0:8081"
path: /health
liveness: /healthz
readiness: /readyz
logging:
level: info
format: json
components:
pkg/component/kafka: warn
pkg/engine: debug
20.6 SLA Monitoring¶
helix:
metrics:
sla:
workflows:
order-processor:
max_latency: 500ms
max_error_rate: 0.01
payment-gateway:
max_latency: 200ms
max_error_rate: 0.001
21. End-to-End Application Patterns¶
21.1 AI-Powered Customer Support Platform¶
Complete customer support system with agent routing, RAG knowledge base, ticket management, and analytics.
helix:
mcp:
server:
enabled: true
transport: http
addr: "0.0.0.0:3000"
name: "support-platform"
agui:
servers:
- enabled: true
addr: "0.0.0.0:5000"
path: /api/chat
shutdown:
timeout: 30s
metrics:
enabled: true
prometheus:
enabled: true
addr: "0.0.0.0:9090"
workflows:
# 1. Knowledge base ingestion (nightly)
- id: kb-ingest
description: Ingest support documentation
trigger:
uri: cron://kb-sync?schedule=0 2 * * *
steps:
- rag-ingest:
reader: directory
path: /data/support-docs
recursive: true
- rag-chunk:
strategy: recursive
chunk_size: 800
overlap: 150
- rag-extract:
fields: ["title", "keywords"]
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: support-kb
# 2. Customer support agent (real-time)
- id: support-agent
description: AI customer support with supervisor routing
trigger:
uri: http-server://0.0.0.0:8080/api/support
steps:
- agent:
id: support-supervisor
type: supervisor
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Route customer queries to the best specialist agent.
memory:
enabled: true
type: composite
session_id_header: "X-Session-ID"
backends:
- type: sliding-window
window_size: 20
- type: sql-long-term
dsn: "${env:DATABASE_URL}"
sub_agents:
- id: general-support
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: "Handle general support queries using the knowledge base."
tools:
- name: search_kb
type: rag-query
description: "Search support knowledge base"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: support-kb
top_k: 5
strategy: hybrid
- name: create_ticket
type: connector
description: "Escalate to human support"
http:
method: POST
url: "http://ticketing:8080/api/tickets"
- id: billing-support
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: "Handle billing inquiries."
tools:
- name: lookup_account
type: function
description: "Look up billing account"
parameters:
type: object
properties:
email: { type: string }
- id: technical-support
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: "Handle technical issues."
tools:
- name: search_kb
type: rag-query
description: "Search technical docs"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: support-kb
top_k: 5
# 3. Feedback analytics (batch)
- id: feedback-analysis
description: Analyze support interactions for trends
trigger:
uri: cron://analyze?schedule=0 6 * * 1
steps:
- target: postgres://db:5432/interactions?operation=query&query=SELECT * FROM support_logs WHERE created_at > now() - interval '7 days'
- agent:
id: analytics-agent
type: plan-execute
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Analyze this week's support interactions. Produce:
1. Top 10 issue categories by volume
2. Average resolution time trends
3. Customer satisfaction patterns
4. Recommendations for knowledge base updates
5. Staffing recommendations
- target: slack://analytics?channel=support-metrics&webhook=${env:SLACK_WEBHOOK}
21.2 Intelligent Document Processing Platform¶
Automated document intake, classification, extraction, and routing.
workflows:
# Document intake
- id: doc-intake
description: Receive and classify incoming documents
trigger:
uri: http-server://0.0.0.0:8080/api/documents
steps:
# OCR if needed
- choice:
when:
- starlang: '"image" in headers["Content-Type"]'
steps:
- rag-ingest:
reader: ocr
backend: docling
endpoint: http://docling:8080
- starlang: '"pdf" in headers["Content-Type"]'
steps:
- rag-ingest:
reader: ocr-streaming
backend: docling
endpoint: http://docling:8080
page_split: true
concurrency: 4
otherwise:
steps:
- rag-ingest:
reader: auto
# AI classification
- agent:
id: classifier
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Classify this document into one of:
- invoice
- contract
- report
- correspondence
- form
Return JSON: {"type": "...", "confidence": 0.95, "summary": "..."}
# Route by classification
- choice:
when:
- starlang: 'payload["type"] == "invoice"'
steps:
- target: seda://invoice-pipeline?bufferSize=1000
- starlang: 'payload["type"] == "contract"'
steps:
- target: seda://contract-pipeline?bufferSize=500
otherwise:
steps:
- target: seda://general-pipeline?bufferSize=2000
# Invoice processing
- id: invoice-pipeline
trigger:
uri: seda://invoice-pipeline?concurrentConsumers=4
steps:
- agent:
id: invoice-extractor
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Extract structured data from this invoice:
- vendor_name, vendor_address
- invoice_number, invoice_date, due_date
- line_items: [{description, quantity, unit_price, total}]
- subtotal, tax, total
- payment_terms
Return as JSON.
- target: postgres://db:5432/invoices?operation=insert
# Contract analysis
- id: contract-pipeline
trigger:
uri: seda://contract-pipeline?concurrentConsumers=2
steps:
- agent:
id: contract-analyzer
type: plan-execute
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Analyze this contract systematically:
1. Identify all parties and their obligations
2. Extract key dates, milestones, and deadlines
3. Identify risk clauses (liability, termination, penalties)
4. Flag compliance requirements
5. Summarize key commercial terms
max_iterations: 15
- target: postgres://db:5432/contracts?operation=insert
- wiretap: slack://legal?channel=contract-review&webhook=${env:SLACK_WEBHOOK}
21.3 Real-Time Data Pipeline with AI Enrichment¶
Stream processing with AI-powered enrichment and anomaly detection.
workflows:
# Ingest events from Kafka
- id: event-enrichment
description: Enrich events with AI-generated insights
trigger:
uri: kafka://raw-events?brokers=kafka:9092&groupId=enrichment
steps:
# Batch events for efficiency
- aggregate:
correlation: 'headers["EventType"]'
completion-size: 50
completion-timeout: 5s
# AI enrichment
- agent:
id: enrichment-agent
type: react
model:
provider: groq
model: llama-3.3-70b-versatile
api_key: "${env:GROQ_API_KEY}"
system_prompt: |
Analyze these events and add enrichment:
- Sentiment score (for user-facing events)
- Category classification
- Anomaly flag (true/false)
- Severity level (info/warning/critical)
Return enriched events as JSON array.
# Route anomalies
- split:
starlang: 'payload'
steps:
- choice:
when:
- starlang: 'payload["anomaly"] == true'
steps:
- target: kafka://anomalies?brokers=kafka:9092
- target: slack://alerts?channel=anomaly-alerts&webhook=${env:SLACK_WEBHOOK}
otherwise:
steps:
- target: kafka://enriched-events?brokers=kafka:9092
- target: elasticsearch://analytics:9200?index=events&operation=index
21.4 Multi-Agent Code Generation Platform¶
AI agents collaborate to generate, review, and test code.
workflows:
- id: code-gen
description: Multi-agent code generation with review
trigger:
uri: http-server://0.0.0.0:8080/api/generate-code
steps:
- agent:
id: code-team
type: sequential
sub_agents:
# Step 1: Requirements analysis
- id: requirements-analyst
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Analyze the code request. Produce:
- Functional requirements
- Technical constraints
- Edge cases to handle
- Suggested architecture
# Step 2: Code generation
- id: code-generator
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Based on the requirements analysis, generate production-quality code.
Include proper error handling, documentation, and follow best practices.
# Step 3: Code review
- id: code-reviewer
type: loop
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Review the generated code for:
- Security vulnerabilities
- Performance issues
- Code quality
- Test coverage
If issues found, describe them and suggest fixes.
When the code passes review, respond with COMPLETE and the approved code.
max_iterations: 3
# Step 4: Test generation
- id: test-generator
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Generate comprehensive tests for the reviewed code:
- Unit tests for each function
- Edge case tests
- Integration tests
- Error path tests
21.5 Healthcare Clinical Decision Support¶
AI-assisted clinical decision support with patient context, medical RAG, drug interaction checking, and HIPAA-compliant audit trails.
helix:
audit:
enabled: true
max_entries: 100000
log_body: true
max_body_preview: 2048
tenant:
enabled: true
header: X-Hospital-ID
required: true
allowlist: ["hospital-central", "hospital-east", "clinic-north"]
shutdown:
timeout: 60s
drain_timeout: 30s
metrics:
enabled: true
prometheus:
enabled: true
addr: "0.0.0.0:9090"
workflows:
# 1. Medical knowledge base ingestion
- id: medical-kb-ingest
description: Ingest medical literature, drug databases, clinical guidelines
trigger:
uri: cron://medical-sync?schedule=0 0 * * 0
steps:
# Ingest clinical guidelines (PDFs)
- rag-ingest:
reader: directory
path: /data/medical/guidelines
recursive: true
- rag-chunk:
strategy: semantic
threshold: 0.45
max_chunk_size: 1200
embed:
provider: openai
model: text-embedding-3-large
api_key: "${env:OPENAI_API_KEY}"
- rag-extract:
fields: ["title", "keywords", "entities"]
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
- rag-embed:
provider: openai
model: text-embedding-3-large
api_key: "${env:OPENAI_API_KEY}"
batch_size: 50
- rag-store:
store: pgvector
connection: "${env:MEDICAL_VECTOR_DB}"
collection: clinical-guidelines
# 2. Drug interaction database
- id: drug-db-ingest
description: Ingest drug interaction and formulary data
trigger:
uri: cron://drug-sync?schedule=0 1 * * *
steps:
- target: http://drug-api:8080/api/v2/interactions?format=json
- split:
starlang: 'payload'
steps:
- rag-chunk:
strategy: fixed
chunk_size: 500
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:MEDICAL_VECTOR_DB}"
collection: drug-interactions
# 3. Clinical decision support agent
- id: clinical-decision-support
description: AI-assisted clinical decision support
trigger:
uri: http-server://0.0.0.0:8080/api/clinical/consult
auth:
type: jwt
jwt:
secret: "${env:JWT_SECRET}"
claims:
role: ["physician", "nurse_practitioner", "pharmacist"]
steps:
# Audit every clinical query
- wiretap: kafka://clinical-audit?brokers=kafka:9092
- agent:
id: clinical-advisor
type: plan-execute
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.1
system_prompt: |
You are a clinical decision support system. You assist healthcare
providers by searching medical literature and drug databases.
IMPORTANT DISCLAIMERS:
- You provide information ONLY, not diagnoses or treatment decisions
- All suggestions must be reviewed by the treating physician
- Flag any contraindications or critical drug interactions prominently
- Cite clinical guidelines and evidence levels (Level I-V)
- Always note when evidence is limited or conflicting
For each query:
1. Search clinical guidelines for relevant evidence
2. Check drug interactions if medications are mentioned
3. Identify contraindications based on patient context
4. Summarize evidence with strength ratings
max_iterations: 12
memory:
enabled: true
type: sliding-window
window_size: 10
session_id_header: "X-Session-ID"
tools:
- name: search_guidelines
type: rag-query
description: "Search clinical guidelines and medical literature"
rag:
store: pgvector
connection: "${env:MEDICAL_VECTOR_DB}"
collection: clinical-guidelines
top_k: 8
strategy: hybrid
alpha: 0.6
postprocess:
- type: cohere-rerank
params:
model: rerank-english-v3.0
api_key: "${env:COHERE_API_KEY}"
top_n: 5
- name: check_drug_interactions
type: rag-query
description: "Check for drug interactions between medications"
rag:
store: pgvector
connection: "${env:MEDICAL_VECTOR_DB}"
collection: drug-interactions
top_k: 10
strategy: hybrid
- name: get_patient_context
type: connector
description: "Retrieve patient demographics, conditions, and current medications from EHR"
http:
method: GET
url: "http://ehr-api:8080/api/patients/${patient_id}/summary"
headers:
Authorization: "Bearer ${env:EHR_API_TOKEN}"
parameters:
type: object
properties:
patient_id: { type: string }
required: ["patient_id"]
- name: lookup_lab_results
type: connector
description: "Retrieve recent lab results for a patient"
http:
method: GET
url: "http://ehr-api:8080/api/patients/${patient_id}/labs?days=30"
headers:
Authorization: "Bearer ${env:EHR_API_TOKEN}"
parameters:
type: object
properties:
patient_id: { type: string }
required: ["patient_id"]
# 4. Clinical alert monitoring
- id: clinical-alerts
description: Monitor for critical drug interaction alerts
trigger:
uri: kafka://medication-orders?brokers=kafka:9092&groupId=interaction-checker
steps:
- agent:
id: interaction-checker
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.0
system_prompt: |
Check this medication order for:
1. Drug-drug interactions (severity: minor/moderate/major/contraindicated)
2. Drug-allergy conflicts
3. Dose range validation
4. Renal/hepatic dose adjustments needed
Return JSON: {"safe": bool, "alerts": [...], "severity": "..."}
tools:
- name: check_interactions
type: rag-query
description: "Check drug interaction database"
rag:
store: pgvector
connection: "${env:MEDICAL_VECTOR_DB}"
collection: drug-interactions
top_k: 10
- choice:
when:
- starlang: 'payload["safe"] == false'
steps:
- target: http://alert-service:8080/api/critical-alert
- target: kafka://clinical-alerts?brokers=kafka:9092
21.6 E-Commerce Personalization Engine¶
AI-powered product recommendations, dynamic pricing, customer segmentation, and personalized marketing.
helix:
cluster:
enabled: true
backend: redis
redis_url: "${env:REDIS_URL}"
metrics:
enabled: true
prometheus:
enabled: true
addr: "0.0.0.0:9090"
sla:
workflows:
recommendation-engine:
max_latency: 200ms
max_error_rate: 0.005
workflows:
# 1. Product catalog RAG index
- id: catalog-indexer
description: Index product catalog for AI-powered search
trigger:
uri: kafka://catalog-updates?brokers=kafka:9092&groupId=indexer
steps:
- transform:
starlang: |
product = read(payload, "application/json")
doc = {
"content": "\n".join([
product["name"],
product["description"],
"Category: " ++ product["category"],
"Brand: " ++ product.get("brand", ""),
"Price: $" ++ str(product["price"]),
"Tags: " ++ ", ".join(product.get("tags", [])),
"Features: " ++ "\n".join(product.get("features", [])),
]),
"metadata": {
"product_id": product["id"],
"category": product["category"],
"price": str(product["price"]),
"in_stock": str(product.get("in_stock", True)),
},
}
write(doc, "application/json")
- rag-chunk:
strategy: fixed
chunk_size: 500
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: qdrant
url: "${env:QDRANT_URL}"
collection: product-catalog
# 2. Customer behavior tracking
- id: behavior-tracker
description: Track and analyze customer behavior
trigger:
uri: kafka://clickstream?brokers=kafka:9092&groupId=behavior
steps:
- aggregate:
correlation: 'headers["CustomerId"]'
completion-size: 20
completion-timeout: 30s
strategy: starlang
expression: |
events = [read(m, "application/json") for m in messages]
profile = {
"customer_id": events[0]["customer_id"],
"viewed_products": [e["product_id"] for e in events if e["type"] == "view"],
"searched_terms": [e["query"] for e in events if e["type"] == "search"],
"cart_items": [e["product_id"] for e in events if e["type"] == "add_to_cart"],
"categories_browsed": list(set(e.get("category", "") for e in events if e.get("category"))),
"session_duration_ms": max(e["timestamp"] for e in events) - min(e["timestamp"] for e in events),
}
write(profile, "application/json")
- target: redis://cache:6379?command=SET&key=profile:${header.CustomerId}&ttl=3600
# 3. AI recommendation engine
- id: recommendation-engine
description: Personalized product recommendations
trigger:
uri: http-server://0.0.0.0:8080/api/recommendations
steps:
# Get customer profile from cache
- target: redis://cache:6379?command=GET&key=profile:${header.CustomerId}
- agent:
id: recommender
type: react
model:
provider: groq
model: llama-3.3-70b-versatile
api_key: "${env:GROQ_API_KEY}"
temperature: 0.4
system_prompt: |
You are a product recommendation engine. Based on the customer's
browsing history, search terms, and cart items, recommend products
they would likely purchase.
Strategy:
1. Analyze the customer's interests from their behavior profile
2. Search for products matching their interests
3. Consider cross-sell and upsell opportunities
4. Diversify recommendations across categories
5. Return top 10 product IDs with reasoning
Return JSON: {"recommendations": [{"product_id": "...", "reason": "...", "score": 0.95}]}
tools:
- name: search_products
type: rag-query
description: "Search product catalog by description, features, or category"
rag:
store: qdrant
url: "${env:QDRANT_URL}"
collection: product-catalog
top_k: 20
strategy: mmr
lambda: 0.6
embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
postprocess:
- type: metadata-filter
params:
conditions:
- key: in_stock
operator: eq
value: "true"
- name: get_purchase_history
type: connector
description: "Get customer's recent purchase history"
http:
method: GET
url: "http://order-service:8080/api/customers/${customer_id}/orders?limit=20"
parameters:
type: object
properties:
customer_id: { type: string }
required: ["customer_id"]
# 4. Smart search with natural language
- id: smart-product-search
description: Natural language product search
trigger:
uri: http-server://0.0.0.0:8080/api/search
steps:
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-retrieve:
strategy: hybrid
top_k: 20
alpha: 0.6
- rag-postprocess:
steps:
- type: metadata-filter
params:
conditions:
- key: in_stock
operator: eq
value: "true"
- type: cohere-rerank
params:
model: rerank-english-v3.0
api_key: "${env:COHERE_API_KEY}"
top_n: 10
- rag-synthesize:
mode: compact
model:
provider: groq
model: llama-3.3-70b-versatile
api_key: "${env:GROQ_API_KEY}"
system_prompt: |
Based on the search results, create a helpful product summary.
Highlight the most relevant products and explain why they match.
Format with product names, prices, and key features.
# 5. Dynamic pricing with market analysis
- id: dynamic-pricing
description: AI-driven dynamic pricing adjustments
trigger:
uri: cron://pricing?schedule=0 */2 * * *
steps:
# Gather market data
- multicast:
parallel: true
uris:
- http://competitor-scraper:8080/api/prices
- postgres://db:5432/analytics?operation=query&query=SELECT product_id, avg(quantity) as avg_demand, stddev(quantity) as demand_volatility FROM orders WHERE created_at > now() - interval '7 days' GROUP BY product_id
- redis://cache:6379?command=GET&key=inventory:levels
- aggregate:
correlation: "pricing-batch"
completion-size: 3
- agent:
id: pricing-agent
type: plan-execute
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.1
system_prompt: |
You are a pricing optimization engine. Analyze:
1. Competitor prices for each product
2. Demand trends (7-day moving average, volatility)
3. Current inventory levels
Apply pricing rules:
- Never price below cost + 10% margin
- Match competitor if within 5% of our price
- Increase price 5-15% for high-demand, low-inventory items
- Decrease price 5-10% for low-demand, high-inventory items
- Cap daily price changes at +/- 20%
Return JSON array of price adjustments.
- split:
starlang: 'payload'
steps:
- target: http://product-service:8080/api/products/${body.product_id}/price
- target: kafka://price-changes?brokers=kafka:9092
# 6. Abandoned cart recovery
- id: cart-recovery
description: AI-personalized abandoned cart emails
trigger:
uri: cron://cart-recovery?schedule=0 */1 * * *
steps:
- target: postgres://db:5432/carts?operation=query&query=SELECT * FROM carts WHERE status='abandoned' AND abandoned_at > now() - interval '24 hours' AND recovery_sent = false LIMIT 100
- split:
starlang: 'payload'
steps:
- agent:
id: email-writer
type: react
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Write a personalized cart recovery email. Include:
- Customer's name
- Items they left in cart with prices
- A compelling reason to complete the purchase
- A sense of urgency (limited stock, price expiring)
- A personalized discount code if cart value > $100
Keep the tone friendly and non-pushy.
Return JSON: {"subject": "...", "html_body": "...", "discount_code": "..."}
- target: smtp://email-server:587?from=shop@example.com&to=${body.customer_email}
- target: postgres://db:5432/carts?operation=update&query=UPDATE carts SET recovery_sent=true WHERE id=${body.cart_id}
21.7 Legal Document Analysis Platform¶
AI-powered contract review, clause extraction, risk assessment, and compliance checking.
helix:
audit:
enabled: true
max_entries: 50000
log_body: false
mcp:
server:
enabled: true
transport: http
addr: "0.0.0.0:3000"
name: "legal-ai"
agui:
servers:
- enabled: true
addr: "0.0.0.0:5000"
path: /api/legal-chat
workflows:
# 1. Legal knowledge base
- id: legal-kb-ingest
description: Ingest legal precedents, regulations, and templates
trigger:
uri: cron://legal-kb?schedule=0 3 * * 0
steps:
# Ingest regulations and case law
- rag-ingest:
reader: directory
path: /data/legal
recursive: true
- rag-chunk:
strategy: semantic
threshold: 0.4
max_chunk_size: 1500
embed:
provider: openai
model: text-embedding-3-large
api_key: "${env:OPENAI_API_KEY}"
- rag-extract:
fields: ["title", "summary", "keywords", "entities"]
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
- rag-embed:
provider: openai
model: text-embedding-3-large
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:LEGAL_VECTOR_DB}"
collection: legal-knowledge
# 2. Contract review pipeline
- id: contract-review
description: Multi-agent contract review
trigger:
uri: http-server://0.0.0.0:8080/api/review-contract
auth:
type: jwt
jwt:
secret: "${env:JWT_SECRET}"
claims:
role: ["attorney", "paralegal", "contract_manager"]
steps:
# OCR for scanned contracts
- rag-ingest:
reader: ocr-streaming
backend: azure
endpoint: "${env:AZURE_DOC_INTEL_ENDPOINT}"
api_key: "${env:AZURE_DOC_INTEL_KEY}"
model: prebuilt-contract
page_split: true
concurrency: 4
# Multi-agent review
- agent:
id: contract-review-team
type: sequential
sub_agents:
# Phase 1: Extract structure
- id: structure-extractor
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Extract the contract structure:
- Parties (names, roles, addresses)
- Effective date and term
- All defined terms
- Section headings and hierarchy
- Signature blocks
Return structured JSON.
# Phase 2: Clause analysis
- id: clause-analyzer
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Analyze each clause in this contract. For each:
- Classify type (indemnification, limitation of liability,
termination, IP, confidentiality, warranty, force majeure, etc.)
- Identify which party it favors
- Rate risk level (low/medium/high/critical)
- Note any unusual or non-standard language
- Compare to standard market terms
tools:
- name: search_precedents
type: rag-query
description: "Search legal precedents and standard clause libraries"
rag:
store: pgvector
connection: "${env:LEGAL_VECTOR_DB}"
collection: legal-knowledge
top_k: 5
strategy: hybrid
# Phase 3: Risk assessment
- id: risk-assessor
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.1
system_prompt: |
Based on the clause analysis, produce a risk assessment:
1. Overall risk score (1-10)
2. Top 5 risk items ranked by severity
3. Missing standard protections
4. Recommendations for negotiation
5. Red flags requiring immediate attention
tools:
- name: check_compliance
type: rag-query
description: "Check against regulatory compliance requirements"
rag:
store: pgvector
connection: "${env:LEGAL_VECTOR_DB}"
collection: legal-knowledge
top_k: 5
strategy: sub_question
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
# Phase 4: Executive summary
- id: summary-writer
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Write a clear executive summary of the contract review:
- 3-sentence overview
- Key commercial terms (table format)
- Risk summary with traffic-light indicators
- Recommended actions
- Items for negotiation (ranked by importance)
Format for a non-legal audience.
# Store review results
- target: postgres://db:5432/contract_reviews?operation=insert
- wiretap: slack://legal?channel=contract-reviews&webhook=${env:SLACK_WEBHOOK}
# 3. Clause comparison across contracts
- id: clause-comparison
description: Compare specific clauses across multiple contracts
trigger:
uri: http-server://0.0.0.0:8080/api/compare-clauses
steps:
- agent:
id: clause-comparator
type: plan-execute
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Compare the specified clause type across contracts:
1. Search for the relevant clause in each contract
2. Extract the key terms and conditions
3. Create a comparison matrix
4. Identify deviations from standard language
5. Recommend the strongest version
tools:
- name: search_contracts
type: rag-query
description: "Search stored contracts for specific clause types"
rag:
store: pgvector
connection: "${env:LEGAL_VECTOR_DB}"
collection: legal-knowledge
top_k: 10
strategy: hybrid
# 4. Legal Q&A for non-lawyers
- id: legal-qa
description: Plain-language legal Q&A
trigger:
uri: direct://legal-qa
steps:
- agent:
id: legal-assistant
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are a legal information assistant. Help users understand
legal concepts in plain language.
DISCLAIMER: You provide general legal information only.
You are not a lawyer and this is not legal advice.
Always recommend consulting a qualified attorney for specific situations.
memory:
enabled: true
type: sliding-window
window_size: 15
session_id_header: "X-Session-ID"
tools:
- name: search_legal_kb
type: rag-query
description: "Search legal knowledge base"
rag:
store: pgvector
connection: "${env:LEGAL_VECTOR_DB}"
collection: legal-knowledge
top_k: 5
strategy: hyde
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
21.8 Financial Trading Intelligence Platform¶
Real-time market analysis, sentiment tracking, risk monitoring, and trade signal generation.
helix:
cluster:
enabled: true
backend: redis
redis_url: "${env:REDIS_URL}"
metrics:
enabled: true
prometheus:
enabled: true
addr: "0.0.0.0:9090"
sla:
workflows:
market-analysis:
max_latency: 1s
risk-monitor:
max_latency: 500ms
audit:
enabled: true
max_entries: 500000
workflows:
# 1. Market data ingestion
- id: market-data-ingest
description: Ingest and index market research reports
trigger:
uri: kafka://market-reports?brokers=kafka:9092&groupId=ingest
steps:
- rag-ingest:
reader: auto
- rag-chunk:
strategy: recursive
chunk_size: 600
overlap: 100
- rag-extract:
fields: ["title", "summary", "keywords", "entities"]
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: market-research
# 2. Real-time news sentiment analysis
- id: news-sentiment
description: Analyze news sentiment for trading signals
trigger:
uri: kafka://financial-news?brokers=kafka:9092&groupId=sentiment
steps:
- throttle:
rate: 50
period: 1s
- agent:
id: sentiment-analyzer
type: react
model:
provider: groq
model: llama-3.3-70b-versatile
api_key: "${env:GROQ_API_KEY}"
temperature: 0.0
system_prompt: |
Analyze this financial news article for market impact:
- Overall sentiment: bullish / bearish / neutral
- Confidence: 0.0-1.0
- Affected tickers: [list]
- Sector impact: [list of sectors]
- Time horizon: immediate / short-term / long-term
- Key facts: [bullet points]
Return JSON only.
- choice:
when:
- starlang: 'payload["confidence"] > 0.8'
steps:
- target: kafka://high-confidence-signals?brokers=kafka:9092
- target: redis://signals:6379?command=SET&key=signal:${body.tickers[0]}&ttl=300
otherwise:
steps:
- target: kafka://low-confidence-signals?brokers=kafka:9092
- target: elasticsearch://analytics:9200?index=news-sentiment&operation=index
# 3. Multi-perspective market analysis
- id: market-analysis
description: Deep market analysis from multiple expert perspectives
trigger:
uri: http-server://0.0.0.0:8080/api/analyze
steps:
- agent:
id: analysis-team
type: parallel
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Synthesize all expert analyses into a unified market report:
- Consensus view with confidence level
- Key divergences between experts
- Risk-reward assessment
- Actionable insights with time horizon
sub_agents:
- id: fundamental-analyst
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Perform fundamental analysis:
- Revenue and earnings trends
- Valuation multiples vs peers
- Balance sheet health
- Management guidance vs consensus
tools:
- name: search_reports
type: rag-query
description: "Search market research reports"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: market-research
top_k: 8
strategy: hybrid
- name: get_financials
type: connector
description: "Get financial data for a ticker"
http:
method: GET
url: "http://data-api:8080/api/financials/${ticker}"
parameters:
type: object
properties:
ticker: { type: string }
required: ["ticker"]
- id: technical-analyst
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Perform technical analysis:
- Price trend and momentum indicators
- Support and resistance levels
- Volume analysis
- Chart pattern recognition
- Moving average signals
tools:
- name: get_price_data
type: connector
description: "Get historical price data"
http:
method: GET
url: "http://data-api:8080/api/prices/${ticker}?period=1y"
parameters:
type: object
properties:
ticker: { type: string }
required: ["ticker"]
- id: sentiment-analyst
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Analyze market sentiment:
- News sentiment trends
- Social media buzz
- Analyst consensus changes
- Insider trading activity
- Options flow analysis
tools:
- name: get_sentiment
type: connector
description: "Get aggregated sentiment data"
http:
method: GET
url: "http://data-api:8080/api/sentiment/${ticker}"
parameters:
type: object
properties:
ticker: { type: string }
required: ["ticker"]
- id: macro-analyst
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Analyze macroeconomic factors:
- Interest rate outlook
- Inflation trends
- Sector rotation signals
- Geopolitical risks
- Currency and commodity impacts
tools:
- name: search_macro_reports
type: rag-query
description: "Search macroeconomic research"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: market-research
top_k: 5
# 4. Portfolio risk monitoring
- id: risk-monitor
description: Real-time portfolio risk assessment
trigger:
uri: cron://risk-check?schedule=*/5 * * * *
steps:
- target: http://portfolio-api:8080/api/positions
- agent:
id: risk-monitor
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.0
system_prompt: |
Assess portfolio risk:
1. Check position concentration (>10% in any single name)
2. Sector exposure (>30% in any sector)
3. Correlation analysis (highly correlated positions)
4. Check against recent sentiment signals
5. Value at Risk estimation
Return JSON with risk flags and recommended actions.
tools:
- name: get_signals
type: connector
description: "Get recent trading signals"
http:
method: GET
url: "http://localhost:8080/api/signals/recent"
- name: search_risk_reports
type: rag-query
description: "Search risk research"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: market-research
top_k: 3
- choice:
when:
- starlang: 'payload["risk_level"] == "critical"'
steps:
- target: slack://trading?channel=risk-alerts&webhook=${env:SLACK_WEBHOOK}
- target: smtp://email:587?from=risk@firm.com&to=${env:RISK_MANAGER_EMAIL}
21.9 DevOps AI Operations (AIOps) Platform¶
Intelligent incident management, root cause analysis, automated remediation, and capacity planning.
helix:
metrics:
enabled: true
prometheus:
enabled: true
addr: "0.0.0.0:9090"
sla:
workflows:
incident-triage:
max_latency: 2s
auto-remediate:
max_latency: 10s
a2a:
servers:
- enabled: true
addr: "0.0.0.0:4000"
name: "aiops-agents"
mcp:
server:
enabled: true
transport: http
addr: "0.0.0.0:3000"
name: "aiops-tools"
workflows:
# 1. Runbook and postmortem ingestion
- id: runbook-ingest
description: Index runbooks, postmortems, and architecture docs
trigger:
uri: cron://runbook-sync?schedule=0 */6 * * *
steps:
- rag-ingest:
reader: directory
path: /data/ops-knowledge
recursive: true
- rag-chunk:
strategy: markdown
chunk_size: 800
retain_headers: true
- rag-extract:
fields: ["title", "keywords", "entities"]
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: ops-runbooks
# 2. Alert ingestion and deduplication
- id: alert-processor
description: Deduplicate and enrich alerts
trigger:
uri: kafka://raw-alerts?brokers=kafka:9092&groupId=alert-processor
steps:
- idempotent-consumer:
expression:
starlang: 'headers["AlertFingerprint"]'
repository: redis
redis-url: "${env:REDIS_URL}"
skip-duplicate: true
# Correlate related alerts
- aggregate:
correlation: 'headers["AlertService"]'
completion-size: 10
completion-timeout: 30s
strategy: starlang
expression: |
alerts = [read(m, "application/json") for m in messages]
correlated = {
"service": alerts[0]["service"],
"alert_count": sizeOf(alerts),
"severity": max(a["severity"] for a in alerts),
"alerts": alerts,
"first_seen": min(a["timestamp"] for a in alerts),
"last_seen": max(a["timestamp"] for a in alerts),
}
write(correlated, "application/json")
- target: seda://incident-triage?bufferSize=500
# 3. AI incident triage
- id: incident-triage
description: AI-powered incident classification and triage
trigger:
uri: seda://incident-triage?concurrentConsumers=4
steps:
- agent:
id: triage-agent
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.0
system_prompt: |
You are an SRE triage agent. For each incident:
1. Classify severity: P1 (critical), P2 (major), P3 (minor), P4 (informational)
2. Identify affected services and potential blast radius
3. Search runbooks for matching remediation procedures
4. Check for similar past incidents and their resolutions
5. Recommend immediate actions
Return JSON:
{
"severity": "P1-P4",
"category": "infrastructure|application|network|security|data",
"affected_services": [...],
"blast_radius": "isolated|partial|widespread",
"recommended_runbook": "runbook-id",
"immediate_actions": [...],
"similar_incidents": [...],
"auto_remediate": true/false
}
tools:
- name: search_runbooks
type: rag-query
description: "Search runbooks and past postmortems"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: ops-runbooks
top_k: 5
strategy: hybrid
- name: get_service_health
type: connector
description: "Check service health and metrics"
http:
method: GET
url: "http://monitoring:8080/api/services/${service_name}/health"
parameters:
type: object
properties:
service_name: { type: string }
required: ["service_name"]
- name: get_recent_deploys
type: connector
description: "Check recent deployments that might correlate"
http:
method: GET
url: "http://deploy-tracker:8080/api/deploys?since=2h"
- name: query_logs
type: connector
description: "Search application logs for errors"
http:
method: POST
url: "http://loki:3100/loki/api/v1/query_range"
parameters:
type: object
properties:
query: { type: string }
limit: { type: integer, default: 50 }
required: ["query"]
# Route by triage result
- choice:
when:
- starlang: 'payload["auto_remediate"] == true'
steps:
- target: seda://auto-remediate
- starlang: 'payload["severity"] == "P1"'
steps:
- target: slack://oncall?channel=incidents-critical&webhook=${env:SLACK_WEBHOOK}
- target: http://pagerduty:8080/api/trigger?severity=critical
otherwise:
steps:
- target: slack://oncall?channel=incidents&webhook=${env:SLACK_WEBHOOK}
- target: postgres://db:5432/incidents?operation=insert
# 4. Automated remediation
- id: auto-remediate
description: Execute automated remediation based on runbook
trigger:
uri: seda://auto-remediate?concurrentConsumers=2
steps:
- agent:
id: remediation-agent
type: plan-execute
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.0
system_prompt: |
You are an automated remediation agent. Given the incident
triage and recommended runbook:
1. Plan the remediation steps based on the runbook
2. Execute each step, checking service health after each
3. If any step fails, stop and escalate to human
4. After remediation, verify the issue is resolved
5. Document what was done for the postmortem
SAFETY RULES:
- Never restart more than 2 services simultaneously
- Never scale beyond 3x current capacity
- Always check health after each action
- Escalate if remediation takes >10 minutes
max_iterations: 15
tools:
- name: restart_service
type: connector
description: "Restart a service (rolling restart)"
http:
method: POST
url: "http://k8s-api:8080/api/services/${service}/restart"
parameters:
type: object
properties:
service: { type: string }
required: ["service"]
- name: scale_service
type: connector
description: "Scale a service to N replicas"
http:
method: POST
url: "http://k8s-api:8080/api/services/${service}/scale"
parameters:
type: object
properties:
service: { type: string }
replicas: { type: integer }
required: ["service", "replicas"]
- name: check_health
type: connector
description: "Check service health after remediation"
http:
method: GET
url: "http://monitoring:8080/api/services/${service}/health"
parameters:
type: object
properties:
service: { type: string }
required: ["service"]
- name: rollback_deploy
type: connector
description: "Rollback to previous deployment"
http:
method: POST
url: "http://deploy-tracker:8080/api/services/${service}/rollback"
parameters:
type: object
properties:
service: { type: string }
required: ["service"]
- target: slack://oncall?channel=incidents&webhook=${env:SLACK_WEBHOOK}
- target: postgres://db:5432/remediations?operation=insert
# 5. Root cause analysis
- id: root-cause-analysis
description: Deep agent for root cause analysis
trigger:
uri: http-server://0.0.0.0:8080/api/rca
steps:
- agent:
id: rca-agent
type: deep
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.1
system_prompt: |
Perform root cause analysis for this incident:
1. Gather timeline of events (logs, alerts, deploys, changes)
2. Identify potential root causes
3. For each potential cause, gather evidence for/against
4. Determine the most likely root cause
5. Identify contributing factors
6. Recommend preventive measures
Use the 5 Whys methodology to drill into the root cause.
deep_max_depth: 4
max_iterations: 20
tools:
- name: search_runbooks
type: rag-query
description: "Search past postmortems and known issues"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: ops-runbooks
top_k: 5
- name: query_logs
type: connector
description: "Query application logs"
http:
method: POST
url: "http://loki:3100/loki/api/v1/query_range"
parameters:
type: object
properties:
query: { type: string }
start: { type: string }
end: { type: string }
required: ["query"]
- name: query_metrics
type: connector
description: "Query Prometheus metrics"
http:
method: GET
url: "http://prometheus:9090/api/v1/query_range"
parameters:
type: object
properties:
query: { type: string }
start: { type: string }
end: { type: string }
step: { type: string, default: "1m" }
required: ["query"]
- name: get_change_log
type: connector
description: "Get infrastructure and deployment changes"
http:
method: GET
url: "http://deploy-tracker:8080/api/changes?since=${since}"
parameters:
type: object
properties:
since: { type: string }
required: ["since"]
# 6. Capacity planning
- id: capacity-planning
description: AI-driven capacity forecasting
trigger:
uri: cron://capacity?schedule=0 6 * * 1
steps:
- multicast:
parallel: true
uris:
- http://prometheus:9090/api/v1/query?query=avg_over_time(container_cpu_usage_total[7d])
- http://prometheus:9090/api/v1/query?query=avg_over_time(container_memory_usage_bytes[7d])
- http://prometheus:9090/api/v1/query?query=sum(rate(http_requests_total[7d]))
- aggregate:
correlation: "capacity-batch"
completion-size: 3
- agent:
id: capacity-planner
type: plan-execute
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Analyze resource utilization trends and forecast capacity needs:
1. Current CPU, memory, and request rate utilization per service
2. Growth trend (linear regression over last 4 weeks)
3. Predicted utilization in 2 weeks, 1 month, 3 months
4. Services at risk of capacity issues
5. Scaling recommendations with cost estimates
Return JSON with per-service forecasts and recommendations.
- target: slack://infra?channel=capacity-reports&webhook=${env:SLACK_WEBHOOK}
- target: postgres://db:5432/capacity_reports?operation=insert
21.10 Education & Learning Platform¶
Adaptive learning with personalized tutoring, quiz generation, progress tracking, and curriculum planning.
helix:
agui:
servers:
- enabled: true
addr: "0.0.0.0:5000"
path: /api/tutor
name: "tutor-chat"
mcp:
server:
enabled: true
transport: http
addr: "0.0.0.0:3000"
name: "education-platform"
workflows:
# 1. Course content ingestion
- id: course-ingest
description: Ingest textbooks, lectures, and course materials
trigger:
uri: cron://course-sync?schedule=0 4 * * *
steps:
- rag-ingest:
reader: directory
path: /data/courses
recursive: true
- rag-chunk:
strategy: hierarchical
levels:
- chunk_size: 2000 # Topic level
- chunk_size: 500 # Concept level
- chunk_size: 150 # Detail level
- rag-extract:
fields: ["title", "summary", "keywords"]
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
# Build knowledge graph of concepts
- rag-graph-build:
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
entity_types: ["Concept", "Formula", "Definition", "Example", "Prerequisite"]
relationship_types: ["requires", "explains", "illustrates", "extends", "contradicts"]
store:
backend: neo4j
url: bolt://neo4j:7687
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: course-materials
# 2. Adaptive tutoring agent
- id: adaptive-tutor
description: Personalized AI tutor that adapts to student level
trigger:
uri: http-server://0.0.0.0:8080/api/tutor
steps:
# Fetch student profile
- target: redis://cache:6379?command=GET&key=student:${header.StudentId}
- agent:
id: tutor
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.5
system_prompt: |
You are an expert tutor. Adapt your teaching style to the student's level.
Teaching principles:
- Use the Socratic method: ask guiding questions rather than giving answers
- Build on what the student already knows (zone of proximal development)
- Use analogies and real-world examples
- Break complex topics into digestible steps
- Encourage when the student struggles
- If the student is advanced, challenge them with deeper questions
- Track which concepts the student has mastered
After each interaction, assess the student's understanding level (1-5)
and update the recommended next topics.
max_iterations: 8
memory:
enabled: true
type: composite
session_id_header: "X-Session-ID"
backends:
- type: sliding-window
window_size: 30
- type: sql-long-term
dsn: "${env:DATABASE_URL}"
- type: vector
top_k: 5
embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
tools:
- name: search_course_material
type: rag-query
description: "Search course materials for explanations and examples"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: course-materials
top_k: 5
strategy: auto_merging
merge_threshold: 0.5
- name: find_prerequisites
type: rag-query
description: "Find prerequisite concepts via knowledge graph"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: course-materials
top_k: 3
strategy: sub_question
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
- name: get_student_progress
type: connector
description: "Get student's learning progress and mastered concepts"
http:
method: GET
url: "http://progress-api:8080/api/students/${student_id}/progress"
parameters:
type: object
properties:
student_id: { type: string }
required: ["student_id"]
- name: update_progress
type: connector
description: "Update student's concept mastery level"
http:
method: POST
url: "http://progress-api:8080/api/students/${student_id}/progress"
parameters:
type: object
properties:
student_id: { type: string }
concept: { type: string }
mastery_level: { type: integer, minimum: 1, maximum: 5 }
required: ["student_id", "concept", "mastery_level"]
# 3. Quiz generation
- id: quiz-generator
description: Generate adaptive quizzes based on student progress
trigger:
uri: http-server://0.0.0.0:8080/api/quiz/generate
steps:
# Get student progress to target weak areas
- target: http://progress-api:8080/api/students/${header.StudentId}/weak-areas
- agent:
id: quiz-agent
type: plan-execute
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Generate a quiz tailored to the student's weak areas.
Quiz requirements:
1. Mix question types: multiple choice, short answer, true/false, fill-in-blank
2. Include 2-3 easy questions (confidence builders)
3. Focus 60% on weak areas
4. Include 1-2 challenge questions slightly above their level
5. Reference specific course material in explanations
6. Provide detailed explanations for each answer
Return JSON:
{
"quiz_id": "uuid",
"questions": [
{
"id": 1,
"type": "multiple_choice",
"difficulty": "easy|medium|hard",
"concept": "concept name",
"question": "...",
"options": ["A", "B", "C", "D"],
"correct_answer": "B",
"explanation": "..."
}
]
}
tools:
- name: search_material
type: rag-query
description: "Search course material for quiz content"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: course-materials
top_k: 10
strategy: hybrid
- target: postgres://db:5432/quizzes?operation=insert
# 4. Essay grading with feedback
- id: essay-grader
description: AI-powered essay grading with constructive feedback
trigger:
uri: http-server://0.0.0.0:8080/api/grade-essay
steps:
- agent:
id: grading-team
type: sequential
sub_agents:
# Rubric evaluation
- id: rubric-evaluator
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.1
system_prompt: |
Evaluate this essay against the rubric:
- Thesis clarity (0-20)
- Evidence and support (0-25)
- Organization and structure (0-20)
- Writing quality and grammar (0-15)
- Critical thinking and analysis (0-20)
For each criterion:
- Score with brief justification
- Specific examples from the text
- One actionable improvement suggestion
# Plagiarism and originality check
- id: originality-checker
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Assess the originality of this essay:
- Check for unusually sophisticated language vs student level
- Look for inconsistent writing style (possible patchwork)
- Search course materials to check if content is directly copied
- Flag any passages that seem AI-generated
tools:
- name: search_sources
type: rag-query
description: "Search course materials for potential source matching"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: course-materials
top_k: 10
# Constructive feedback writer
- id: feedback-writer
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Based on the rubric evaluation and originality check,
write constructive, encouraging feedback:
- Start with what the student did well (specific praise)
- Address the top 3 improvement areas with clear guidance
- Suggest specific revision strategies
- End with encouragement and next steps
- Tone: supportive mentor, not harsh critic
- target: postgres://db:5432/grades?operation=insert
# 5. Curriculum recommendation engine
- id: curriculum-recommender
description: Recommend personalized learning paths
trigger:
uri: http-server://0.0.0.0:8080/api/learning-path
steps:
- target: http://progress-api:8080/api/students/${header.StudentId}/full-profile
- agent:
id: curriculum-planner
type: plan-execute
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Design a personalized learning path for this student:
1. Assess current knowledge from their progress profile
2. Identify prerequisite gaps using the knowledge graph
3. Determine their learning goals
4. Create a sequenced curriculum:
- Fill prerequisite gaps first
- Build progressively in difficulty
- Mix theory with practice
- Include review/reinforcement periods
5. Estimate time for each module
6. Suggest supplementary resources
Return a structured learning path with milestones.
tools:
- name: search_course_catalog
type: rag-query
description: "Search available courses and modules"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: course-materials
top_k: 20
strategy: hybrid
- name: query_prerequisites
type: connector
description: "Query the knowledge graph for concept prerequisites"
http:
method: POST
url: "http://neo4j-api:8080/api/query"
parameters:
type: object
properties:
concept: { type: string }
depth: { type: integer, default: 3 }
required: ["concept"]
21.11 Supply Chain Intelligence Platform¶
End-to-end supply chain visibility with demand forecasting, supplier risk assessment, and automated procurement.
helix:
cluster:
enabled: true
backend: redis
redis_url: "${env:REDIS_URL}"
metrics:
enabled: true
prometheus:
enabled: true
addr: "0.0.0.0:9090"
audit:
enabled: true
workflows:
# 1. Supplier intelligence RAG
- id: supplier-intel-ingest
description: Ingest supplier reports, certifications, and news
trigger:
uri: cron://supplier-sync?schedule=0 5 * * *
steps:
- multicast:
parallel: true
uris:
# Supplier documents
- file:///data/supplier-docs?recursive=true
# Supplier news feeds
- http://news-aggregator:8080/api/supplier-news?since=24h
- rag-chunk:
strategy: recursive
chunk_size: 800
overlap: 150
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: supplier-intelligence
# 2. Demand forecasting
- id: demand-forecast
description: AI-powered demand forecasting
trigger:
uri: cron://forecast?schedule=0 7 * * 1
steps:
- multicast:
parallel: true
uris:
- postgres://db:5432/sales?operation=query&query=SELECT product_id, date_trunc('week', sale_date) as week, sum(quantity) as units FROM sales WHERE sale_date > now() - interval '52 weeks' GROUP BY 1, 2 ORDER BY 1, 2
- postgres://db:5432/inventory?operation=query&query=SELECT product_id, current_stock, reorder_point, lead_time_days FROM inventory
- http://external-data:8080/api/economic-indicators
- aggregate:
correlation: "forecast-batch"
completion-size: 3
- agent:
id: demand-forecaster
type: plan-execute
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Forecast demand for the next 4 weeks:
1. Analyze 52-week sales trends for each product
2. Identify seasonality patterns
3. Factor in economic indicators
4. Compare current inventory to predicted demand
5. Flag products at risk of stockout
6. Recommend optimal order quantities and timing
Consider:
- Lead times for each product
- Minimum order quantities
- Storage capacity constraints
- Bulk discount thresholds
Return JSON with per-product forecasts and order recommendations.
tools:
- name: search_market_trends
type: rag-query
description: "Search market trend reports"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: supplier-intelligence
top_k: 5
- split:
starlang: 'payload["recommendations"]'
steps:
- choice:
when:
- starlang: 'payload["urgency"] == "critical"'
steps:
- target: seda://auto-procurement
- target: slack://supply-chain?channel=urgent-orders&webhook=${env:SLACK_WEBHOOK}
otherwise:
steps:
- target: postgres://db:5432/order_recommendations?operation=insert
# 3. Supplier risk assessment
- id: supplier-risk
description: Continuous supplier risk monitoring
trigger:
uri: cron://supplier-risk?schedule=0 8 * * *
steps:
- target: postgres://db:5432/suppliers?operation=query&query=SELECT * FROM active_suppliers
- split:
starlang: 'payload'
parallel: true
steps:
- agent:
id: risk-assessor
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.1
system_prompt: |
Assess this supplier's risk profile:
1. Search for recent news (financial trouble, lawsuits, recalls)
2. Check delivery performance history
3. Evaluate geographic and geopolitical risks
4. Assess financial stability indicators
5. Check compliance and certification status
Return:
{
"supplier_id": "...",
"risk_score": 1-10,
"risk_factors": [...],
"recommendation": "maintain|watch|diversify|replace",
"alternative_suppliers": [...]
}
tools:
- name: search_supplier_news
type: rag-query
description: "Search supplier news and reports"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: supplier-intelligence
top_k: 5
strategy: hybrid
- name: get_delivery_history
type: connector
description: "Get supplier delivery performance"
http:
method: GET
url: "http://procurement-api:8080/api/suppliers/${supplier_id}/performance"
parameters:
type: object
properties:
supplier_id: { type: string }
required: ["supplier_id"]
- aggregate:
correlation: "risk-batch"
strategy: starlang
expression: |
assessments = [read(m, "application/json") for m in messages]
high_risk = [a for a in assessments if a["risk_score"] >= 7]
write({"total": sizeOf(assessments), "high_risk": high_risk, "avg_score": sum(a["risk_score"] for a in assessments) / sizeOf(assessments)}, "application/json")
completion-timeout: 60s
- choice:
when:
- starlang: 'payload["high_risk"][0]'
steps:
- target: slack://supply-chain?channel=supplier-risks&webhook=${env:SLACK_WEBHOOK}
- target: postgres://db:5432/risk_assessments?operation=insert
# 4. Automated procurement
- id: auto-procurement
description: Automated purchase order generation
trigger:
uri: seda://auto-procurement?concurrentConsumers=2
steps:
- agent:
id: procurement-agent
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.0
system_prompt: |
Generate a purchase order based on the demand forecast:
1. Select the best supplier (price, reliability, lead time)
2. Calculate optimal order quantity (EOQ formula)
3. Apply any volume discounts
4. Set delivery date based on lead time + safety margin
5. Generate PO in standard format
Return JSON PO that can be sent to the supplier API.
tools:
- name: get_supplier_quotes
type: connector
description: "Get current quotes from approved suppliers"
http:
method: GET
url: "http://procurement-api:8080/api/quotes?product_id=${product_id}"
parameters:
type: object
properties:
product_id: { type: string }
required: ["product_id"]
- name: check_budget
type: connector
description: "Check available procurement budget"
http:
method: GET
url: "http://finance-api:8080/api/budget/procurement"
- choice:
when:
- starlang: 'payload["total_cost"] < 10000'
steps:
# Auto-approve small orders
- target: http://procurement-api:8080/api/purchase-orders
- target: smtp://email:587?from=procurement@company.com&to=${body.supplier_email}
otherwise:
steps:
# Large orders need approval
- target: slack://supply-chain?channel=po-approvals&webhook=${env:SLACK_WEBHOOK}
- target: postgres://db:5432/pending_orders?operation=insert
21.12 Content Moderation & Safety Platform¶
AI-powered content moderation with multi-layer analysis, appeals handling, and policy enforcement.
helix:
metrics:
enabled: true
prometheus:
enabled: true
addr: "0.0.0.0:9090"
sla:
workflows:
content-moderation:
max_latency: 3s
max_error_rate: 0.001
workflows:
# 1. Policy knowledge base
- id: policy-ingest
description: Ingest content moderation policies and guidelines
trigger:
uri: cron://policy-sync?schedule=0 0 * * *
steps:
- rag-ingest:
reader: directory
path: /data/policies
recursive: true
- rag-chunk:
strategy: markdown
chunk_size: 600
retain_headers: true
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: moderation-policies
# 2. Multi-layer content moderation
- id: content-moderation
description: AI content moderation pipeline
trigger:
uri: kafka://user-content?brokers=kafka:9092&groupId=moderation
steps:
# Layer 1: Fast keyword/pattern check
- transform:
starlang: |
content = read(payload, "application/json")
text = content.get("text", "")
# Fast pattern matching for obvious violations
severe_patterns = ["explicit_term_1", "explicit_term_2"]
has_severe = any(p in text.lower() for p in severe_patterns)
content["fast_check"] = {"severe": has_severe}
write(content, "application/json")
- choice:
when:
- starlang: 'payload["fast_check"]["severe"] == true'
steps:
# Immediate block for severe violations
- set-header:
name: ModerationDecision
constant: blocked
- target: kafka://moderation-results?brokers=kafka:9092
otherwise:
steps:
# Layer 2: AI analysis
- agent:
id: content-moderator
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.0
system_prompt: |
You are a content moderator. Analyze this content for:
1. Hate speech or discrimination
2. Harassment or bullying
3. Violence or threats
4. Sexual content
5. Misinformation or harmful advice
6. Spam or manipulation
7. Personal information exposure (PII)
8. Self-harm or dangerous behavior
For each category:
- Score: 0.0 (safe) to 1.0 (clear violation)
- Evidence: specific text that triggered the flag
Return JSON:
{
"decision": "approve|flag|block",
"confidence": 0.95,
"categories": {...},
"explanation": "...",
"suggested_action": "none|warn_user|remove_content|suspend_account"
}
tools:
- name: check_policy
type: rag-query
description: "Check content against moderation policies"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: moderation-policies
top_k: 3
# Route by decision
- choice:
when:
- starlang: 'payload["decision"] == "block"'
steps:
- target: kafka://moderation-results?brokers=kafka:9092
- target: postgres://db:5432/moderation_log?operation=insert
- starlang: 'payload["decision"] == "flag"'
steps:
# Layer 3: Human review queue for edge cases
- target: kafka://human-review-queue?brokers=kafka:9092
otherwise:
steps:
- target: kafka://moderation-results?brokers=kafka:9092
- target: elasticsearch://analytics:9200?index=moderation-events&operation=index
# 3. Appeal handling
- id: appeal-handler
description: Handle user appeals of moderation decisions
trigger:
uri: http-server://0.0.0.0:8080/api/appeals
steps:
# Fetch original moderation decision
- target: postgres://db:5432/moderation_log?operation=query&query=SELECT * FROM moderation_log WHERE content_id = '${header.ContentId}'
- agent:
id: appeal-reviewer
type: sequential
sub_agents:
# Independent re-review
- id: re-reviewer
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.0
system_prompt: |
Independently re-review this content:
- Ignore the original moderation decision
- Apply current policies strictly
- Consider cultural context and nuance
- Look for satire, sarcasm, or educational intent
- Rate each violation category independently
tools:
- name: check_policy
type: rag-query
description: "Check moderation policies"
rag:
store: pgvector
connection: "${env:VECTOR_DB_URL}"
collection: moderation-policies
top_k: 5
# Compare and decide
- id: appeal-decider
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
Compare the original moderation decision with the re-review:
- If both agree: maintain the decision
- If they disagree: lean toward the less restrictive option
- Consider the user's appeal reasoning
- Provide a clear, empathetic explanation to the user
Return:
{
"appeal_decision": "upheld|overturned|modified",
"new_action": "...",
"explanation_to_user": "...",
"internal_notes": "..."
}
- target: postgres://db:5432/appeals?operation=insert
- target: kafka://appeal-results?brokers=kafka:9092
# 4. Trend analysis and policy updates
- id: moderation-trends
description: Analyze moderation trends and suggest policy updates
trigger:
uri: cron://moderation-trends?schedule=0 9 * * 1
steps:
- target: postgres://db:5432/moderation_log?operation=query&query=SELECT category, count(*), avg(confidence) FROM moderation_log WHERE created_at > now() - interval '7 days' GROUP BY category ORDER BY count DESC
- agent:
id: trend-analyst
type: plan-execute
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
Analyze this week's moderation data:
1. Identify emerging trends (new types of violations)
2. Categories with increasing/decreasing volume
3. False positive rate estimation
4. Appeals success rate by category
5. Recommend policy clarifications or updates
6. Suggest automated rule improvements
Format as a weekly moderation report.
- target: slack://trust-safety?channel=moderation-reports&webhook=${env:SLACK_WEBHOOK}
22. Advanced AI Agent & RAG Patterns¶
Advanced production patterns that combine multiple agent types, RAG pipeline stages, memory systems, tool integration, and protocol interop into sophisticated AI systems.
22.1 Self-Reflective Agent with Quality Feedback Loop¶
An agent that evaluates its own responses, critiques weaknesses, and iterates until quality thresholds are met. Uses a loop agent with an inner ReAct that has RAG tools, plus evaluation metrics to decide when output is good enough.
workflows:
# Ingest knowledge base on startup
- id: ingest-knowledge-base
trigger:
uri: timer://startup?repeatCount=1
steps:
- rag-ingest:
reader: directory
path: ./knowledge-base
recursive: true
extensions:
- ".md"
- ".pdf"
- ".txt"
- rag-chunk:
strategy: semantic
threshold: 0.5
chunk_size: 1024
overlap: 128
- rag-embed:
provider: openai
model: text-embedding-3-large
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
dimensions: 3072
# Self-reflective Q&A endpoint
- id: self-reflective-qa
trigger:
uri: http-server://0.0.0.0:8080/api/ask
steps:
- agent:
type: loop
max_iterations: 4
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.3
system_prompt: |
You are a quality-driven research assistant. For each iteration:
1. RESEARCH: Use search_knowledge to find relevant information
2. DRAFT: Write a comprehensive answer based on retrieved context
3. SELF-EVALUATE: Score your answer on these criteria (each 1-10):
- Completeness: Does it address all aspects of the question?
- Accuracy: Is every claim supported by the retrieved sources?
- Clarity: Is it well-structured and easy to understand?
- Specificity: Does it include concrete details, not vague generalities?
4. DECIDE:
- If ALL scores >= 8: Output "COMPLETE" followed by your final answer
- Otherwise: Identify the weakest criterion, explain why, and search
for additional information to improve it. Then revise and re-evaluate.
Always cite sources by document name when making factual claims.
tools:
- name: search_knowledge
type: rag-query
description: "Search the knowledge base. Use specific, targeted queries."
parameters:
type: object
properties:
query:
type: string
description: "Precise search query targeting specific information"
top_k:
type: integer
default: 5
description: "Number of results (increase for broader searches)"
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
retriever: hybrid
alpha: 0.7
- name: search_diverse
type: rag-query
description: "Search with diversity emphasis to find complementary information"
parameters:
type: object
properties:
query:
type: string
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
retriever: mmr
lambda: 0.3
fetch_k: 20
22.2 Multi-Agent Research Team with Shared RAG Memory¶
A supervisor coordinates specialized research agents that share a common knowledge base and contribute findings to a shared vector memory. Each agent has domain expertise and can build on discoveries from other agents.
workflows:
# Shared knowledge base ingestion
- id: ingest-research-corpus
trigger:
uri: timer://startup?repeatCount=1
steps:
- rag-ingest:
reader: directory
path: ./research-corpus
recursive: true
- rag-chunk:
strategy: recursive
chunk_size: 800
overlap: 100
separators:
- "\n\n"
- "\n"
- ". "
- " "
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: research_corpus
# Multi-agent research endpoint
- id: deep-research
trigger:
uri: http-server://0.0.0.0:8080/api/research
steps:
- agent:
type: supervisor
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.4
system_prompt: |
You are a research director coordinating a team of specialists.
Analyze the user's research question and delegate to the right agents:
- literature_analyst: For finding and synthesizing published research, papers, and documented knowledge
- data_investigator: For quantitative analysis, finding statistics, metrics, and data-driven insights
- critical_reviewer: For evaluating claims, finding contradictions, assessing evidence quality
- synthesis_writer: For combining all findings into a coherent, well-structured research report
Strategy:
1. First send to literature_analyst to establish the knowledge foundation
2. Then data_investigator to find supporting/contradicting quantitative evidence
3. Then critical_reviewer to assess the quality and gaps in findings
4. Finally synthesis_writer to produce the final report
You may route to multiple agents or re-route if initial findings are insufficient.
memory:
enabled: true
type: composite
backends:
- type: sliding-window
window_size: 40
- type: vector
top_k: 10
embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
sub_agents:
- id: literature_analyst
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.2
system_prompt: |
You are a literature analyst. Your job is to find and synthesize
relevant published knowledge on the topic. Use the search tools
to find relevant documents, then provide a structured literature
review with:
- Key findings from each source
- Areas of consensus across sources
- Identified gaps in the literature
- Relevant quotes with source attribution
Use multiple search queries with different phrasings to ensure
comprehensive coverage. Search at least 3 different angles.
tools:
- name: search_corpus
type: rag-query
description: "Search the research corpus for relevant documents"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 8
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: research_corpus
retriever: hybrid
alpha: 0.6
- name: search_broad
type: rag-query
description: "Broad search for diverse perspectives on a topic"
parameters:
type: object
properties:
query:
type: string
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: research_corpus
retriever: mmr
lambda: 0.4
fetch_k: 25
- name: save_finding
type: connector
description: "Save an important finding to shared research memory"
parameters:
type: object
properties:
finding:
type: string
category:
type: string
enum:
- key_insight
- supporting_evidence
- contradiction
- gap
source:
type: string
config:
endpoint: "http://localhost:8080/api/findings"
method: POST
max_iterations: 8
- id: data_investigator
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.1
system_prompt: |
You are a data investigator. Find quantitative evidence:
statistics, metrics, benchmarks, performance numbers, dates,
and measurable outcomes. Cross-reference numbers across sources.
Flag any inconsistencies in reported data.
Present findings as structured data points with source citations.
tools:
- name: search_data
type: rag-query
description: "Search for quantitative data and statistics"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 10
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: research_corpus
retriever: hybrid
alpha: 0.8
max_iterations: 6
- id: critical_reviewer
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.5
system_prompt: |
You are a critical reviewer. Evaluate the research findings
provided to you. Assess:
1. Evidence quality: Are claims well-supported?
2. Bias detection: Are there one-sided perspectives?
3. Logical consistency: Do the findings contradict each other?
4. Completeness: What important aspects are missing?
5. Recency: Is the information current or outdated?
Search for counter-evidence and alternative viewpoints.
Be constructively critical — identify weaknesses AND suggest
what additional evidence would strengthen the analysis.
tools:
- name: search_counter_evidence
type: rag-query
description: "Search for evidence that contradicts or qualifies current findings"
parameters:
type: object
properties:
query:
type: string
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: research_corpus
retriever: mmr
lambda: 0.2
fetch_k: 30
max_iterations: 5
- id: synthesis_writer
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.6
max_tokens: 4096
system_prompt: |
You are a synthesis writer. Take all research findings, data,
and critical review and produce a comprehensive research report:
## Structure
1. Executive Summary (2-3 sentences)
2. Key Findings (numbered, with confidence levels)
3. Supporting Evidence (data points, quotes)
4. Contradictions & Uncertainties
5. Knowledge Gaps & Future Research
6. Conclusion & Recommendations
7. Sources
Use clear academic language. Distinguish between well-supported
claims and tentative conclusions. Include confidence levels
(High/Medium/Low) for each finding.
max_iterations: 3
22.3 Conversational RAG with Persistent Session Memory¶
A production chat system that maintains conversation history in PostgreSQL, uses vector memory for semantic recall of past interactions, and performs RAG over a knowledge base — with query rewriting based on conversation context.
workflows:
# Knowledge base ingestion (scheduled)
- id: nightly-ingestion
trigger:
uri: cron://nightly?schedule=0 2 * * *
steps:
- rag-ingest:
reader: directory
path: ./docs
recursive: true
- rag-chunk:
strategy: semantic
threshold: 0.45
chunk_size: 1000
overlap: 150
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_docs
# Conversational RAG chat endpoint
- id: conversational-rag-chat
trigger:
uri: http-server://0.0.0.0:8080/api/chat
steps:
- agent:
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.4
max_tokens: 2048
system_prompt: |
You are a helpful assistant with access to a knowledge base and
memory of past conversations with this user.
IMPORTANT WORKFLOW:
1. First, check recall_past_conversations if the user references
something from a previous chat ("as we discussed", "remember when",
"last time", etc.)
2. Rewrite the user's question as a standalone query incorporating
conversation context (resolve pronouns, references, implicit topics)
3. Search the knowledge base with your rewritten query
4. If results are insufficient, try alternative phrasings or broader queries
5. Synthesize an answer grounded in the retrieved documents
6. Always cite which documents support your claims
If you cannot find relevant information, say so honestly rather than
guessing. Distinguish between "the knowledge base doesn't cover this"
and "I found partial information."
memory:
enabled: true
type: composite
backends:
# Recent conversation context (current session)
- type: sliding-window
window_size: 20
# Semantic recall of past sessions
- type: vector
top_k: 5
embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
# Persistent cross-session memory
- type: sql-long-term
dsn: "${env:DATABASE_URL}"
session_id_header: "X-Session-ID"
tools:
- name: search_knowledge
type: rag-query
description: "Search the knowledge base for information relevant to the user's question. Rewrite the query to be standalone and specific."
parameters:
type: object
properties:
query:
type: string
description: "Standalone search query (rewritten from conversation context)"
top_k:
type: integer
default: 5
required:
- query
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_docs
retriever: hybrid
alpha: 0.7
- name: search_detailed
type: rag-query
description: "Search for specific detailed information when initial search was too broad"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 3
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_docs
retriever: simple
- name: recall_past_conversations
type: rag-query
description: "Search past conversation history for relevant context when user references previous discussions"
parameters:
type: object
properties:
query:
type: string
description: "What to search for in past conversations"
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: conversation_history
retriever: mmr
lambda: 0.6
max_iterations: 6
# Background: index conversation turns for future recall
- id: index-conversations
trigger:
uri: seda://conversation-indexer?concurrentConsumers=2
steps:
- rag-chunk:
strategy: sentence
chunk_size: 500
overlap: 50
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: conversation_history
22.4 Plan-Execute Agent for Complex Multi-Step RAG Research¶
A plan-execute agent that creates a research plan, then executes each step with targeted RAG retrieval, building up a comprehensive answer through systematic investigation.
workflows:
- id: planned-research
trigger:
uri: http-server://0.0.0.0:8080/api/research
steps:
- agent:
type: plan-execute
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.3
system_prompt: |
You are a systematic research agent. When given a complex question:
PLANNING PHASE:
Create a research plan as a JSON array of steps. Each step should be
a focused, answerable sub-question. Order steps so that earlier findings
inform later searches. Example plan:
[
"What is the current state of X?",
"What are the main challenges of X?",
"What solutions have been proposed for the challenges identified?",
"What evidence supports the effectiveness of these solutions?",
"Synthesize findings into recommendations"
]
EXECUTION PHASE:
For each step, use the search tools to find relevant information.
Use different search strategies for different types of questions:
- Factual queries: use search_precise with specific terms
- Conceptual queries: use search_broad for diverse perspectives
- Comparative queries: search for each item separately, then compare
Accumulate findings across steps. Reference earlier findings when
they inform later analysis.
FINAL OUTPUT:
A structured report with sections corresponding to your plan steps,
a synthesis section connecting the findings, and actionable recommendations.
tools:
- name: search_precise
type: rag-query
description: "Precise search for specific facts, definitions, or named concepts"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: research_docs
retriever: hybrid
alpha: 0.8
- name: search_broad
type: rag-query
description: "Broad search for diverse perspectives and related concepts"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 8
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: research_docs
retriever: mmr
lambda: 0.3
fetch_k: 30
- name: search_similar
type: rag-query
description: "Find documents semantically similar to a specific passage or concept"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: research_docs
retriever: simple
max_iterations: 15
22.5 Deep Agent for Hierarchical Knowledge Exploration¶
A deep agent that recursively decomposes complex topics, exploring each branch with targeted RAG retrieval. Each recursion level builds on parent findings, creating a hierarchical knowledge map.
workflows:
- id: deep-knowledge-exploration
trigger:
uri: http-server://0.0.0.0:8080/api/explore
steps:
- agent:
type: deep
deep_max_depth: 3
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.4
max_tokens: 3000
system_prompt: |
You are a deep knowledge explorer. Given a complex topic:
DECOMPOSITION RULES:
- Break the topic into 2-4 orthogonal subtopics
- Each subtopic should be independently researchable
- Subtopics should collectively cover the full scope
- At each level, search the knowledge base BEFORE decomposing further
- Stop decomposing when you have enough detail to answer confidently
SEARCH STRATEGY BY DEPTH:
- Depth 0 (root): Broad overview searches to map the landscape
- Depth 1 (branches): Focused searches on each major subtopic
- Depth 2 (leaves): Specific detail searches for evidence and examples
AGGREGATION:
When combining sub-results, don't just concatenate. Synthesize:
- Identify cross-cutting themes across subtopics
- Note where subtopics interact or conflict
- Highlight the most important insights at each level
- Build a coherent narrative from bottom-up findings
tools:
- name: explore_topic
type: rag-query
description: "Explore a specific topic or subtopic in the knowledge base"
parameters:
type: object
properties:
query:
type: string
description: "Focused query about the specific subtopic being explored"
top_k:
type: integer
default: 6
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
retriever: hybrid
alpha: 0.6
- name: find_examples
type: rag-query
description: "Find concrete examples, case studies, or implementations"
parameters:
type: object
properties:
query:
type: string
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
retriever: mmr
lambda: 0.5
fetch_k: 20
- name: find_definitions
type: rag-query
description: "Find precise definitions, specifications, or formal descriptions"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 3
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
retriever: simple
max_iterations: 8
22.6 Parallel Agent Consensus with RAG Verification¶
Multiple agents independently research the same question using RAG, then a verifier agent cross-references their answers against the source material to produce a consensus answer with confidence levels.
workflows:
- id: consensus-research
trigger:
uri: http-server://0.0.0.0:8080/api/verified-answer
steps:
- agent:
type: sequential
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: "Coordinate parallel research and verification."
sub_agents:
# Phase 1: Three independent researchers
- id: parallel-researchers
type: parallel
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: "Run three independent research perspectives."
sub_agents:
- id: researcher-a
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.3
system_prompt: |
You are Research Analyst A. Search for information and provide
a thorough, fact-based answer. Focus on the most authoritative
and specific sources. Always include exact quotes from sources.
Label each claim with [Source: document_name].
tools:
- name: search
type: rag-query
description: "Search knowledge base"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 8
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
retriever: hybrid
alpha: 0.7
max_iterations: 5
- id: researcher-b
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.5
system_prompt: |
You are Research Analyst B. Search for diverse perspectives
and alternative viewpoints. Look beyond the obvious answers.
Use MMR search to find diverse sources. Explicitly note where
sources agree and disagree. Label claims with [Source: name].
tools:
- name: search
type: rag-query
description: "Search for diverse perspectives"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 10
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
retriever: mmr
lambda: 0.3
fetch_k: 30
max_iterations: 5
- id: researcher-c
type: react
model:
provider: groq
model: llama-3.3-70b-versatile
api_key: "${env:GROQ_API_KEY}"
temperature: 0.2
system_prompt: |
You are Research Analyst C. Focus on finding quantitative
evidence, data points, and concrete examples. Prioritize
specificity over breadth. Every claim must be backed by
a specific data point. Label with [Source: name].
tools:
- name: search
type: rag-query
description: "Search for data and evidence"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 6
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
retriever: hybrid
alpha: 0.9
max_iterations: 5
# Phase 2: Cross-reference verifier
- id: verifier
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.1
max_tokens: 3000
system_prompt: |
You are a verification analyst. You receive three independent
research reports on the same question. Your job:
1. IDENTIFY CONSENSUS: Claims that appear in 2+ reports
2. IDENTIFY CONFLICTS: Claims where reports disagree
3. VERIFY KEY CLAIMS: Search the knowledge base to verify the
most important claims from each report
4. PRODUCE FINAL ANSWER with confidence levels:
- HIGH CONFIDENCE: Claimed by 2+ analysts AND verified in sources
- MEDIUM CONFIDENCE: Claimed by 1 analyst AND verified in sources
- LOW CONFIDENCE: Claimed but not directly verifiable
- DISPUTED: Analysts disagree on this point
Output format:
## Verified Answer
[Main answer with inline confidence markers]
## Evidence Summary
| Claim | Analysts | Verified | Confidence |
|-------|----------|----------|------------|
## Unresolved Questions
[What couldn't be determined from available sources]
tools:
- name: verify_claim
type: rag-query
description: "Search to verify a specific claim from the research reports"
parameters:
type: object
properties:
query:
type: string
description: "The specific claim to verify"
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
retriever: simple
max_iterations: 8
22.7 Adaptive RAG with Strategy Selection Agent¶
An agent that analyzes the incoming query, selects the optimal RAG retrieval strategy, chunking approach, and synthesis mode — then executes the customized pipeline.
workflows:
- id: adaptive-rag
trigger:
uri: http-server://0.0.0.0:8080/api/smart-query
steps:
# Step 1: Agent classifies query and selects RAG strategy
- agent:
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.1
response_format: json_object
system_prompt: |
Classify the user's query and select the optimal RAG strategy.
Return a JSON object with your selections:
{
"query_type": "factual|conceptual|comparative|procedural|analytical",
"complexity": "simple|moderate|complex",
"retrieval_strategy": "simple|hybrid|mmr|hyde|multi_query|sub_question",
"top_k": 3-15,
"synthesis_mode": "compact|refine|tree-summarize|accumulate",
"reasoning": "Brief explanation of why these settings are optimal"
}
Decision guide:
- Factual (who/what/when): hybrid, top_k=3-5, compact
- Conceptual (explain/describe): mmr for diversity, top_k=8, refine
- Comparative (vs/differ/compare): sub_question to decompose, top_k=10, accumulate
- Procedural (how to/steps): hybrid, top_k=5-8, refine (step-by-step)
- Analytical (why/impact/evaluate): hyde + mmr, top_k=10-15, tree-summarize
- Simple questions: fewer results, compact synthesis
- Complex questions: more results, iterative synthesis
max_iterations: 1
# Step 2: Route to strategy-specific pipeline based on classification
- choice:
when:
# Simple factual queries — fast path
- starlang: |
data = read(payload, "application/json")
data.get("complexity", "") == "simple"
steps:
- rag-retrieve:
strategy: hybrid
top_k: 3
alpha: 0.8
- rag-postprocess:
steps:
- type: similarity-filter
params:
min_score: "0.5"
- rag-synthesize:
mode: compact
model:
provider: groq
model: llama-3.3-70b-versatile
api_key: "${env:GROQ_API_KEY}"
# Comparative queries — decompose and search each aspect
- starlang: |
data = read(payload, "application/json")
data.get("query_type", "") == "comparative"
steps:
- rag-retrieve:
strategy: sub_question
top_k: 10
- rag-postprocess:
steps:
- type: dedup
- type: cohere-rerank
params:
api_key: "${env:COHERE_API_KEY}"
top_n: "8"
- rag-synthesize:
mode: accumulate
parallel: true
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
# Analytical queries — deep exploration
- starlang: |
data = read(payload, "application/json")
data.get("query_type", "") == "analytical"
steps:
- rag-retrieve:
strategy: hyde
top_k: 15
- rag-postprocess:
steps:
- type: cohere-rerank
params:
api_key: "${env:COHERE_API_KEY}"
top_n: "10"
- type: diversity
params:
threshold: "0.85"
- type: long-context-reorder
- rag-synthesize:
mode: tree-summarize
chunks_per_level: 4
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
# Default: balanced pipeline
otherwise:
steps:
- rag-retrieve:
strategy: hybrid
top_k: 8
alpha: 0.7
- rag-postprocess:
steps:
- type: similarity-filter
params:
min_score: "0.4"
- type: cohere-rerank
params:
api_key: "${env:COHERE_API_KEY}"
top_n: "5"
- rag-synthesize:
mode: refine
max_chunks: 8
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
# Step 3: Evaluate quality and retry if needed
- rag-eval:
metrics:
- faithfulness
- answer_relevancy
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
- choice:
when:
- starlang: |
faith = float(properties.get("rag_faithfulness", "1.0"))
relevancy = float(properties.get("rag_answer_relevancy", "1.0"))
faith < 0.6 or relevancy < 0.5
steps:
# Fallback: re-retrieve with broader strategy
- rag-retrieve:
strategy: multi_query
top_k: 15
- rag-postprocess:
steps:
- type: cohere-rerank
params:
api_key: "${env:COHERE_API_KEY}"
top_n: "8"
- rag-synthesize:
mode: refine
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
22.8 Knowledge Graph-Augmented Agent with Multi-Hop Reasoning¶
An agent that combines vector RAG with knowledge graph traversal for multi-hop reasoning — answering questions that require connecting information across multiple documents and relationships.
workflows:
# Build knowledge graph from documents
- id: build-knowledge-graph
trigger:
uri: timer://startup?repeatCount=1
steps:
- rag-ingest:
reader: directory
path: ./company-docs
recursive: true
- rag-chunk:
strategy: sentence
chunk_size: 400
overlap: 50
# Store chunks in vector store
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: company_docs
# Also build knowledge graph for relationships
- rag-graph-build:
store: in-memory
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
# Multi-hop reasoning endpoint
- id: multi-hop-qa
trigger:
uri: http-server://0.0.0.0:8080/api/reason
steps:
- agent:
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.3
max_tokens: 3000
system_prompt: |
You are a multi-hop reasoning agent. Many questions require connecting
information from multiple sources. Your strategy:
1. DECOMPOSE: Break the question into atomic information needs
2. SEARCH: Use vector search for each atomic need
3. TRAVERSE: Use graph queries to find relationships between entities
4. CHAIN: Follow relationship chains to connect facts
5. SYNTHESIZE: Combine vector results + graph traversals into a
coherent answer that shows your reasoning chain
Example multi-hop question: "What projects does the CTO's team work on?"
- Step 1: Search for "CTO" → find person name
- Step 2: Graph traverse from CTO → "manages" → teams
- Step 3: Graph traverse from teams → "works_on" → projects
- Step 4: Search for details on each discovered project
Always show your reasoning chain so the user understands HOW you
connected the information. Use format:
[Hop 1] Found: ... → [Hop 2] Found: ... → [Conclusion] ...
tools:
- name: vector_search
type: rag-query
description: "Search documents by semantic similarity. Best for finding specific facts, descriptions, or passages."
parameters:
type: object
properties:
query:
type: string
description: "What to search for"
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: company_docs
retriever: hybrid
alpha: 0.7
- name: graph_search_entity
type: rag-query
description: "Find entities (people, teams, projects, technologies) in the knowledge graph by name or description"
parameters:
type: object
properties:
query:
type: string
description: "Entity name or description to search for"
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_graph_entities
retriever: simple
- name: graph_traverse
type: rag-query
description: "Traverse relationships from an entity. Returns connected entities up to N hops away. Use for 'who works with', 'what depends on', 'who reports to' type questions."
parameters:
type: object
properties:
entity:
type: string
description: "Starting entity name"
relationship_type:
type: string
description: "Relationship to follow (e.g., manages, works_on, uses, depends_on, reports_to)"
hops:
type: integer
default: 2
description: "How many relationship hops to traverse (1-3)"
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_graph_relations
- name: graph_find_path
type: rag-query
description: "Find the shortest relationship path between two entities. Use to discover how two things are connected."
parameters:
type: object
properties:
from_entity:
type: string
to_entity:
type: string
max_hops:
type: integer
default: 3
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_graph_paths
max_iterations: 10
22.9 Multi-Collection Federated RAG with Agent Router¶
An agent that intelligently routes queries across multiple specialized knowledge bases, retrieves from the most relevant collections, and fuses results — handling cases where information spans multiple domains.
workflows:
# Ingestion pipelines per collection
- id: ingest-engineering-docs
trigger:
uri: cron://nightly?schedule=0 1 * * *
steps:
- rag-ingest:
reader: directory
path: ./docs/engineering
recursive: true
- rag-chunk:
strategy: code
language: go
chunk_size: 1200
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: engineering
- id: ingest-product-docs
trigger:
uri: cron://nightly?schedule=0 1 * * *
steps:
- rag-ingest:
reader: directory
path: ./docs/product
recursive: true
- rag-chunk:
strategy: markdown
retain_headers: true
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: product
- id: ingest-policy-docs
trigger:
uri: cron://nightly?schedule=0 1 * * *
steps:
- rag-ingest:
reader: directory
path: ./docs/policies
recursive: true
- rag-chunk:
strategy: semantic
threshold: 0.5
chunk_size: 800
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: policies
- id: ingest-support-tickets
trigger:
uri: cron://hourly?schedule=0 * * * *
steps:
- rag-ingest:
reader: json
- rag-chunk:
strategy: sentence
chunk_size: 500
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: support_tickets
# Federated query endpoint
- id: federated-search
trigger:
uri: http-server://0.0.0.0:8080/api/search
steps:
- agent:
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.2
system_prompt: |
You are a federated search agent with access to 4 knowledge bases:
1. **engineering**: API docs, architecture decisions, code documentation, runbooks
2. **product**: Product specs, feature requirements, roadmaps, release notes
3. **policies**: HR policies, security policies, compliance docs, SOPs
4. **support_tickets**: Past customer support issues, resolutions, known issues
ROUTING STRATEGY:
- Analyze the query to determine which collections are relevant
- For cross-domain questions, search multiple collections
- Start with the most likely collection, then broaden if results are insufficient
- For "how to" questions: check engineering first, then product
- For policy/process questions: check policies first
- For bug/issue questions: check support_tickets first, then engineering
- For feature questions: check product first
FUSION STRATEGY:
- When combining results from multiple collections, note the source
- If information conflicts between collections, prefer: policies > product > engineering > support
- Always indicate which collection each piece of information came from
Provide clear, actionable answers with source attribution.
tools:
- name: search_engineering
type: rag-query
description: "Search engineering docs: APIs, architecture, code docs, runbooks"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: engineering
retriever: hybrid
alpha: 0.7
- name: search_product
type: rag-query
description: "Search product docs: specs, features, roadmaps, release notes"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: product
retriever: hybrid
alpha: 0.6
- name: search_policies
type: rag-query
description: "Search policies: HR, security, compliance, SOPs"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: policies
retriever: simple
- name: search_support_tickets
type: rag-query
description: "Search past support tickets: issues, resolutions, workarounds"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 8
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: support_tickets
retriever: hybrid
alpha: 0.5
max_iterations: 8
22.10 Multimodal RAG Agent with Vision and Document Analysis¶
An agent that processes images, PDFs, and text together — using vision models for image understanding, OCR for scanned documents, and text RAG for written content — all unified in a single conversational interface.
workflows:
# Ingest text + image documents
- id: ingest-multimodal-corpus
trigger:
uri: timer://startup?repeatCount=1
steps:
- multicast:
uris:
- direct://ingest-text-docs
- direct://ingest-scanned-pdfs
- direct://ingest-images
# Branch 1: Text documents
- id: ingest-text-docs
trigger:
uri: direct://ingest-text-docs
steps:
- rag-ingest:
reader: directory
path: ./corpus/text
recursive: true
extensions:
- ".md"
- ".txt"
- ".html"
- rag-chunk:
strategy: semantic
threshold: 0.5
chunk_size: 800
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: text_docs
# Branch 2: Scanned PDFs (OCR)
- id: ingest-scanned-pdfs
trigger:
uri: direct://ingest-scanned-pdfs
steps:
- rag-ingest:
reader: pdf
config:
ocr_backend: docling
ocr_url: "http://localhost:5001"
output_format: markdown
extract_tables: true
- rag-chunk:
strategy: markdown
retain_headers: true
chunk_size: 1000
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: scanned_docs
# Branch 3: Images with CLIP embeddings
- id: ingest-images
trigger:
uri: direct://ingest-images
steps:
- rag-ingest:
reader: image
- rag-embed:
provider: multimodal
model: clip
- rag-store:
store: qdrant
connection: "http://localhost:6333"
collection: images
dimensions: 768
# Multimodal Q&A endpoint
- id: multimodal-qa
trigger:
uri: http-server://0.0.0.0:8080/api/multimodal-ask
steps:
- agent:
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.3
max_tokens: 2048
system_prompt: |
You are a multimodal research assistant that can analyze images,
search text documents, and query scanned PDFs.
CAPABILITIES:
- search_text: Find information in written documents
- search_scanned: Find information in OCR-processed scanned documents
- search_images: Find relevant images by description
- analyze_image: Describe and analyze an image the user uploads
STRATEGY:
1. If the user uploads an image, analyze it first with your vision
capabilities, then search for related text documents
2. For text questions, search both text_docs and scanned_docs
3. For image-related questions, search the image collection
4. Combine visual and textual evidence in your answer
5. When referencing images, describe what they show
6. When referencing scanned docs, note they are from scanned sources
Always distinguish between what you SEE in images and what you
READ in documents.
tools:
- name: search_text
type: rag-query
description: "Search text documents (markdown, txt, html)"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: text_docs
retriever: hybrid
alpha: 0.7
- name: search_scanned
type: rag-query
description: "Search OCR-processed scanned PDF documents"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: scanned_docs
retriever: hybrid
alpha: 0.6
- name: search_images
type: rag-query
description: "Search images by text description using CLIP embeddings"
parameters:
type: object
properties:
query:
type: string
description: "Text description of the image to find"
top_k:
type: integer
default: 3
config:
store: qdrant
connection: "http://localhost:6333"
collection: images
retriever: simple
max_iterations: 8
22.11 Agent-Driven Corrective RAG with Iterative Refinement¶
A loop agent that retrieves, evaluates retrieval quality, and iteratively refines its search strategy — switching between retrieval strategies and query reformulations until the answer meets quality thresholds.
workflows:
- id: corrective-rag-agent
trigger:
uri: http-server://0.0.0.0:8080/api/reliable-answer
steps:
- agent:
type: loop
max_iterations: 5
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.2
system_prompt: |
You are a corrective RAG agent. Your goal is to produce HIGH-QUALITY
answers by iteratively searching, evaluating, and refining.
ITERATION PROTOCOL:
1. SEARCH: Use the most appropriate search strategy for this iteration
- Iteration 1: Start with standard hybrid search
- If results seem off-topic: try rephrasing the query
- If results lack diversity: use diverse_search
- If results miss context: use hypothetical_search (HyDE)
2. EVALUATE: After each search, assess the results:
- Relevance: Are the retrieved documents on-topic? (score 1-10)
- Coverage: Do they cover all aspects of the question? (score 1-10)
- Specificity: Do they contain specific, actionable information? (score 1-10)
3. DECIDE:
- If all scores >= 7: Draft your answer and evaluate it
- If Relevance < 7: Reformulate the query, try different terms
- If Coverage < 7: Search for missing aspects specifically
- If Specificity < 7: Search with more precise, narrow queries
4. SELF-CHECK your draft answer:
- Is every claim supported by the retrieved documents?
- Are there any logical gaps?
- Would a reader find this complete and actionable?
5. If self-check passes: Output "COMPLETE" followed by your final answer
If self-check fails: Identify what's missing and search again
Track your iteration history to avoid repeating failed strategies.
tools:
- name: search_standard
type: rag-query
description: "Standard hybrid search (vector + keyword). Good default for most queries."
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
retriever: hybrid
alpha: 0.7
- name: search_diverse
type: rag-query
description: "Diversity-focused search. Use when standard search returns too-similar results."
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 8
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
retriever: mmr
lambda: 0.3
fetch_k: 25
- name: search_hypothetical
type: rag-query
description: "HyDE search. Generates a hypothetical answer first, then searches. Use for conceptual or abstract queries."
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
retriever: hyde
- name: search_precise
type: rag-query
description: "Precise keyword-weighted search. Use for specific terms, names, or IDs."
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 3
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
retriever: hybrid
alpha: 0.9
max_iterations: 8
# Final quality gate — reject if evaluation too low
- rag-eval:
metrics:
- faithfulness
- answer_relevancy
- context_relevancy
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
- choice:
when:
- starlang: |
faith = float(properties.get("rag_faithfulness", "0"))
faith < 0.5
steps:
- set-body:
starlang: |
write({
"answer": payload,
"confidence": "low",
"warning": "Answer may not be fully supported by available sources. Please verify independently."
}, "application/json")
otherwise:
steps:
- set-body:
starlang: |
write({
"answer": payload,
"confidence": "high",
"faithfulness": properties.get("rag_faithfulness", "N/A"),
"relevancy": properties.get("rag_answer_relevancy", "N/A")
}, "application/json")
22.12 Agent with MCP Tool Server and RAG Augmentation¶
An agent that consumes tools from external MCP servers (filesystem, database, web) while also having access to a RAG knowledge base — enabling it to both retrieve knowledge and take actions on external systems.
workflows:
- id: agentic-mcp-rag
trigger:
uri: http-server://0.0.0.0:8080/api/agent
steps:
- agent:
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.4
max_tokens: 4096
system_prompt: |
You are an intelligent operations agent with access to:
1. A knowledge base (RAG) for looking up documentation, procedures, and policies
2. Filesystem tools for reading and writing files
3. Database tools for querying and modifying data
4. GitHub tools for code and issue management
WORKFLOW:
- First, understand what the user needs
- If it's a knowledge question: search the knowledge base
- If it requires action: look up the procedure in the knowledge base FIRST,
then execute the action using the appropriate tools
- For complex tasks: plan the steps, validate each against documentation,
then execute sequentially
- Always confirm destructive operations with the user before proceeding
SAFETY RULES:
- Never delete or modify production data without explicit confirmation
- Always read files before modifying them
- Log all database modifications
- Follow procedures from the knowledge base exactly
memory:
enabled: true
type: composite
backends:
- type: sliding-window
window_size: 30
- type: sql-long-term
dsn: "${env:DATABASE_URL}"
session_id_header: "X-Session-ID"
tools:
# RAG tools
- name: search_docs
type: rag-query
description: "Search internal documentation, runbooks, and procedures"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: operations_docs
retriever: hybrid
alpha: 0.7
- name: search_incidents
type: rag-query
description: "Search past incident reports and post-mortems"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: incidents
retriever: hybrid
alpha: 0.5
# MCP filesystem tools
- name: filesystem
type: mcp
description: "Read, write, and list files on the server"
mcp_server:
url: "http://localhost:3001"
transport: http
headers:
Authorization: "Bearer ${env:MCP_FS_TOKEN}"
tools:
- read_file
- write_file
- list_directory
- search_files
# MCP database tools
- name: database
type: mcp
description: "Query and modify the application database"
mcp_server:
url: "http://localhost:3002"
transport: http
headers:
Authorization: "Bearer ${env:MCP_DB_TOKEN}"
tools:
- query
- execute
- list_tables
- describe_table
# MCP GitHub tools
- name: github
type: mcp
description: "Interact with GitHub repositories, issues, and PRs"
mcp_server:
command: "npx"
args:
- "@modelcontextprotocol/server-github"
transport: stdio
tools:
- search_code
- get_file_contents
- create_issue
- list_issues
- create_pull_request
max_iterations: 15
22.13 Streaming RAG Chat with AG-UI Frontend Protocol¶
A complete streaming RAG chat system that uses AG-UI protocol to stream responses to a CopilotKit frontend — with real-time token streaming, RAG context display, and interactive follow-up suggestions.
workflows:
# AG-UI streaming endpoint for frontends
- id: streaming-rag-chat
trigger:
uri: agui://0.0.0.0:8080/api/chat
name: rag-assistant
steps:
- agent:
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.4
max_tokens: 2048
system_prompt: |
You are a helpful knowledge assistant. Search the knowledge base to
answer user questions. Stream your response as you generate it.
For each answer:
1. Search for relevant information
2. Synthesize a clear, well-structured response
3. End with 2-3 suggested follow-up questions the user might ask
Format follow-up suggestions as:
**Suggested questions:**
- [question 1]
- [question 2]
- [question 3]
memory:
enabled: true
type: sliding-window
window_size: 20
tools:
- name: search
type: rag-query
description: "Search the knowledge base"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
retriever: hybrid
alpha: 0.7
max_iterations: 5
# A2A endpoint — expose the same agent to other agents
- id: a2a-rag-service
trigger:
uri: a2a-server://0.0.0.0:8081
steps:
- agent:
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.3
system_prompt: |
You are a knowledge base query service for other AI agents.
Search the knowledge base and return structured, factual responses.
Include source citations and confidence levels.
Return JSON format for easy parsing by other agents.
tools:
- name: search
type: rag-query
description: "Search the knowledge base"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 8
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_base
retriever: hybrid
alpha: 0.7
max_iterations: 5
# MCP endpoint — expose as MCP tools for Claude Desktop
- id: mcp-rag-tools
trigger:
uri: mcp-server://0.0.0.0:3000?transport=http&name=rag-knowledge-base
steps: []
helix:
mcp:
server:
enabled: true
transport: http
addr: "0.0.0.0:3000"
name: "rag-knowledge-base"
version: "1.0.0"
workflows:
- streaming-rag-chat
tools:
streaming-rag-chat:
description: "Search and answer questions from the knowledge base"
input_schema: '{"type":"object","properties":{"query":{"type":"string"}}}'
a2a:
servers:
- enabled: true
addr: "0.0.0.0:8081"
name: "knowledge-base-agent"
description: "RAG-powered knowledge base agent for answering questions"
version: "1.0.0"
agui:
servers:
- enabled: true
addr: "0.0.0.0:8080"
path: "/api/chat"
name: "rag-assistant"
22.14 RAG Pipeline with Continuous Quality Monitoring¶
A production RAG system with built-in quality monitoring that evaluates every response, logs metrics, alerts on degradation, and automatically adjusts retrieval parameters when quality drops.
workflows:
# Main RAG query workflow with quality monitoring
- id: monitored-rag-query
trigger:
uri: http-server://0.0.0.0:8080/api/ask
steps:
# Retrieve
- rag-retrieve:
strategy: hybrid
top_k: 10
alpha: 0.7
# Postprocess
- rag-postprocess:
steps:
- type: similarity-filter
params:
min_score: "0.3"
- type: cohere-rerank
params:
api_key: "${env:COHERE_API_KEY}"
top_n: "5"
- type: long-context-reorder
# Synthesize
- rag-synthesize:
mode: compact
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
# Evaluate EVERY response
- rag-eval:
metrics:
- faithfulness
- answer_relevancy
- context_relevancy
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
# Wire tap: send metrics to monitoring without blocking response
- wiretap: seda://rag-metrics-collector
# Quality gate: flag low-quality responses
- choice:
when:
- starlang: |
faith = float(properties.get("rag_faithfulness", "1"))
relevancy = float(properties.get("rag_answer_relevancy", "1"))
faith < 0.5 or relevancy < 0.4
steps:
# Low quality: add disclaimer
- set-header:
name: X-RAG-Quality
starlang: '"low"'
- set-header:
name: X-RAG-Faithfulness
starlang: 'properties.get("rag_faithfulness", "N/A")'
- transform:
starlang: |
answer = payload
write({
"answer": answer,
"quality": "low",
"disclaimer": "This answer may not be fully supported by available sources. Please verify.",
"metrics": {
"faithfulness": properties.get("rag_faithfulness", "N/A"),
"answer_relevancy": properties.get("rag_answer_relevancy", "N/A"),
"context_relevancy": properties.get("rag_context_relevancy", "N/A")
}
}, "application/json")
otherwise:
steps:
- set-header:
name: X-RAG-Quality
starlang: '"high"'
- transform:
starlang: |
answer = payload
write({
"answer": answer,
"quality": "high",
"metrics": {
"faithfulness": properties.get("rag_faithfulness", "N/A"),
"answer_relevancy": properties.get("rag_answer_relevancy", "N/A"),
"context_relevancy": properties.get("rag_context_relevancy", "N/A")
}
}, "application/json")
# Metrics collection and alerting
- id: rag-metrics-processor
trigger:
uri: seda://rag-metrics-collector?concurrentConsumers=2
steps:
# Log metrics for Prometheus/Grafana
- target: log://rag-metrics?level=info
# Check for quality degradation
- choice:
when:
- starlang: |
faith = float(properties.get("rag_faithfulness", "1"))
faith < 0.3
steps:
- set-body:
starlang: |
write({
"alert": "critical_quality_drop",
"metric": "faithfulness",
"value": properties.get("rag_faithfulness", "N/A"),
"threshold": "0.3",
"query": properties.get("rag_query", "unknown")
}, "application/json")
- target: slack://ai-ops?channel=rag-alerts&webhook=${env:SLACK_WEBHOOK}
# Periodic quality report (daily)
- id: daily-rag-quality-report
trigger:
uri: cron://daily-report?schedule=0 9 * * *
steps:
- agent:
type: react
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.2
system_prompt: |
You are a RAG quality analyst. Query the metrics database and
produce a daily quality report with:
1. Average faithfulness, relevancy, context relevancy scores
2. Trend analysis (improving/degrading/stable)
3. Worst-performing query categories
4. Recommendations for improvement (reindex, adjust parameters, etc.)
Format as a concise Slack-friendly report.
tools:
- name: query_metrics
type: connector
description: "Query RAG metrics from the monitoring database"
parameters:
type: object
properties:
time_range:
type: string
default: "24h"
metric:
type: string
enum:
- faithfulness
- answer_relevancy
- context_relevancy
- all
config:
endpoint: "http://localhost:9090/api/v1/query_range"
method: GET
max_iterations: 5
- target: slack://ai-ops?channel=rag-daily-report&webhook=${env:SLACK_WEBHOOK}
22.15 Multi-Tenant RAG Agent with Tenant-Isolated Knowledge Bases¶
A multi-tenant SaaS deployment where each tenant has its own isolated knowledge base, memory, and usage tracking — with per-tenant auth and rate limiting.
workflows:
# Per-tenant document ingestion
- id: tenant-ingest
trigger:
uri: http-server://0.0.0.0:8080/api/tenants/:tenant_id/ingest
security:
- type: jwt
jwks_url: "${env:JWKS_URL}"
claims:
tenant_id: "${header:tenant_id}"
steps:
- throttle:
rate: 100
per: minute
key:
starlang: 'headers["tenant_id"]'
- rag-ingest:
reader: auto
- rag-chunk:
strategy: recursive
chunk_size: 800
overlap: 100
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:DATABASE_URL}"
collection:
starlang: '"tenant_" ++ headers["tenant_id"]'
# Per-tenant RAG query
- id: tenant-query
trigger:
uri: http-server://0.0.0.0:8080/api/tenants/:tenant_id/ask
security:
- type: jwt
jwks_url: "${env:JWKS_URL}"
claims:
tenant_id: "${header:tenant_id}"
steps:
- throttle:
rate: 60
per: minute
key:
starlang: 'headers["tenant_id"]'
- agent:
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.3
system_prompt: |
You are a helpful assistant. Answer questions using ONLY the
tenant's knowledge base. Never reference information from
other tenants or your training data.
memory:
enabled: true
type: sql-long-term
dsn: "${env:DATABASE_URL}"
session_id_header: "X-Session-ID"
tools:
- name: search
type: rag-query
description: "Search the tenant's knowledge base"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection:
starlang: '"tenant_" ++ headers["tenant_id"]'
retriever: hybrid
alpha: 0.7
max_iterations: 5
# Track usage per tenant
- wiretap: seda://usage-tracker
- transform:
starlang: |
write({
"answer": payload,
"tenant": headers["tenant_id"],
"usage": {
"model": "gpt-4o",
"tokens_prompt": headers["HelixLlmPromptTokens"],
"tokens_completion": headers["HelixLlmCompletionTokens"]
}
}, "application/json")
# Usage tracking and billing
- id: usage-tracker
trigger:
uri: seda://usage-tracker?concurrentConsumers=4
steps:
- target: postgres://usage_db?table=api_usage
helix:
tenant:
enabled: true
header: "tenant_id"
required: true
audit:
enabled: true
buffer_size: 10000
shutdown:
timeout: 30s
drain_timeout: 15s
22.16 Agent Flow Pipeline with RAG-Augmented Stages¶
A pipeline flow where each stage is a specialized agent with its own RAG context — document extraction feeds into classification, which feeds into domain-specific analysis, and finally quality review.
workflows:
- id: document-analysis-pipeline
trigger:
uri: http-server://0.0.0.0:8080/api/analyze-document
steps:
- agent:
type: sequential
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: "Process documents through the analysis pipeline."
sub_agents:
# Stage 1: Extract and structure
- id: extractor
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.1
response_format: json_object
system_prompt: |
Extract structured information from the document. Return JSON:
{
"title": "...",
"type": "contract|invoice|report|correspondence|legal|technical|other",
"date": "YYYY-MM-DD or null",
"parties": ["..."],
"key_terms": ["..."],
"summary": "2-3 sentence summary",
"raw_text": "full extracted text"
}
max_iterations: 1
# Stage 2: Classify and route
- id: classifier
type: react
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.0
response_format: json_object
system_prompt: |
Given the extracted document data, classify it and determine
the analysis template to use. Search the classification rules
knowledge base for guidance. Return JSON:
{
"category": "...",
"subcategory": "...",
"risk_level": "low|medium|high|critical",
"analysis_template": "...",
"priority_areas": ["..."]
}
tools:
- name: lookup_rules
type: rag-query
description: "Look up document classification rules and analysis templates"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 3
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: classification_rules
retriever: simple
max_iterations: 3
# Stage 3: Domain-specific deep analysis
- id: analyst
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.3
max_tokens: 4096
system_prompt: |
Perform deep analysis of the classified document. Use the
domain knowledge base to:
1. Compare against standard templates and requirements
2. Identify deviations, risks, or notable clauses
3. Cross-reference with regulatory requirements
4. Assess completeness against checklist
Produce a structured analysis report with findings categorized as:
- COMPLIANT: Meets all requirements
- DEVIATION: Differs from standard but may be acceptable
- RISK: Potential issue requiring attention
- MISSING: Required element not found
- FLAG: Unusual clause requiring human review
tools:
- name: search_domain_knowledge
type: rag-query
description: "Search domain-specific knowledge (regulations, templates, best practices)"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 8
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: domain_knowledge
retriever: hybrid
alpha: 0.6
- name: search_precedents
type: rag-query
description: "Search past analyzed documents for similar cases and precedents"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: analyzed_documents
retriever: mmr
lambda: 0.5
max_iterations: 10
# Stage 4: Quality review
- id: reviewer
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.5
system_prompt: |
Review the analysis report for quality and completeness.
Check against the review checklist from the knowledge base.
Produce a final report with:
1. Executive summary (3-5 sentences)
2. Key findings (prioritized by risk level)
3. Recommended actions
4. Confidence assessment (how reliable is this analysis?)
5. Areas requiring human expert review
tools:
- name: lookup_checklist
type: rag-query
description: "Look up quality review checklists"
parameters:
type: object
properties:
query:
type: string
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: review_checklists
retriever: simple
max_iterations: 4
22.17 Supervisor Agent with Dynamic Sub-Agent Selection and RAG Context Injection¶
A supervisor that dynamically selects which sub-agents to invoke based on query analysis, injects relevant RAG context into each sub-agent's system prompt, and aggregates their responses with source deduplication.
workflows:
- id: dynamic-supervisor
trigger:
uri: http-server://0.0.0.0:8080/api/expert
steps:
- agent:
type: supervisor
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.3
system_prompt: |
You are an expert coordinator managing a team of specialists.
Each specialist has access to their own domain knowledge base.
AVAILABLE SPECIALISTS:
- technical_expert: Software architecture, APIs, infrastructure, DevOps
- business_analyst: Strategy, market analysis, business processes, metrics
- compliance_officer: Regulations, policies, audit requirements, risk assessment
- data_scientist: Analytics, ML/AI, data pipelines, statistical methods
DELEGATION RULES:
1. Analyze the query to determine which expertise is needed
2. For single-domain questions: delegate to one specialist
3. For cross-domain questions: delegate to multiple specialists
and synthesize their responses
4. For ambiguous questions: start with the most likely specialist,
then consult others if the answer is incomplete
5. Always delegate — never answer directly without consulting a specialist
When synthesizing multi-specialist responses:
- Resolve any contradictions (prefer compliance > technical > business)
- Deduplicate information
- Present a unified answer organized by topic, not by specialist
sub_agents:
- id: technical_expert
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.2
system_prompt: |
You are a senior technical expert. Search the engineering
knowledge base to provide accurate, detailed technical answers.
Include code examples, architecture patterns, and best practices
when relevant. Always cite your sources.
tools:
- name: search_engineering
type: rag-query
description: "Search engineering documentation"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 8
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: engineering_docs
retriever: hybrid
alpha: 0.7
max_iterations: 6
- id: business_analyst
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.4
system_prompt: |
You are a business analyst. Search the business knowledge
base for market data, strategies, processes, and metrics.
Provide data-driven insights with clear business impact analysis.
tools:
- name: search_business
type: rag-query
description: "Search business documentation and market data"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 6
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: business_docs
retriever: hybrid
alpha: 0.6
max_iterations: 5
- id: compliance_officer
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.1
system_prompt: |
You are a compliance officer. Search regulatory and policy
documents. Be precise about requirements — distinguish between
"MUST" (mandatory) and "SHOULD" (recommended). Always cite
the specific regulation, section, and version.
tools:
- name: search_regulations
type: rag-query
description: "Search regulatory and compliance documents"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: compliance_docs
retriever: simple
max_iterations: 5
- id: data_scientist
type: react
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.3
system_prompt: |
You are a data scientist. Search the analytics knowledge base
for data methodology, ML patterns, and statistical approaches.
Provide technically rigorous answers with methodology citations.
tools:
- name: search_data_science
type: rag-query
description: "Search data science and analytics documentation"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 6
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: data_science_docs
retriever: mmr
lambda: 0.5
max_iterations: 5
22.18 Full-Stack AI Application: Ingestion, Indexing, Query, Evaluation, and Monitoring¶
A complete production-grade RAG application with all lifecycle stages: scheduled ingestion with incremental updates, multi-strategy retrieval, agent-powered synthesis, quality evaluation, caching, monitoring, and alerting.
workflows:
# ============================================================
# INGESTION LAYER — Incremental document processing
# ============================================================
# Full reindex (weekly)
- id: full-reindex
trigger:
uri: cron://weekly?schedule=0 0 * * 0
steps:
- rag-ingest:
reader: directory
path: ./knowledge-base
recursive: true
extensions:
- ".md"
- ".pdf"
- ".html"
- ".txt"
- ".docx"
- rag-chunk:
strategy: semantic
threshold: 0.45
chunk_size: 1000
overlap: 150
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_v2
dimensions: 1536
# Extract metadata for filtering
- rag-extract:
extractors:
- title
- keywords
- summary
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
# Build knowledge graph
- rag-graph-build:
store: in-memory
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
- target: log://ingestion?level=info
# Incremental updates (hourly — new/modified files only)
- id: incremental-update
trigger:
uri: cron://hourly?schedule=0 * * * *
steps:
- rag-ingest:
reader: directory
path: ./knowledge-base
recursive: true
- filter:
starlang: |
import "time"
doc_time = headers["HelixFileLastModified"]
doc_time > time.now() - time.hour
- rag-chunk:
strategy: semantic
threshold: 0.45
chunk_size: 1000
overlap: 150
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_v2
# ============================================================
# QUERY LAYER — Agent-powered RAG with caching
# ============================================================
- id: production-rag-query
trigger:
uri: http-server://0.0.0.0:8080/api/ask
security:
- type: jwt
jwks_url: "${env:JWKS_URL}"
steps:
# Rate limit per user
- throttle:
rate: 30
per: minute
key:
starlang: 'headers["Authorization"]'
# Check cache first
- rag-cache:
enabled: true
max_size: 50000
# Agent-driven retrieval and synthesis
- agent:
type: react
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
temperature: 0.3
max_tokens: 2048
system_prompt: |
You are a production knowledge assistant. For each question:
1. Search the knowledge base with a well-crafted query
2. If results are insufficient, try alternative queries
3. For complex questions, decompose and search each part
4. Synthesize a clear, sourced answer
5. Rate your confidence: HIGH (multiple supporting sources),
MEDIUM (some supporting evidence), LOW (limited information)
Always cite sources. If you cannot find sufficient information,
say so clearly rather than speculating.
memory:
enabled: true
type: composite
backends:
- type: sliding-window
window_size: 15
- type: sql-long-term
dsn: "${env:DATABASE_URL}"
session_id_header: "X-Session-ID"
tools:
- name: search
type: rag-query
description: "Search the knowledge base"
parameters:
type: object
properties:
query:
type: string
top_k:
type: integer
default: 5
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_v2
retriever: hybrid
alpha: 0.7
- name: search_diverse
type: rag-query
description: "Search with diversity for broader coverage"
parameters:
type: object
properties:
query:
type: string
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_v2
retriever: mmr
lambda: 0.4
fetch_k: 20
- name: graph_query
type: rag-query
description: "Query knowledge graph for entity relationships"
parameters:
type: object
properties:
query:
type: string
config:
store: pgvector
connection: "${env:DATABASE_URL}"
collection: knowledge_graph_entities
max_iterations: 6
# ============================================================
# EVALUATION LAYER — Quality assessment
# ============================================================
- rag-eval:
metrics:
- faithfulness
- answer_relevancy
- context_relevancy
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
# Send metrics to monitoring (non-blocking)
- wiretap: seda://rag-monitoring
# Format response with metadata
- transform:
starlang: |
answer = payload
write({
"answer": answer,
"quality": {
"faithfulness": properties.get("rag_faithfulness", "N/A"),
"relevancy": properties.get("rag_answer_relevancy", "N/A"),
"context_relevancy": properties.get("rag_context_relevancy", "N/A")
},
"session_id": headers["X-Session-ID"]
}, "application/json")
# ============================================================
# MONITORING LAYER — Metrics, alerts, reports
# ============================================================
- id: rag-monitoring
trigger:
uri: seda://rag-monitoring?concurrentConsumers=4
steps:
# Log for Prometheus scraping
- target: log://rag-quality-metrics?level=info
# Alert on critical quality drops
- choice:
when:
- starlang: |
faith = float(properties.get("rag_faithfulness", "1"))
faith < 0.3
steps:
- set-body:
starlang: |
write({
"level": "critical",
"alert": "RAG faithfulness below 0.3",
"value": properties.get("rag_faithfulness", "N/A"),
"query_preview": payload[:200]
}, "application/json")
- target: slack://ai-ops?channel=rag-alerts&webhook=${env:SLACK_WEBHOOK}
# Daily quality digest
- id: daily-quality-digest
trigger:
uri: cron://daily-digest?schedule=0 9 * * 1-5
steps:
- agent:
type: react
model:
provider: openai
model: gpt-4o-mini
api_key: "${env:OPENAI_API_KEY}"
temperature: 0.1
system_prompt: |
Generate a daily RAG quality report. Query metrics and summarize:
- Query volume (last 24h)
- Average quality scores (faithfulness, relevancy)
- Quality trend (improving/stable/degrading)
- Top 5 worst-performing queries
- Recommendations for improvement
tools:
- name: query_metrics
type: connector
description: "Query Prometheus for RAG metrics"
parameters:
type: object
properties:
query:
type: string
config:
endpoint: "http://prometheus:9090/api/v1/query"
method: GET
max_iterations: 3
- target: slack://ai-ops?channel=rag-daily-report&webhook=${env:SLACK_WEBHOOK}
helix:
shutdown:
timeout: 30s
drain_timeout: 15s
audit:
enabled: true
buffer_size: 10000
logging:
level: info
format: json
Strategy Selection Guides¶
Agent Type Selection¶
| Use Case | Agent Type | Why |
|---|---|---|
| Simple Q&A with tools | ReAct | Standard reasoning + tool loop |
| Customer routing | Supervisor | Routes to specialist sub-agents |
| Research tasks | Plan-Execute | Creates plan first, then executes |
| Complex analysis | Deep | Recursive decomposition |
| Content pipeline | Sequential | Refinement chain |
| Multi-perspective | Parallel | Independent parallel analysis |
| Quality refinement | Loop | Iterate until quality threshold |
Memory Backend Selection¶
| Use Case | Memory Type | Why |
|---|---|---|
| Simple chatbot | Sliding Window | Bounded, predictable cost |
| Long sessions | Summary | Auto-compresses old context |
| Cost control | Token Window | Strict token budget |
| Persistent sessions | SQL Long-Term | Survives restarts |
| Semantic recall | Vector | Find relevant past conversations |
| Production apps | Composite | Best of all worlds |
Retrieval Strategy Selection¶
| Use Case | Strategy | Why |
|---|---|---|
| General search | Hybrid | Best balance of precision and recall |
| Diverse results | MMR | Avoids near-duplicate chunks |
| Conceptual queries | HyDE | Bridges vocabulary gap |
| Broad recall | MultiQuery | Multiple query perspectives |
| Domain-specific | StepBack | Abstract concept matching |
| Complex questions | SubQuestion | Decomposes into sub-queries |
| Multiple sources | Fusion (RRF) | Combines retriever results |
| Hierarchical docs | Auto-Merging | Returns full parent context |
Synthesis Mode Selection¶
| Use Case | Mode | Cost | Quality | Speed |
|---|---|---|---|---|
| Quick answers | Compact | Low | Good | Fast |
| Complex questions | Refine | High | Best | Slow |
| Many documents | Tree Summarize | Medium | Good | Medium |
| Parallel processing | Accumulate | Medium | Good | Fast |
| No generation needed | Simple | None | N/A | Instant |
Chunking Strategy Selection¶
| Content Type | Strategy | Why |
|---|---|---|
| Mixed prose | Recursive | Tries multiple separator levels |
| Source code | Code | Respects function/class boundaries |
| Markdown docs | Markdown | Respects heading hierarchy |
| HTML pages | HTML | Respects block elements |
| Technical docs | Semantic | AI-detected topic boundaries |
| Dense retrieval | Sliding Window | Constant overlap for recall |
| Auto-merging RAG | Hierarchical | Parent-child chunk relationships |
| Chat/dialogue | Sentence | Natural conversation breaks |
| Token budgets | Token | Precise token-level control |
23. memx Vector Database Integration¶
23.1 RAG Pipeline with memx (Ingest + Search)¶
End-to-end RAG pipeline using memx as the vector store. memx provides both ES-compatible HTTP API (38% faster for vector ops) and pgwire SQL interface.
Auto-discovery: When connecting to an existing memx table, column names are auto-detected from table indexes (HNSW/IVF/DiskANN for embeddings, FULLTEXT for content). No id_column, embedding_column, or content_column configuration is needed. Column overrides remain available as optional escape hatches.
database: field: Both rag-store and rag-retrieve now support a database: field for targeting a specific memx database.
Raw results: memx retrieve returns raw []map[string]any -- all columns from the table with embedding excluded and _score included. No forced Chunk wrapping -- agents receive raw JSON.
pipelines:
- id: memx-ingest
source: http-server://0.0.0.0:8080/ingest
steps:
- rag-ingest:
reader: directory
path: ./knowledge
recursive: true
- rag-chunk:
strategy: recursive
chunk_size: 512
overlap: 64
separators: ["\n\n", "\n", ". ", " "]
- rag-embed:
provider: openai
model: text-embedding-3-small
api_key: "${env:OPENAI_API_KEY}"
- rag-store:
store: memx
connection: "http://localhost:8080"
database: knowledge_db
collection: knowledge_base
- id: knowledge-search-pipeline
steps:
- rag-retrieve:
strategy: memx_hybrid
store: memx
connection: "http://localhost:8080"
database: knowledge_db
collection: knowledge_base
top_k: 5
agents:
- id: knowledge-search
type: react
trigger: http-server://0.0.0.0:8080/ask
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
You are a knowledge assistant. Search the knowledge base
to find relevant information before answering.
tools:
- name: search
type: tool
tool_id: knowledge-search-pipeline
description: "Search the knowledge base"
parameters:
type: object
properties:
query:
type: string
description: "Search query"
23.2 Agent with memx Vector Memory¶
An agent that uses memx as its long-term vector memory backend, enabling semantic recall of past conversations.
agents:
- id: memx-memory-agent
type: react
trigger: http-server://0.0.0.0:8080/chat
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are a personal assistant with long-term memory.
You remember past conversations and can recall relevant
context from previous sessions.
memory:
enabled: true
type: composite
composite:
backends:
- type: sliding-window
window_size: 20
- type: vector
vector:
store: memx
connection: "http://localhost:8080"
collection: agent_memory
embed_provider: openai
embed_model: text-embedding-3-small
embed_api_key: "${env:OPENAI_API_KEY}"
top_k: 5
23.3 Direct SQL Queries via memx:// Component¶
Use the memx:// component for direct SQL queries against the pgwire interface. This is useful for complex joins, aggregations, and metadata-heavy queries that benefit from SQL expressiveness.
workflows:
- id: memx-sql-query
trigger:
uri: http-server://0.0.0.0:8080/api/analytics
steps:
- target: >-
memx://query?connection=postgres://localhost:5433/memx
&query=SELECT doc_id, metadata->>'source' as source, COUNT(*) as chunks
FROM knowledge_base GROUP BY doc_id, metadata->>'source'
ORDER BY chunks DESC LIMIT 20
- transform:
starlang: |
data = read(payload, "application/json")
result = write({"analytics": data, "count": sizeOf(data)}, "application/json")
- id: memx-metadata-search
trigger:
uri: http-server://0.0.0.0:8080/api/by-source
steps:
- target: >-
memx://query?connection=postgres://localhost:5433/memx
&query=SELECT content, metadata FROM knowledge_base
WHERE metadata->>'source' = :source
ORDER BY created_at DESC LIMIT 10
- log:
message: "Metadata query completed"
23.4 Hybrid Search (Dense + BM25 Fusion)¶
Combine dense vector similarity with BM25 keyword scoring for best-of-both-worlds retrieval. The alpha parameter controls the blend (1.0 = pure vector, 0.0 = pure BM25).
agents:
- id: hybrid-search-agent
type: react
trigger: http-server://0.0.0.0:8080/search
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: |
You are a search assistant. Use hybrid search for general queries
and keyword search for exact term lookups.
tools:
- name: hybrid_search
type: rag-query
description: "Hybrid search combining semantic similarity and keyword matching"
parameters:
type: object
properties:
query:
type: string
description: "Natural language search query"
alpha:
type: number
default: 0.7
description: "Balance between vector (1.0) and keyword (0.0) scoring"
top_k:
type: integer
default: 10
config:
store: memx
connection: "http://localhost:8080"
collection: knowledge_base
retriever: fusion
fusion_method: rrf
retrievers:
- type: simple
weight: 0.7
- type: bm25
weight: 0.3
- name: keyword_search
type: rag-query
description: "Exact keyword search for specific terms, IDs, or codes"
parameters:
type: object
properties:
query:
type: string
description: "Keyword query (exact matching)"
top_k:
type: integer
default: 5
config:
store: memx
connection: "http://localhost:8080"
collection: knowledge_base
retriever: bm25
23.5 E2E Agent + memx Pipeline (Verified Working)¶
A verified end-to-end configuration where a ReAct agent uses a memx-backed pipeline as a tool, exposed via AG-UI for frontend integration. Auto-discovery eliminates column configuration -- memx detects embedding and content columns from table indexes.
pipelines:
- id: product-search
steps:
- rag-retrieve:
strategy: memx
store: memx
connection: http://localhost:8080
database: products_db
collection: products
top_k: 3
agents:
- id: product-agent
type: react
model: { provider: openai, model: gpt-4o, api_key: ${OPENAI_API_KEY} }
system_prompt: Use the search tool to find products.
tools:
- name: search-products
type: tool
tool_id: product-search
description: Search the product catalog
parameters:
type: object
properties:
query: { type: string, description: What to search for }
trigger: agui://0.0.0.0:4500/products
The agent receives raw JSON results with all table columns (embedding excluded, _score included). No Chunk wrapping -- data flows as []map[string]any. Use Starlang transforms downstream if reshaping is needed.
24. Code Agent with Built-in Functions¶
24.1 Full-Featured Code Assistant¶
A code agent with all 14 built-in functions and SmartMemory for long coding sessions.
agents:
- id: code-assistant
type: react
trigger: http-server://0.0.0.0:8080/chat
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are an expert software engineer. You can read, write, search, and
edit files in the project. Always read a file before editing it.
Run tests after making changes. Use grep to find relevant code.
memory:
enabled: true
type: smart
max_tokens: 100000
tools:
- name: read-file
type: function
function: fs.read
description: Read a file from disk
- name: write-file
type: function
function: fs.write
description: Write content to a file
- name: edit-file
type: function
function: fs.edit
description: Apply a targeted string replacement to a file
- name: list-dir
type: function
function: fs.list
description: List files in a directory
- name: find-files
type: function
function: fs.glob
description: Find files matching a glob pattern
- name: search-code
type: function
function: fs.grep
description: Search file contents with regex
- name: mkdir
type: function
function: fs.mkdir
description: Create a directory
- name: delete
type: function
function: fs.delete
description: Delete a file or empty directory
- name: run-command
type: function
function: shell.exec
description: Execute a shell command
- name: fetch-url
type: function
function: web.fetch
description: Fetch a web page as plain text
- name: parse-json
type: function
function: data.json_parse
description: Parse a JSON string with optional path extraction
- name: format-json
type: function
function: data.json_format
description: Pretty-print a JSON value
- name: parse-csv
type: function
function: data.csv_parse
description: Parse CSV text into objects
- name: calc
type: function
function: math.eval
description: Evaluate a math expression
Run interactively:
Or as an HTTP server:
helix run code-agent.yaml
curl -X POST http://localhost:8080/chat -d '{"message": "Find all TODO comments in the codebase"}'
24.2 Minimal Code Agent (Read + Edit + Run)¶
A minimal code agent with just the essential tools.
agents:
- id: mini-coder
type: react
trigger: http-server://0.0.0.0:8080/code
model:
provider: openai
model: gpt-4o
api_key: "${env:OPENAI_API_KEY}"
system_prompt: "You are a coding assistant. Read files before editing."
memory:
enabled: true
type: smart
max_tokens: 50000
tools:
- name: read
type: function
function: fs.read
description: Read a file
- name: edit
type: function
function: fs.edit
description: Edit a file
- name: search
type: function
function: fs.grep
description: Search files
- name: run
type: function
function: shell.exec
description: Run a command
25. Browser Automation Agent¶
25.1 Web Research Agent with Playwright¶
An agent that can browse the web, navigate pages, and extract information using Playwright MCP.
agents:
- id: web-researcher
type: react
trigger: http-server://0.0.0.0:8080/research
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are a web research agent. You can browse websites, fill out forms,
click buttons, and take screenshots. Use browser_snapshot for fast
page reading. Use browser_take_screenshot only when visual context matters.
Always start by navigating to a URL with browser_navigate.
memory:
enabled: true
type: smart
max_tokens: 100000
tools:
- name: browser
type: mcp
description: "Browse the web with Playwright"
mcp_server:
command: npx
args: ["@playwright/mcp"]
transport: stdio
Run interactively:
25.2 Combined Code + Browser Agent¶
An agent that can both code and browse -- useful for looking up documentation, checking API specs, and implementing code.
agents:
- id: dev-agent
type: react
trigger: http-server://0.0.0.0:8080/dev
model:
provider: anthropic
model: claude-sonnet-4-20250514
api_key: "${env:ANTHROPIC_API_KEY}"
system_prompt: |
You are a development assistant. You can read and edit code files,
run commands, and browse the web for documentation. When implementing
a feature, look up relevant docs first, then write the code.
memory:
enabled: true
type: smart
max_tokens: 100000
tools:
- name: read-file
type: function
function: fs.read
description: Read a file
- name: edit-file
type: function
function: fs.edit
description: Edit a file
- name: search
type: function
function: fs.grep
description: Search files
- name: run
type: function
function: shell.exec
description: Run a shell command
- name: find
type: function
function: fs.glob
description: Find files by pattern
- name: browser
type: mcp
description: "Browse the web"
mcp_server:
command: npx
args: ["@playwright/mcp"]
transport: stdio
- name: fetch
type: function
function: web.fetch
description: Quick-fetch a URL as text (no JS)
This document covers all Helix framework capabilities with production-ready YAML examples. For API reference and Go fluent builder examples, see the User Guide. For detailed agent internals, see the AI Agents Guide. For RAG pipeline details, see the RAG Guide.