CLARISSA Tutorial 04: LLM Conversation Layer¶

Learning Objectives:

• Design effective prompts for reservoir simulation domain
• Manage multi-turn conversations with context
• Classify user intents and extract entities
• Optimize context window usage

Prerequisites: Notebooks 01-03 (ECLIPSE fundamentals, OPM Flow, Knowledge Layer)

Estimated Time: 60 minutes

In [ ]:

# Colab Setup — API Keys & Dependencies
import sys, os
IN_COLAB = 'google.colab' in sys.modules

if IN_COLAB:
    !pip install -q anthropic openai
    from google.colab import userdata
    # Set keys from Colab Secrets (Settings → Secrets → Add)
    try: os.environ['ANTHROPIC_API_KEY'] = userdata.get('ANTHROPIC_API_KEY')
    except: pass
    try: os.environ['OPENAI_API_KEY'] = userdata.get('OPENAI_API_KEY')
    except: pass
    try: os.environ['GITLAB_TOKEN'] = userdata.get('GITLAB_TOKEN')
    except: pass
    # Fallback: manual input
    if not os.environ.get('ANTHROPIC_API_KEY'):
        import getpass
        for key in ['ANTHROPIC_API_KEY', 'OPENAI_API_KEY', 'GITLAB_TOKEN']:
            if not os.environ.get(key):
                try: os.environ[key] = getpass.getpass(f'{key}: ')
                except: pass

print(f'Environment: {"Colab" if IN_COLAB else "Local"}')
for k in ['ANTHROPIC_API_KEY','OPENAI_API_KEY','GITLAB_TOKEN']:
    print(f'  {k}: {"✓" if os.environ.get(k) else "✗"}')

# Colab Setup — API Keys & Dependencies import sys, os IN_COLAB = 'google.colab' in sys.modules if IN_COLAB: !pip install -q anthropic openai from google.colab import userdata # Set keys from Colab Secrets (Settings → Secrets → Add) try: os.environ['ANTHROPIC_API_KEY'] = userdata.get('ANTHROPIC_API_KEY') except: pass try: os.environ['OPENAI_API_KEY'] = userdata.get('OPENAI_API_KEY') except: pass try: os.environ['GITLAB_TOKEN'] = userdata.get('GITLAB_TOKEN') except: pass # Fallback: manual input if not os.environ.get('ANTHROPIC_API_KEY'): import getpass for key in ['ANTHROPIC_API_KEY', 'OPENAI_API_KEY', 'GITLAB_TOKEN']: if not os.environ.get(key): try: os.environ[key] = getpass.getpass(f'{key}: ') except: pass print(f'Environment: {"Colab" if IN_COLAB else "Local"}') for k in ['ANTHROPIC_API_KEY','OPENAI_API_KEY','GITLAB_TOKEN']: print(f' {k}: {"✓" if os.environ.get(k) else "✗"}')

Overview¶

The LLM Conversation Layer is CLARISSA's brain - it interprets user intent, maintains conversation context, and orchestrates the other components. This tutorial covers:

Component	Purpose
System Prompt	Define CLARISSA's role and capabilities
Intent Classifier	Understand what the user wants
Entity Extractor	Pull out reservoir parameters
Conversation Manager	Track multi-turn state
Context Optimizer	Fit relevant info in token limits

In [ ]:

# Setup - works in both GitPod and Colab
import os
import json
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple, Any
from enum import Enum, auto
import re
from datetime import datetime

# Check for API keys (optional - we'll use mock responses if not available)
OPENAI_API_KEY = os.environ.get('OPENAI_API_KEY')
ANTHROPIC_API_KEY = os.environ.get('ANTHROPIC_API_KEY')

if OPENAI_API_KEY or ANTHROPIC_API_KEY:
    print('API key found - will use real LLM calls')
else:
    print('No API key - using mock responses (tutorial still works!)')

# Setup - works in both GitPod and Colab import os import json from dataclasses import dataclass, field from typing import List, Dict, Optional, Tuple, Any from enum import Enum, auto import re from datetime import datetime # Check for API keys (optional - we'll use mock responses if not available) OPENAI_API_KEY = os.environ.get('OPENAI_API_KEY') ANTHROPIC_API_KEY = os.environ.get('ANTHROPIC_API_KEY') if OPENAI_API_KEY or ANTHROPIC_API_KEY: print('API key found - will use real LLM calls') else: print('No API key - using mock responses (tutorial still works!)')

Section 1: System Prompt Design¶

The system prompt defines CLARISSA's personality, capabilities, and constraints. A well-designed prompt is crucial for consistent, accurate responses.

In [ ]:

CLARISSA_SYSTEM_PROMPT = '''
You are CLARISSA (Conversational Language Agent for Reservoir Integrated Simulation System Analysis), 
an AI assistant specialized in helping reservoir engineers build and run simulations.

## Your Capabilities
1. Generate ECLIPSE-format simulation decks from natural language descriptions
2. Explain reservoir simulation concepts and keywords
3. Suggest reasonable defaults based on analog reservoirs
4. Validate physics constraints (pressure, saturation, rates)
5. Help interpret simulation results

## Your Constraints
- Always ask for clarification when specifications are ambiguous
- Flag assumptions explicitly so engineers can review them
- Never generate decks with physically impossible parameters
- Recommend OPM Flow compatible keywords when possible

## Response Format
- For deck generation: provide complete, runnable sections
- For explanations: be concise but technically accurate
- For clarifications: ask one question at a time
- Always specify units (FIELD or METRIC)

## Current Session Context
{context}
'''

def build_system_prompt(context: Dict[str, Any]) -> str:
    """Build system prompt with current session context."""
    context_str = json.dumps(context, indent=2) if context else "No active model"
    return CLARISSA_SYSTEM_PROMPT.format(context=context_str)

# Example context
session_context = {
    "model_name": "Permian_Waterflood_v1",
    "grid_size": [20, 20, 5],
    "phases": ["OIL", "WATER"],
    "units": "FIELD",
    "wells_defined": 5
}

print(build_system_prompt(session_context))

CLARISSA_SYSTEM_PROMPT = ''' You are CLARISSA (Conversational Language Agent for Reservoir Integrated Simulation System Analysis), an AI assistant specialized in helping reservoir engineers build and run simulations. ## Your Capabilities 1. Generate ECLIPSE-format simulation decks from natural language descriptions 2. Explain reservoir simulation concepts and keywords 3. Suggest reasonable defaults based on analog reservoirs 4. Validate physics constraints (pressure, saturation, rates) 5. Help interpret simulation results ## Your Constraints - Always ask for clarification when specifications are ambiguous - Flag assumptions explicitly so engineers can review them - Never generate decks with physically impossible parameters - Recommend OPM Flow compatible keywords when possible ## Response Format - For deck generation: provide complete, runnable sections - For explanations: be concise but technically accurate - For clarifications: ask one question at a time - Always specify units (FIELD or METRIC) ## Current Session Context {context} ''' def build_system_prompt(context: Dict[str, Any]) -> str: """Build system prompt with current session context.""" context_str = json.dumps(context, indent=2) if context else "No active model" return CLARISSA_SYSTEM_PROMPT.format(context=context_str) # Example context session_context = { "model_name": "Permian_Waterflood_v1", "grid_size": [20, 20, 5], "phases": ["OIL", "WATER"], "units": "FIELD", "wells_defined": 5 } print(build_system_prompt(session_context))

Section 2: Intent Classification¶

CLARISSA needs to understand what the user wants to do. We define a set of intents and use pattern matching (or an LLM) to classify them.

In [ ]:

class UserIntent(Enum):
    """Possible user intents in CLARISSA."""
    CREATE_MODEL = auto()      # Start a new simulation model
    MODIFY_MODEL = auto()      # Change existing model parameters
    ADD_WELL = auto()          # Add a well to the model
    SET_SCHEDULE = auto()      # Define time steps and controls
    RUN_SIMULATION = auto()    # Execute the simulation
    EXPLAIN_CONCEPT = auto()   # Ask about reservoir concepts
    EXPLAIN_KEYWORD = auto()   # Ask about ECLIPSE keywords
    SHOW_RESULTS = auto()      # View simulation results
    EXPORT_DECK = auto()       # Export the deck file
    CLARIFY = auto()           # User providing clarification
    UNKNOWN = auto()           # Cannot determine intent

# Pattern-based intent classification (fast, no API needed)
INTENT_PATTERNS = {
    UserIntent.CREATE_MODEL: [
        r'create.*model', r'new.*simulation', r'build.*deck',
        r'start.*model', r'set up.*reservoir'
    ],
    UserIntent.ADD_WELL: [
        r'add.*well', r'new.*well', r'place.*well',
        r'drill.*well', r'producer', r'injector'
    ],
    UserIntent.MODIFY_MODEL: [
        r'change.*perm', r'modify.*poro', r'update.*grid',
        r'set.*pressure', r'adjust'
    ],
    UserIntent.RUN_SIMULATION: [
        r'run.*sim', r'execute', r'start.*run', r'simulate'
    ],
    UserIntent.EXPLAIN_CONCEPT: [
        r'what is', r'explain', r'how does', r'why',
        r'tell me about', r'describe'
    ],
    UserIntent.EXPLAIN_KEYWORD: [
        r'keyword', r'WELSPECS', r'COMPDAT', r'WCONPROD',
        r'EQUIL', r'syntax'
    ],
    UserIntent.EXPORT_DECK: [
        r'export', r'save.*deck', r'download', r'generate.*file'
    ],
    UserIntent.SHOW_RESULTS: [
        r'show.*result', r'plot', r'graph', r'production.*rate',
        r'water.*cut', r'pressure.*profile'
    ]
}

def classify_intent_patterns(user_message: str) -> Tuple[UserIntent, float]:
    """Classify intent using regex patterns.
    
    Returns (intent, confidence).
    """
    message_lower = user_message.lower()
    
    for intent, patterns in INTENT_PATTERNS.items():
        for pattern in patterns:
            if re.search(pattern, message_lower):
                return intent, 0.8  # Pattern match = 80% confidence
    
    return UserIntent.UNKNOWN, 0.3

# Test intent classification
test_messages = [
    "I want to create a new waterflood model",
    "Add a producer well at coordinates 10,10",
    "What is relative permeability?",
    "Run the simulation for 5 years",
    "Show me the oil production rate",
    "Something completely random"
]

print("Intent Classification Results:")
print("-" * 60)
for msg in test_messages:
    intent, conf = classify_intent_patterns(msg)
    print(f"{msg[:40]:40} -> {intent.name:20} ({conf:.0%})")

class UserIntent(Enum): """Possible user intents in CLARISSA.""" CREATE_MODEL = auto() # Start a new simulation model MODIFY_MODEL = auto() # Change existing model parameters ADD_WELL = auto() # Add a well to the model SET_SCHEDULE = auto() # Define time steps and controls RUN_SIMULATION = auto() # Execute the simulation EXPLAIN_CONCEPT = auto() # Ask about reservoir concepts EXPLAIN_KEYWORD = auto() # Ask about ECLIPSE keywords SHOW_RESULTS = auto() # View simulation results EXPORT_DECK = auto() # Export the deck file CLARIFY = auto() # User providing clarification UNKNOWN = auto() # Cannot determine intent # Pattern-based intent classification (fast, no API needed) INTENT_PATTERNS = { UserIntent.CREATE_MODEL: [ r'create.*model', r'new.*simulation', r'build.*deck', r'start.*model', r'set up.*reservoir' ], UserIntent.ADD_WELL: [ r'add.*well', r'new.*well', r'place.*well', r'drill.*well', r'producer', r'injector' ], UserIntent.MODIFY_MODEL: [ r'change.*perm', r'modify.*poro', r'update.*grid', r'set.*pressure', r'adjust' ], UserIntent.RUN_SIMULATION: [ r'run.*sim', r'execute', r'start.*run', r'simulate' ], UserIntent.EXPLAIN_CONCEPT: [ r'what is', r'explain', r'how does', r'why', r'tell me about', r'describe' ], UserIntent.EXPLAIN_KEYWORD: [ r'keyword', r'WELSPECS', r'COMPDAT', r'WCONPROD', r'EQUIL', r'syntax' ], UserIntent.EXPORT_DECK: [ r'export', r'save.*deck', r'download', r'generate.*file' ], UserIntent.SHOW_RESULTS: [ r'show.*result', r'plot', r'graph', r'production.*rate', r'water.*cut', r'pressure.*profile' ] } def classify_intent_patterns(user_message: str) -> Tuple[UserIntent, float]: """Classify intent using regex patterns. Returns (intent, confidence). """ message_lower = user_message.lower() for intent, patterns in INTENT_PATTERNS.items(): for pattern in patterns: if re.search(pattern, message_lower): return intent, 0.8 # Pattern match = 80% confidence return UserIntent.UNKNOWN, 0.3 # Test intent classification test_messages = [ "I want to create a new waterflood model", "Add a producer well at coordinates 10,10", "What is relative permeability?", "Run the simulation for 5 years", "Show me the oil production rate", "Something completely random" ] print("Intent Classification Results:") print("-" * 60) for msg in test_messages: intent, conf = classify_intent_patterns(msg) print(f"{msg[:40]:40} -> {intent.name:20} ({conf:.0%})")

Section 3: Entity Extraction¶

Once we know the intent, we need to extract specific parameters (entities) from the user's message.

In [ ]:

@dataclass
class ExtractedEntities:
    """Entities extracted from user message."""
    well_names: List[str] = field(default_factory=list)
    coordinates: List[Tuple[int, int]] = field(default_factory=list)
    depths: List[float] = field(default_factory=list)
    rates: List[float] = field(default_factory=list)
    pressures: List[float] = field(default_factory=list)
    percentages: List[float] = field(default_factory=list)
    time_values: List[Tuple[float, str]] = field(default_factory=list)  # (value, unit)
    keywords: List[str] = field(default_factory=list)
    well_types: List[str] = field(default_factory=list)  # producer, injector
    fluids: List[str] = field(default_factory=list)  # oil, water, gas

def extract_entities(message: str) -> ExtractedEntities:
    """Extract reservoir-specific entities from user message."""
    entities = ExtractedEntities()
    message_lower = message.lower()
    
    # Well names (P1, INJ-1, PROD_01, etc.)
    well_pattern = r'\b([A-Z]+[-_]?\d+)\b'
    entities.well_names = re.findall(well_pattern, message)
    
    # Coordinates (10,10 or i=10, j=10)
    coord_pattern = r'(?:at\s+)?(?:coordinates?\s+)?(\d+)\s*[,\s]\s*(\d+)'
    coords = re.findall(coord_pattern, message_lower)
    entities.coordinates = [(int(i), int(j)) for i, j in coords]
    
    # Depths (8000 ft, 2500m)
    depth_pattern = r'(\d+(?:\.\d+)?)\s*(?:ft|feet|m|meters?)\s*(?:depth|deep|tvd)?'
    entities.depths = [float(d) for d in re.findall(depth_pattern, message_lower)]
    
    # Rates (1000 stb/d, 5000 bpd, 10 mscf/d)
    rate_pattern = r'(\d+(?:\.\d+)?)\s*(?:stb/d|bpd|bbl/d|mscf/d|scf/d)'
    entities.rates = [float(r) for r in re.findall(rate_pattern, message_lower)]
    
    # Pressures (3000 psi, 200 bar)
    pressure_pattern = r'(\d+(?:\.\d+)?)\s*(?:psi|psia|bar|kpa)'
    entities.pressures = [float(p) for p in re.findall(pressure_pattern, message_lower)]
    
    # Percentages (20%, 0.2 porosity)
    pct_pattern = r'(\d+(?:\.\d+)?)\s*%'
    entities.percentages = [float(p) for p in re.findall(pct_pattern, message_lower)]
    
    # Time values (5 years, 365 days, 12 months)
    time_pattern = r'(\d+(?:\.\d+)?)\s*(years?|months?|days?|weeks?)'
    entities.time_values = [(float(v), u) for v, u in re.findall(time_pattern, message_lower)]
    
    # ECLIPSE keywords
    keywords = ['WELSPECS', 'COMPDAT', 'WCONPROD', 'WCONINJE', 'EQUIL', 
                'PERMX', 'PORO', 'SWOF', 'PVDO', 'TSTEP', 'DIMENS']
    entities.keywords = [kw for kw in keywords if kw in message.upper()]
    
    # Well types
    if 'producer' in message_lower or 'production' in message_lower:
        entities.well_types.append('producer')
    if 'injector' in message_lower or 'injection' in message_lower:
        entities.well_types.append('injector')
    
    # Fluids
    for fluid in ['oil', 'water', 'gas']:
        if fluid in message_lower:
            entities.fluids.append(fluid)
    
    return entities

# Test entity extraction
test_messages = [
    "Add a producer well P1 at coordinates 10, 15 with 1000 stb/d rate",
    "Set the reservoir depth to 8500 ft and initial pressure to 3800 psi",
    "Run simulation for 5 years with water injection at 2000 bpd",
    "Change PERMX to 150 md and porosity to 22%"
]

for msg in test_messages:
    print(f"\nMessage: {msg}")
    entities = extract_entities(msg)
    for field_name, value in entities.__dict__.items():
        if value:
            print(f"  {field_name}: {value}")

@dataclass class ExtractedEntities: """Entities extracted from user message.""" well_names: List[str] = field(default_factory=list) coordinates: List[Tuple[int, int]] = field(default_factory=list) depths: List[float] = field(default_factory=list) rates: List[float] = field(default_factory=list) pressures: List[float] = field(default_factory=list) percentages: List[float] = field(default_factory=list) time_values: List[Tuple[float, str]] = field(default_factory=list) # (value, unit) keywords: List[str] = field(default_factory=list) well_types: List[str] = field(default_factory=list) # producer, injector fluids: List[str] = field(default_factory=list) # oil, water, gas def extract_entities(message: str) -> ExtractedEntities: """Extract reservoir-specific entities from user message.""" entities = ExtractedEntities() message_lower = message.lower() # Well names (P1, INJ-1, PROD_01, etc.) well_pattern = r'\b([A-Z]+[-_]?\d+)\b' entities.well_names = re.findall(well_pattern, message) # Coordinates (10,10 or i=10, j=10) coord_pattern = r'(?:at\s+)?(?:coordinates?\s+)?(\d+)\s*[,\s]\s*(\d+)' coords = re.findall(coord_pattern, message_lower) entities.coordinates = [(int(i), int(j)) for i, j in coords] # Depths (8000 ft, 2500m) depth_pattern = r'(\d+(?:\.\d+)?)\s*(?:ft|feet|m|meters?)\s*(?:depth|deep|tvd)?' entities.depths = [float(d) for d in re.findall(depth_pattern, message_lower)] # Rates (1000 stb/d, 5000 bpd, 10 mscf/d) rate_pattern = r'(\d+(?:\.\d+)?)\s*(?:stb/d|bpd|bbl/d|mscf/d|scf/d)' entities.rates = [float(r) for r in re.findall(rate_pattern, message_lower)] # Pressures (3000 psi, 200 bar) pressure_pattern = r'(\d+(?:\.\d+)?)\s*(?:psi|psia|bar|kpa)' entities.pressures = [float(p) for p in re.findall(pressure_pattern, message_lower)] # Percentages (20%, 0.2 porosity) pct_pattern = r'(\d+(?:\.\d+)?)\s*%' entities.percentages = [float(p) for p in re.findall(pct_pattern, message_lower)] # Time values (5 years, 365 days, 12 months) time_pattern = r'(\d+(?:\.\d+)?)\s*(years?|months?|days?|weeks?)' entities.time_values = [(float(v), u) for v, u in re.findall(time_pattern, message_lower)] # ECLIPSE keywords keywords = ['WELSPECS', 'COMPDAT', 'WCONPROD', 'WCONINJE', 'EQUIL', 'PERMX', 'PORO', 'SWOF', 'PVDO', 'TSTEP', 'DIMENS'] entities.keywords = [kw for kw in keywords if kw in message.upper()] # Well types if 'producer' in message_lower or 'production' in message_lower: entities.well_types.append('producer') if 'injector' in message_lower or 'injection' in message_lower: entities.well_types.append('injector') # Fluids for fluid in ['oil', 'water', 'gas']: if fluid in message_lower: entities.fluids.append(fluid) return entities # Test entity extraction test_messages = [ "Add a producer well P1 at coordinates 10, 15 with 1000 stb/d rate", "Set the reservoir depth to 8500 ft and initial pressure to 3800 psi", "Run simulation for 5 years with water injection at 2000 bpd", "Change PERMX to 150 md and porosity to 22%" ] for msg in test_messages: print(f"\nMessage: {msg}") entities = extract_entities(msg) for field_name, value in entities.__dict__.items(): if value: print(f" {field_name}: {value}")

Section 4: Conversation Manager¶

The Conversation Manager tracks the state across multiple turns, maintaining context and handling clarifications.

In [ ]:

@dataclass
class Message:
    """A single message in the conversation."""
    role: str  # 'user', 'assistant', 'system'
    content: str
    timestamp: datetime = field(default_factory=datetime.now)
    intent: Optional[UserIntent] = None
    entities: Optional[ExtractedEntities] = None

@dataclass
class ConversationState:
    """Current state of the simulation being built."""
    model_name: Optional[str] = None
    grid_defined: bool = False
    props_defined: bool = False
    wells: List[Dict] = field(default_factory=list)
    schedule_defined: bool = False
    pending_clarification: Optional[str] = None
    assumptions: List[str] = field(default_factory=list)

class ConversationManager:
    """Manages multi-turn conversation state."""
    
    def __init__(self, max_history: int = 20):
        self.messages: List[Message] = []
        self.state = ConversationState()
        self.max_history = max_history
    
    def add_user_message(self, content: str) -> Message:
        """Process and add a user message."""
        intent, confidence = classify_intent_patterns(content)
        entities = extract_entities(content)
        
        msg = Message(
            role='user',
            content=content,
            intent=intent,
            entities=entities
        )
        self.messages.append(msg)
        self._update_state(msg)
        return msg
    
    def add_assistant_message(self, content: str) -> Message:
        """Add an assistant response."""
        msg = Message(role='assistant', content=content)
        self.messages.append(msg)
        return msg
    
    def _update_state(self, msg: Message):
        """Update conversation state based on user message."""
        if msg.intent == UserIntent.CREATE_MODEL:
            self.state.model_name = "New_Model"
        
        if msg.entities and msg.entities.well_names:
            for name in msg.entities.well_names:
                if not any(w['name'] == name for w in self.state.wells):
                    well_type = 'injector' if 'injector' in msg.entities.well_types else 'producer'
                    self.state.wells.append({
                        'name': name,
                        'type': well_type,
                        'coordinates': msg.entities.coordinates[0] if msg.entities.coordinates else None
                    })
    
    def get_context_for_llm(self) -> List[Dict[str, str]]:
        """Get conversation history formatted for LLM API."""
        # Keep only recent messages to fit context window
        recent = self.messages[-self.max_history:]
        return [
            {'role': m.role, 'content': m.content}
            for m in recent
        ]
    
    def get_state_summary(self) -> str:
        """Get a summary of current model state."""
        lines = ["Current Model State:"]
        lines.append(f"  Model: {self.state.model_name or 'Not started'}")
        lines.append(f"  Grid defined: {self.state.grid_defined}")
        lines.append(f"  Properties defined: {self.state.props_defined}")
        lines.append(f"  Wells: {len(self.state.wells)}")
        for w in self.state.wells:
            lines.append(f"    - {w['name']} ({w['type']}) at {w.get('coordinates', 'TBD')}")
        if self.state.assumptions:
            lines.append(f"  Assumptions made: {len(self.state.assumptions)}")
        return "\n".join(lines)

# Demo conversation
conv = ConversationManager()

# Simulate a multi-turn conversation
turns = [
    "I want to create a new waterflood model for the Permian Basin",
    "Add a producer well P1 at coordinates 10, 10",
    "Add water injector INJ1 at 1, 1 with 2000 bpd injection rate",
]

for turn in turns:
    msg = conv.add_user_message(turn)
    print(f"\nUser: {turn}")
    print(f"  Intent: {msg.intent.name}")
    conv.add_assistant_message(f"Processed: {msg.intent.name}")

print("\n" + "="*50)
print(conv.get_state_summary())

@dataclass class Message: """A single message in the conversation.""" role: str # 'user', 'assistant', 'system' content: str timestamp: datetime = field(default_factory=datetime.now) intent: Optional[UserIntent] = None entities: Optional[ExtractedEntities] = None @dataclass class ConversationState: """Current state of the simulation being built.""" model_name: Optional[str] = None grid_defined: bool = False props_defined: bool = False wells: List[Dict] = field(default_factory=list) schedule_defined: bool = False pending_clarification: Optional[str] = None assumptions: List[str] = field(default_factory=list) class ConversationManager: """Manages multi-turn conversation state.""" def __init__(self, max_history: int = 20): self.messages: List[Message] = [] self.state = ConversationState() self.max_history = max_history def add_user_message(self, content: str) -> Message: """Process and add a user message.""" intent, confidence = classify_intent_patterns(content) entities = extract_entities(content) msg = Message( role='user', content=content, intent=intent, entities=entities ) self.messages.append(msg) self._update_state(msg) return msg def add_assistant_message(self, content: str) -> Message: """Add an assistant response.""" msg = Message(role='assistant', content=content) self.messages.append(msg) return msg def _update_state(self, msg: Message): """Update conversation state based on user message.""" if msg.intent == UserIntent.CREATE_MODEL: self.state.model_name = "New_Model" if msg.entities and msg.entities.well_names: for name in msg.entities.well_names: if not any(w['name'] == name for w in self.state.wells): well_type = 'injector' if 'injector' in msg.entities.well_types else 'producer' self.state.wells.append({ 'name': name, 'type': well_type, 'coordinates': msg.entities.coordinates[0] if msg.entities.coordinates else None }) def get_context_for_llm(self) -> List[Dict[str, str]]: """Get conversation history formatted for LLM API.""" # Keep only recent messages to fit context window recent = self.messages[-self.max_history:] return [ {'role': m.role, 'content': m.content} for m in recent ] def get_state_summary(self) -> str: """Get a summary of current model state.""" lines = ["Current Model State:"] lines.append(f" Model: {self.state.model_name or 'Not started'}") lines.append(f" Grid defined: {self.state.grid_defined}") lines.append(f" Properties defined: {self.state.props_defined}") lines.append(f" Wells: {len(self.state.wells)}") for w in self.state.wells: lines.append(f" - {w['name']} ({w['type']}) at {w.get('coordinates', 'TBD')}") if self.state.assumptions: lines.append(f" Assumptions made: {len(self.state.assumptions)}") return "\n".join(lines) # Demo conversation conv = ConversationManager() # Simulate a multi-turn conversation turns = [ "I want to create a new waterflood model for the Permian Basin", "Add a producer well P1 at coordinates 10, 10", "Add water injector INJ1 at 1, 1 with 2000 bpd injection rate", ] for turn in turns: msg = conv.add_user_message(turn) print(f"\nUser: {turn}") print(f" Intent: {msg.intent.name}") conv.add_assistant_message(f"Processed: {msg.intent.name}") print("\n" + "="*50) print(conv.get_state_summary())

Section 5: Context Window Optimization¶

LLMs have limited context windows. We need to intelligently select what information to include.

In [ ]:

def estimate_tokens(text: str) -> int:
    """Rough token estimate (4 chars per token for English)."""
    return len(text) // 4

@dataclass
class ContextBudget:
    """Token budget allocation for context window."""
    total: int = 8000  # Conservative for most models
    system_prompt: int = 1000
    conversation_history: int = 2000
    retrieved_knowledge: int = 3000
    current_deck: int = 1500
    response_buffer: int = 500

class ContextOptimizer:
    """Optimizes context window usage."""
    
    def __init__(self, budget: ContextBudget = None):
        self.budget = budget or ContextBudget()
    
    def build_context(
        self,
        system_prompt: str,
        conversation: List[Dict[str, str]],
        knowledge: List[str],
        current_deck: Optional[str] = None
    ) -> Tuple[str, Dict[str, int]]:
        """Build optimized context within token budget.
        
        Returns (context_string, token_usage).
        """
        usage = {}
        parts = []
        
        # 1. System prompt (always include, truncate if needed)
        sys_tokens = estimate_tokens(system_prompt)
        if sys_tokens > self.budget.system_prompt:
            # Truncate to fit
            char_limit = self.budget.system_prompt * 4
            system_prompt = system_prompt[:char_limit] + "..."
        parts.append(system_prompt)
        usage['system'] = estimate_tokens(system_prompt)
        
        # 2. Conversation history (most recent first)
        conv_parts = []
        conv_tokens = 0
        for msg in reversed(conversation):
            msg_text = f"{msg['role']}: {msg['content']}"
            msg_tokens = estimate_tokens(msg_text)
            if conv_tokens + msg_tokens > self.budget.conversation_history:
                break
            conv_parts.insert(0, msg_text)
            conv_tokens += msg_tokens
        if conv_parts:
            parts.append("\n--- Conversation History ---\n" + "\n".join(conv_parts))
        usage['conversation'] = conv_tokens
        
        # 3. Retrieved knowledge (prioritize by relevance)
        knowledge_parts = []
        knowledge_tokens = 0
        for doc in knowledge:
            doc_tokens = estimate_tokens(doc)
            if knowledge_tokens + doc_tokens > self.budget.retrieved_knowledge:
                break
            knowledge_parts.append(doc)
            knowledge_tokens += doc_tokens
        if knowledge_parts:
            parts.append("\n--- Relevant Knowledge ---\n" + "\n".join(knowledge_parts))
        usage['knowledge'] = knowledge_tokens
        
        # 4. Current deck state (if available)
        if current_deck:
            deck_tokens = estimate_tokens(current_deck)
            if deck_tokens > self.budget.current_deck:
                # Summarize deck instead of including full text
                current_deck = self._summarize_deck(current_deck)
            parts.append("\n--- Current Deck ---\n" + current_deck)
            usage['deck'] = estimate_tokens(current_deck)
        
        usage['total'] = sum(usage.values())
        usage['remaining'] = self.budget.total - usage['total']
        
        return "\n".join(parts), usage
    
    def _summarize_deck(self, deck: str) -> str:
        """Create a summary of deck contents."""
        lines = deck.split('\n')
        sections = []
        current_section = None
        
        for line in lines:
            stripped = line.strip()
            if stripped in ['RUNSPEC', 'GRID', 'PROPS', 'SOLUTION', 'SCHEDULE']:
                current_section = stripped
                sections.append(current_section)
        
        return f"Deck contains sections: {', '.join(sections)}\n(Full deck truncated for context)"

# Demo context optimization
optimizer = ContextOptimizer()

context, usage = optimizer.build_context(
    system_prompt=CLARISSA_SYSTEM_PROMPT.format(context="Demo model"),
    conversation=[
        {'role': 'user', 'content': 'Create a waterflood model'},
        {'role': 'assistant', 'content': 'I will create a 5-spot waterflood pattern.'},
        {'role': 'user', 'content': 'Add a producer at 10,10'}
    ],
    knowledge=[
        "WELSPECS defines well name, group, and location...",
        "Typical Permian Basin permeability: 50-200 md..."
    ],
    current_deck="RUNSPEC\nDIMENS\n10 10 5 /\n..."
)

print("Token Usage:")
for key, value in usage.items():
    print(f"  {key}: {value} tokens")

def estimate_tokens(text: str) -> int: """Rough token estimate (4 chars per token for English).""" return len(text) // 4 @dataclass class ContextBudget: """Token budget allocation for context window.""" total: int = 8000 # Conservative for most models system_prompt: int = 1000 conversation_history: int = 2000 retrieved_knowledge: int = 3000 current_deck: int = 1500 response_buffer: int = 500 class ContextOptimizer: """Optimizes context window usage.""" def __init__(self, budget: ContextBudget = None): self.budget = budget or ContextBudget() def build_context( self, system_prompt: str, conversation: List[Dict[str, str]], knowledge: List[str], current_deck: Optional[str] = None ) -> Tuple[str, Dict[str, int]]: """Build optimized context within token budget. Returns (context_string, token_usage). """ usage = {} parts = [] # 1. System prompt (always include, truncate if needed) sys_tokens = estimate_tokens(system_prompt) if sys_tokens > self.budget.system_prompt: # Truncate to fit char_limit = self.budget.system_prompt * 4 system_prompt = system_prompt[:char_limit] + "..." parts.append(system_prompt) usage['system'] = estimate_tokens(system_prompt) # 2. Conversation history (most recent first) conv_parts = [] conv_tokens = 0 for msg in reversed(conversation): msg_text = f"{msg['role']}: {msg['content']}" msg_tokens = estimate_tokens(msg_text) if conv_tokens + msg_tokens > self.budget.conversation_history: break conv_parts.insert(0, msg_text) conv_tokens += msg_tokens if conv_parts: parts.append("\n--- Conversation History ---\n" + "\n".join(conv_parts)) usage['conversation'] = conv_tokens # 3. Retrieved knowledge (prioritize by relevance) knowledge_parts = [] knowledge_tokens = 0 for doc in knowledge: doc_tokens = estimate_tokens(doc) if knowledge_tokens + doc_tokens > self.budget.retrieved_knowledge: break knowledge_parts.append(doc) knowledge_tokens += doc_tokens if knowledge_parts: parts.append("\n--- Relevant Knowledge ---\n" + "\n".join(knowledge_parts)) usage['knowledge'] = knowledge_tokens # 4. Current deck state (if available) if current_deck: deck_tokens = estimate_tokens(current_deck) if deck_tokens > self.budget.current_deck: # Summarize deck instead of including full text current_deck = self._summarize_deck(current_deck) parts.append("\n--- Current Deck ---\n" + current_deck) usage['deck'] = estimate_tokens(current_deck) usage['total'] = sum(usage.values()) usage['remaining'] = self.budget.total - usage['total'] return "\n".join(parts), usage def _summarize_deck(self, deck: str) -> str: """Create a summary of deck contents.""" lines = deck.split('\n') sections = [] current_section = None for line in lines: stripped = line.strip() if stripped in ['RUNSPEC', 'GRID', 'PROPS', 'SOLUTION', 'SCHEDULE']: current_section = stripped sections.append(current_section) return f"Deck contains sections: {', '.join(sections)}\n(Full deck truncated for context)" # Demo context optimization optimizer = ContextOptimizer() context, usage = optimizer.build_context( system_prompt=CLARISSA_SYSTEM_PROMPT.format(context="Demo model"), conversation=[ {'role': 'user', 'content': 'Create a waterflood model'}, {'role': 'assistant', 'content': 'I will create a 5-spot waterflood pattern.'}, {'role': 'user', 'content': 'Add a producer at 10,10'} ], knowledge=[ "WELSPECS defines well name, group, and location...", "Typical Permian Basin permeability: 50-200 md..." ], current_deck="RUNSPEC\nDIMENS\n10 10 5 /\n..." ) print("Token Usage:") for key, value in usage.items(): print(f" {key}: {value} tokens")

Section 6: LLM Integration (Mock + Real)¶

Now we tie it all together with actual LLM calls (or mock responses for testing).

In [ ]:

class MockLLM:
    """Mock LLM for testing without API keys."""
    
    def __init__(self):
        self.responses = {
            UserIntent.CREATE_MODEL: "I'll help you create a new simulation model. What type of reservoir are you modeling? (e.g., waterflood, gas reservoir, compositional)",
            UserIntent.ADD_WELL: "I'll add the well to your model. Let me generate the WELSPECS and COMPDAT entries.",
            UserIntent.EXPLAIN_CONCEPT: "Here's an explanation of that concept...",
            UserIntent.RUN_SIMULATION: "I'll prepare your deck for simulation. Let me verify all required sections are complete.",
            UserIntent.UNKNOWN: "I'm not sure what you're asking. Could you please clarify?"
        }
    
    def generate(self, messages: List[Dict], intent: UserIntent = None) -> str:
        """Generate a mock response."""
        if intent and intent in self.responses:
            return self.responses[intent]
        return self.responses[UserIntent.UNKNOWN]

class CLARISSAAgent:
    """Main CLARISSA conversation agent."""
    
    def __init__(self, use_real_llm: bool = False):
        self.conversation = ConversationManager()
        self.optimizer = ContextOptimizer()
        self.llm = MockLLM()  # Use mock by default
        self.knowledge_cache: List[str] = []
    
    def chat(self, user_input: str) -> str:
        """Process user input and generate response."""
        # 1. Add and analyze user message
        msg = self.conversation.add_user_message(user_input)
        
        # 2. Build optimized context
        context, usage = self.optimizer.build_context(
            system_prompt=build_system_prompt({}),
            conversation=self.conversation.get_context_for_llm(),
            knowledge=self.knowledge_cache
        )
        
        # 3. Generate response
        response = self.llm.generate(
            self.conversation.get_context_for_llm(),
            intent=msg.intent
        )
        
        # 4. Add response to history
        self.conversation.add_assistant_message(response)
        
        return response
    
    def get_status(self) -> str:
        """Get current conversation and model status."""
        return self.conversation.get_state_summary()

# Demo the agent
agent = CLARISSAAgent()

print("CLARISSA Conversation Demo")
print("=" * 50)

demo_inputs = [
    "I want to create a waterflood model for the Permian Basin",
    "Add a producer well P1 at location 10, 10",
    "Run the simulation for 5 years"
]

for user_input in demo_inputs:
    print(f"\nUser: {user_input}")
    response = agent.chat(user_input)
    print(f"CLARISSA: {response}")

print("\n" + "=" * 50)
print(agent.get_status())

class MockLLM: """Mock LLM for testing without API keys.""" def __init__(self): self.responses = { UserIntent.CREATE_MODEL: "I'll help you create a new simulation model. What type of reservoir are you modeling? (e.g., waterflood, gas reservoir, compositional)", UserIntent.ADD_WELL: "I'll add the well to your model. Let me generate the WELSPECS and COMPDAT entries.", UserIntent.EXPLAIN_CONCEPT: "Here's an explanation of that concept...", UserIntent.RUN_SIMULATION: "I'll prepare your deck for simulation. Let me verify all required sections are complete.", UserIntent.UNKNOWN: "I'm not sure what you're asking. Could you please clarify?" } def generate(self, messages: List[Dict], intent: UserIntent = None) -> str: """Generate a mock response.""" if intent and intent in self.responses: return self.responses[intent] return self.responses[UserIntent.UNKNOWN] class CLARISSAAgent: """Main CLARISSA conversation agent.""" def __init__(self, use_real_llm: bool = False): self.conversation = ConversationManager() self.optimizer = ContextOptimizer() self.llm = MockLLM() # Use mock by default self.knowledge_cache: List[str] = [] def chat(self, user_input: str) -> str: """Process user input and generate response.""" # 1. Add and analyze user message msg = self.conversation.add_user_message(user_input) # 2. Build optimized context context, usage = self.optimizer.build_context( system_prompt=build_system_prompt({}), conversation=self.conversation.get_context_for_llm(), knowledge=self.knowledge_cache ) # 3. Generate response response = self.llm.generate( self.conversation.get_context_for_llm(), intent=msg.intent ) # 4. Add response to history self.conversation.add_assistant_message(response) return response def get_status(self) -> str: """Get current conversation and model status.""" return self.conversation.get_state_summary() # Demo the agent agent = CLARISSAAgent() print("CLARISSA Conversation Demo") print("=" * 50) demo_inputs = [ "I want to create a waterflood model for the Permian Basin", "Add a producer well P1 at location 10, 10", "Run the simulation for 5 years" ] for user_input in demo_inputs: print(f"\nUser: {user_input}") response = agent.chat(user_input) print(f"CLARISSA: {response}") print("\n" + "=" * 50) print(agent.get_status())

Summary¶

In this tutorial, we learned:

1. System Prompt Design: Define CLARISSA's role, capabilities, and constraints
2. Intent Classification: Understand what the user wants using patterns or LLM
3. Entity Extraction: Pull reservoir-specific parameters from natural language
4. Conversation Management: Track state across multiple turns
5. Context Optimization: Fit relevant information within token limits

Key Takeaways:

• Domain-specific prompts improve accuracy
• Hybrid approach (patterns + LLM) balances speed and accuracy
• Context window management is critical for complex models

Next Tutorial: 05_Constraint_Engine.ipynb - Physics validation with Z3