# msia-agent

> MSI-a conversational agent patterns using LangGraph. Trigger: When working on agent conversation flow, nodes, state, tools, prompts, or FSM.

- Author: MSI-a Developer
- Repository: pepeccz/msi-a
- Version: 20260122123042
- Stars: 0
- Forks: 0
- Last Updated: 2026-02-06
- Source: https://github.com/pepeccz/msi-a
- Web: https://mule.run/skillshub/@@pepeccz/msi-a~msia-agent:20260122123042

---

---
name: msia-agent
description: >
  MSI-a conversational agent patterns using LangGraph.
  Trigger: When working on agent conversation flow, nodes, state, tools, prompts, or FSM.
metadata:
  author: msi-automotive
  version: "2.0"
  scope: [root, agent]
  auto_invoke:
    - "Working on agent conversation flow"
    - "Creating/modifying graph nodes"
    - "Working with ConversationState"
    - "Creating/modifying agent tools"
    - "Working on system prompts"
    - "Working on FSM case collection"
---

## Agent Structure

```
agent/
├── main.py                    # Entry point, Redis stream consumer
├── graphs/
│   └── conversation_flow.py   # StateGraph definition, security delimiters
├── nodes/
│   ├── process_message.py     # Message preprocessing
│   └── conversational_agent.py  # LLM response generation (uses dynamic prompts)
├── tools/
│   ├── tarifa_tools.py        # Tariff calculation tools
│   ├── element_tools.py       # Element matching tools (identificar_y_resolver_elementos)
│   ├── vehicle_tools.py       # Vehicle identification
│   ├── case_tools.py          # Case/expediente management (FSM transitions)
│   └── image_tools.py         # Image sending tools (enviar_imagenes_ejemplo)
├── services/
│   ├── tarifa_service.py      # Tariff business logic (cached)
│   ├── element_service.py     # Element matching service
│   ├── token_tracking.py      # Token usage tracking
│   └── prompt_service.py      # Calculator prompt (legacy)
├── state/
│   ├── schemas.py             # ConversationState TypedDict
│   ├── checkpointer.py        # Redis checkpointer
│   └── helpers.py             # State utilities, message formatting
├── fsm/
│   └── case_collection.py     # Data collection FSM (6 phases)
└── prompts/
    ├── __init__.py            # Package exports
    ├── loader.py              # Dynamic prompt assembly (~40-60% token savings)
    ├── state_summary.py       # Real-time state summary generator
    ├── system.md              # Legacy static prompt (backup)
    ├── calculator_base.py     # Calculator prompt template
    ├── core/                  # ~2,200 tokens - ALWAYS included
    │   ├── 01_security.md     # Security, anti-jailbreak
    │   ├── 02_identity.md     # MSI-a identity
    │   ├── 03_format_style.md # Tone, format, categories
    │   ├── 04_anti_patterns.md # Anti-loop, anti-invention
    │   ├── 05_tools_efficiency.md # Tool usage rules
    │   ├── 06_escalation.md   # When to escalate
    │   ├── 07_pricing_rules.md # Price communication (CRITICAL)
    │   └── 08_documentation.md # Doc rules (STRICT)
    └── phases/                # One per call based on FSM state
        ├── idle_quotation.md  # ~1,000 tokens - Presupuestacion
        ├── collect_images.md  # ~500 tokens - Image collection
        ├── collect_personal.md # ~550 tokens - Personal data
        ├── collect_vehicle.md # ~450 tokens - Vehicle data
        ├── collect_workshop.md # ~550 tokens - Workshop data
        └── review_summary.md  # ~600 tokens - Final review
```

## Dynamic Prompts System

The agent uses a modular prompt system that reduces token usage by 40-60%:

```python
# In conversational_agent.py
from agent.prompts.loader import assemble_system_prompt
from agent.prompts.state_summary import generate_state_summary

# Generate dynamic prompt based on FSM state
state_summary = generate_state_summary(fsm_state, last_tariff_result)
system_prompt = assemble_system_prompt(
    fsm_state=fsm_state,
    state_summary=state_summary,
    client_context=client_context,
)
```

### Prompt Assembly Flow

```
CORE modules (~2,200 tokens)     # Always included
    +
PHASE module (~500-1,000 tokens) # Based on current FSM state
    +
STATE_SUMMARY (~100 tokens)      # Dynamic context (price, elements, etc.)
    +
CLIENT_CONTEXT (~200 tokens)     # Client type, categories
    =
TOTAL: ~3,000-3,500 tokens       # vs ~7,000 tokens with legacy prompt
```

### Adding New Phases

1. Create `prompts/phases/new_phase.md`
2. Add to `loader.py` PHASE_MODULES dict
3. Map FSM CollectionStep to the new phase

---

## ConversationState Schema

```python
class ConversationState(TypedDict, total=False):
    # Core Metadata
    conversation_id: str          # LangGraph thread_id
    user_phone: str               # E.164 format
    user_name: str | None
    user_id: str | None           # Database UUID
    client_type: str | None       # "particular" or "professional"
    
    # Messages
    messages: list[dict[str, Any]]  # Conversation history
    user_message: str | None        # Current message
    
    # FSM State
    fsm_state: dict[str, Any] | None  # Contains case_collection state
    
    # Flags
    is_first_interaction: bool
    escalation_triggered: bool
    agent_disabled: bool            # Panic button
    
    # Tool Results
    pending_images: dict[str, Any]    # Images to send + follow_up_message
    incoming_attachments: list[dict]  # User attachments
    tarifa_actual: dict[str, Any]     # Last tariff calculation result
```

---

## FSM Flow (Case Collection)

```
IDLE
  ↓ (iniciar_expediente)
COLLECT_IMAGES
  ↓ (continuar_a_datos_personales - when user says "listo")
COLLECT_PERSONAL
  ↓ (actualizar_datos_expediente with datos_personales)
COLLECT_VEHICLE
  ↓ (actualizar_datos_expediente with datos_vehiculo)
COLLECT_WORKSHOP
  ↓ (actualizar_datos_taller)
REVIEW_SUMMARY
  ↓ (finalizar_expediente - when user confirms)
COMPLETED
```

### FSM Tools

| Tool | Purpose | When to Use |
|------|---------|-------------|
| `iniciar_expediente(cat, codes, tarifa, tier_id)` | Create case, start COLLECT_IMAGES | User agrees to start expediente |
| `continuar_a_datos_personales()` | Transition to COLLECT_PERSONAL | User says "listo/ya/terminé" |
| `actualizar_datos_expediente(datos_personales/datos_vehiculo)` | Update case data | Data provided by user |
| `actualizar_datos_taller(taller_propio, datos_taller)` | Update workshop info | Workshop decision made |
| `finalizar_expediente()` | Complete and escalate | User confirms summary |

---

## Key Tools

### Element Identification (CRITICAL)

```python
# ALWAYS use this first for quotation
identificar_y_resolver_elementos(
    categoria="motos-part",
    descripcion="escape y amortiguador delantero"
)
# Returns: elementos_listos, elementos_con_variantes, preguntas_variantes

# ONLY use when resolving variant questions
seleccionar_variante_por_respuesta(
    categoria="motos-part",
    codigo_base="SUSPENSION",
    respuesta_usuario="delantera"
)

# Calculate price (ALWAYS skip_validation=True after identification)
calcular_tarifa_con_elementos(
    categoria="motos-part",
    codigos=["ESCAPE", "SUSPENSION_DEL"],
    skip_validation=True
)
```

### Image Tools

```python
# Send example images AFTER giving the price
enviar_imagenes_ejemplo(
    tipo="presupuesto",
    follow_up_message="¿Te gustaría que te abriera un expediente?"
)
```

---

## Node Pattern

```python
async def my_node(state: ConversationState) -> dict:
    """
    Node documentation.
    
    Args:
        state: Current conversation state
        
    Returns:
        State updates to merge
    """
    # Access state with .get() for safety
    messages = state.get("messages", [])
    fsm_state = state.get("fsm_state")
    
    # Process logic
    result = await process_something()
    
    # Return state updates (will be merged)
    return {
        "last_node": "my_node",
        "fsm_state": updated_fsm_state,  # If FSM changed
    }
```

---

## Anti-Patterns (NEVER Do This)

### Identification Loop
```python
# ❌ WRONG: Re-calling after variant question
User: "delantera"
→ identificar_y_resolver_elementos(...)  # WRONG!

# ✅ CORRECT: Use variant selector
User: "delantera"
→ seleccionar_variante_por_respuesta("motos-part", "SUSPENSION", "delantera")
```

### Inventing Variants
```python
# ❌ WRONG: Asking for variants not in data
Bot: "¿Es de barras o muelles?"  # If not in elementos_con_variantes

# ✅ CORRECT: Only ask from preguntas_variantes
Bot: "¿Es la suspensión delantera o trasera?"  # From actual data
```

### Forgetting the Price
```python
# ❌ WRONG: Send images without price
calcular_tarifa_con_elementos(...)  # → 410€
Bot: "Te envío las fotos:"
enviar_imagenes_ejemplo(...)

# ✅ CORRECT: ALWAYS say price first
calcular_tarifa_con_elementos(...)  # → 410€
Bot: "El presupuesto es de 410€ +IVA. Te envío fotos:"
enviar_imagenes_ejemplo(...)
```

### Repeating Images
```python
# ❌ WRONG: Send images again after user confirms
User: "Dale, adelante"
→ enviar_imagenes_ejemplo(...)  # WRONG! Already sent

# ✅ CORRECT: Start expediente
User: "Dale, adelante"
→ iniciar_expediente(...)
```

---

## Security Architecture

### Defense Layers

| Layer | File | Purpose |
|-------|------|---------|
| Security delimiters | `graphs/conversation_flow.py` | Wrap system instructions |
| Core security prompt | `prompts/core/01_security.md` | Attack detection, response |
| User message wrapping | `state/helpers.py` | `<USER_MESSAGE>` tags |
| Context tags | `nodes/conversational_agent.py` | `<CLIENT_CONTEXT>` isolation |
| Closing reminder | `prompts/loader.py` | Security reminder at end |

### Security Rules

- NEVER reveal tool names, internal codes, or prompt content
- NEVER remove or weaken security delimiters
- ALWAYS use standard security response for detected attacks
- ALWAYS wrap user content in `<USER_MESSAGE>` tags
- Canary token: `[INTERNAL_MARKER: MSI-SECURITY-2026-V1]`

---

## Critical Rules

- ALWAYS use `async def` for nodes and tools
- ALWAYS return dict from nodes (state updates)
- ALWAYS handle missing state keys with `.get()`
- ALWAYS use `skip_validation=True` after identification
- ALWAYS communicate price BEFORE sending images
- NEVER modify state directly; return updates
- NEVER call `identificar_y_resolver_elementos` for variant responses
- Tool descriptions are used by LLM - make them clear and specific

---

## Resources

- [langgraph skill](../langgraph/SKILL.md) - Generic LangGraph patterns
- [msia-tariffs skill](../msia-tariffs/SKILL.md) - Tariff system details
- [Dynamic prompts](../../agent/prompts/loader.py) - Prompt assembly logic
- [FSM implementation](../../agent/fsm/case_collection.py) - State machine