Building a RAG-powered AI Assistant with Julep

julep-assistant/
├── agent.yaml              # Agent configuration
├── task/                   # Julep task definitions
│   ├── main.yaml          # Main workflow task
│   ├── crawl.yaml         # Web crawling sub-task
│   └── full_task.yaml     # Complete task with all steps
├── scripts/               # Utility scripts
│   ├── crawler.py         # Standalone web crawler
│   └── indexer.py         # Document indexing utility
├── chainlit-ui/           # Web interface
│   ├── app.py            # Main Chainlit application
│   ├── feedback/         # Feedback handling system
│   └── requirements.txt  # Python dependencies
└── julep-assistant-notebook.ipynb  # Interactive notebook demo

name: Julep Support Assistant
about: >-
  You are the official Julep AI support assistant. You help developers 
  understand and use the Julep platform effectively.
  
model: claude-sonnet-4

instructions: |-
  You are the official Julep AI support assistant. Your purpose is to help 
  developers build AI applications using the Julep platform.
  
  Your core responsibilities:
  1. **Workflow Assistance**: Help users write, debug, and optimize Julep workflows
  2. **Concept Explanation**: Clearly explain Julep concepts like agents, tasks, sessions, and tools
  3. **Code Examples**: Provide working code examples in Python, YAML, or JavaScript
  4. **API Guidance**: Help users understand and use Julep's API effectively
  5. **Best Practices**: Share proven patterns and architectural recommendations
  
  Guidelines:
  - Always provide accurate, up-to-date information from the official documentation
  - Include code examples whenever possible
  - Use proper syntax highlighting for code blocks
  - Explain the "why" behind recommendations, not just the "how"

name: Julep Documentation Crawler Task
description: A Julep agent that can crawl the Julep documentation website and store the content in the document store with proper contextualization.

input_schema:
  type: object
  properties:
    url:
      type: string
      description: "The URL of the documentation page"
  required:
    - url

tools:
- name: spider_crawler
  type: integration
  integration:
    provider: spider
    setup:
      spider_api_key: {spider_api_key}

main:
- tool: spider_crawler
  arguments:
    url: $ _['url']
    params:
      request: smart_mode
      return_format: markdown
      proxy_enabled: $ True
      filter_output_images: $ True
      filter_output_svg: $ True
      readability: $ True

main:
# Step 0: Crawl the Julep documentation using Spider (will crawl multiple pages)
- tool: spider_crawler
  arguments:
    url: $ _['url']
    params:
      request: smart_mode
      limit: 2
      return_format: markdown
      proxy_enabled: $ True
      filter_output_images: $ True
      filter_output_svg: $ True
      readability: $ True

# Step 1: Process each crawled page individually
- over: $ [page for page in _.result if page.status == 200 and page.content]
  parallelism: 5
  map:
    workflow: process_single_page
    arguments:
      url: $ _.url
      content: $ _.content

input_schema:
  type: object
  properties:
    url:
      type: string
      description: "The URL of the documentation page"
    content:
      type: string
      description: "The markdown content of the documentation page"
  required:
    - url
    - content

@cl.on_chat_start
async def on_chat_start():
    """Initialize a new chat session"""
    
    # Create session with RAG search options
    session = await julep_client.sessions.create(
        agent=AGENT_UUID,
        recall_options={
            "mode": "hybrid",      # Uses both vector and text search
            "confidence": 0.7,     # Confidence threshold
            "limit": 10,          # Max number of results
            "embed_text": True    # Embed query for vector search
        }
    )
    
    # Store session for later use
    cl.user_session.set("session", session)

@cl.on_message
async def on_message(message: cl.Message):
    """Handle incoming user messages"""
    
    # Get the session
    session = cl.user_session.get("session")
    
    # Send message to Julep and stream response
    msg = cl.Message(content="")
    
    response = await julep_client.sessions.chat(
        session_id=session.id,
        message={
            "role": "user",
            "content": message.content
        },
        stream=True
    )
    
    # Stream tokens to user
    async for chunk in response:
        if chunk.choices[0].delta.content:
            await msg.stream_token(chunk.choices[0].delta.content)
    
    await msg.send()

recall_options={
    "mode": "hybrid",      # Combines vector and text search
    "confidence": 0.7,     # Minimum confidence score
    "limit": 10,          # Maximum results to retrieve
    "embed_text": True    # Enable embeddings for vector search
}

class FeedbackHandler:
    async def process_feedback(
        self, 
        feedback_text: str, 
        user_question: str, 
        agent_response: str,
        session_id: str
    ) -> Dict[str, Any]:
        """Process user feedback and update agent instructions if valid"""
        
        # Get current agent details
        agent = await self.client.agents.get(agent_id=self.agent_id)
        
        # Validate the feedback using AI
        validation_result = await self.validator.validate_feedback(
            feedback_text=feedback_text,
            user_question=user_question,
            agent_response=agent_response,
            agent_instructions=current_instructions_str
        )
        
        # If feedback is valid with high confidence (>= 0.7)
        if validation_result.get("is_valid") and validation_result.get("confidence", 0) >= 0.7:
            updated_instructions = validation_result.get("updated_instructions")
            
            if updated_instructions:
                # Update the agent with new instructions
                await self.client.agents.create_or_update(
                    agent_id=self.agent_id,
                    name=agent.name,
                    instructions=updated_instructions, 
                )

class FeedbackValidator:
    async def validate_feedback(self, feedback_text, user_question, agent_response, agent_instructions):
        """Validate feedback using AI to ensure it's constructive and applicable"""
        
        # AI validates if feedback is:
        # 1. Constructive and specific
        # 2. Relevant to improving the agent
        # 3. Not contradicting core functionality
        # 4. Actionable for instruction updates
        
        # Returns:
        # - is_valid: boolean
        # - confidence: 0-1 score
        # - category: type of feedback
        # - updated_instructions: new instructions if applicable

def create_feedback_actions(self, message_id: str) -> list:
    """Create Chainlit actions for feedback collection"""
    return [
        cl.Action(
            name="feedback_helpful",
            payload={"value": "helpful"},
            label="👍 Helpful"
        ),
        cl.Action(
            name="feedback_not_helpful",
            payload={"value": "not_helpful"}, 
            label="👎 Not Helpful"
        ),
        cl.Action(
            name="feedback_detailed",
            payload={"value": "detailed"},
            label="💭 Give Detailed Feedback"
        )
    ]

# Original instruction
"Provide code examples when explaining concepts"

# After user feedback: "The code examples are too basic"
"Provide comprehensive code examples with edge cases and best practices when explaining concepts"

cd julep-assistant
pip install -r chainlit-ui/requirements.txt

# Create .env file
JULEP_API_KEY=your_julep_api_key_here
AGENT_UUID=your_agent_uuid  # Or use the default
SPIDER_API_KEY=your_spider_api_key  # Optional

cd chainlit-ui
chainlit run app.py

python scripts/crawler.py --url https://docs.julep.ai --max-pages 100

python scripts/indexer.py

# Execute the full crawl + index task
execution = await julep_client.tasks.execute(
    task_id=TASK_ID,
    input={
        "url": "https://docs.julep.ai",
        "max_pages": 100
    }
)

# Monitor execution status
while execution.status in ["pending", "running"]:
    execution = await julep_client.executions.get(execution.id)
    print(f"Status: {execution.status}")
    
    # Get execution steps for detailed progress
    steps = await julep_client.executions.steps.list(execution.id)
    for step in steps:
        print(f"  Step {step.name}: {step.status}")
    
    await asyncio.sleep(5)