How to run a chat engine in LlamaIndex

Conversational retrieval needs both relevant indexed text and the turns that give an ambiguous follow-up its meaning. A LlamaIndex chat engine combines those inputs so a question such as “Where should that ticket be escalated?” can refer to a ticket identified earlier in the conversation.

The as_chat_engine() API creates a stateful interface from an existing index. Its context mode retrieves nodes for each message, supplies their text to the language model, and retains user and assistant messages in the engine's chat history.

The local run uses MockEmbedding plus a deterministic CustomLLM adapter, which keeps the transcript repeatable without an API key or model server. The adapter extracts values from the retrieved runbook and earlier user turns; isolated controls prove that neither input can produce the ticket-specific follow-up answer alone.

1. Install LlamaIndex core in the active project environment.

   $ python3 -m pip install llama-index-core

2. Create chat_engine_run.py with the imports and prompt-parsing helpers.

   import re
   from typing import Any
    
   from llama_index.core import Document, Settings, VectorStoreIndex
   from llama_index.core.embeddings import MockEmbedding
   from llama_index.core.llms import CompletionResponse, CustomLLM, LLMMetadata
   from llama_index.core.llms.callbacks import llm_completion_callback
    
    
   NO_MATCH = "No matching runbook entry was found."
   CONTEXT_MARKER = "--------------------\n"
    
    
   def prompt_context(prompt: str) -> str:
       parts = prompt.split(CONTEXT_MARKER, 2)
       return parts[1].strip() if len(parts) == 3 else ""
    
    
   def prompt_user_messages(prompt: str) -> list[str]:
       return re.findall(r"^user: (.+)$", prompt, flags=re.MULTILINE)

3. Add the context-dependent model below the prompt helpers.

   class SupportRunbookLLM(CustomLLM):
       @property
       def metadata(self) -> LLMMetadata:
           return LLMMetadata(
               context_window=2048,
               num_output=128,
               model_name="support-runbook-llm",
           )
    
       def answer_from_prompt(self, prompt: str) -> str:
           context = prompt_context(prompt)
           user_messages = prompt_user_messages(prompt)
           if not user_messages:
               return NO_MATCH
    
           question = user_messages[-1]
           if "who owns" in question.lower():
               ticket = re.search(r"\bticket (\d+)\b", question, flags=re.IGNORECASE)
               if ticket:
                   owner = re.search(
                       rf"ticket {re.escape(ticket.group(1))} belongs to ([A-Z][A-Za-z-]+)\.",
                       context,
                   )
                   if owner:
                       return f"Ticket {ticket.group(1)} belongs to {owner.group(1)}."
    
           if "where should" in question.lower() and "escalated" in question.lower():
               earlier_turns = "\n".join(user_messages[:-1])
               ticket = re.search(
                   r"\bticket (\d+)\b",
                   earlier_turns,
                   flags=re.IGNORECASE,
               )
               if ticket:
                   escalation = re.search(
                       rf"([A-Z][^.]*) for ticket {re.escape(ticket.group(1))} "
                       r"should be escalated to the ([a-z0-9-]+ queue)\.",
                       context,
                   )
                   if escalation:
                       return (
                           f"{escalation.group(1)} for ticket {ticket.group(1)} "
                           f"should be escalated to the {escalation.group(2)}."
                       )
    
           return NO_MATCH

   SupportRunbookLLM is test support for a repeatable local run, not a production language model.

4. Complete SupportRunbookLLM with its synchronous completion methods.

       @llm_completion_callback()
       def complete(
           self,
           prompt: str,
           formatted: bool = False,
           **kwargs: Any,
       ) -> CompletionResponse:
           return CompletionResponse(text=self.answer_from_prompt(prompt))
    
       @llm_completion_callback()
       def stream_complete(
           self,
           prompt: str,
           formatted: bool = False,
           **kwargs: Any,
       ):
           text = self.answer_from_prompt(prompt)
           running_text = ""
           for token in text.split():
               running_text = f"{running_text}{token} "
               yield CompletionResponse(text=running_text.strip(), delta=f"{token} ")

5. Append the local model settings and indexed support runbook after SupportRunbookLLM.

   Settings.llm = SupportRunbookLLM()
   Settings.embed_model = MockEmbedding(embed_dim=8)
    
   documents = [
       Document(
           text=(
               "Support runbook: ticket 7421 belongs to Maya. "
               "Billing chatbot answers for ticket 7421 should be escalated "
               "to the docs-review queue."
           )
       )
   ]
   index = VectorStoreIndex.from_documents(documents)

6. Append the positive conversation and isolated controls after the index.

   chat_engine = index.as_chat_engine(chat_mode="context", similarity_top_k=1)
   first_response = chat_engine.chat("Who owns ticket 7421?")
   follow_up_response = chat_engine.chat(
       "Where should that ticket's chatbot answers be escalated?"
   )
    
   missing_history_engine = index.as_chat_engine(chat_mode="context", similarity_top_k=1)
   missing_history_response = missing_history_engine.chat(
       "Where should that ticket's chatbot answers be escalated?"
   )
    
   unrelated_index = VectorStoreIndex.from_documents(
       [Document(text="Office support hours are 09:00 to 17:00 on weekdays.")]
   )
   missing_context_engine = unrelated_index.as_chat_engine(
       chat_mode="context",
       similarity_top_k=1,
   )
   missing_context_engine.chat("Who owns ticket 7421?")
   missing_context_response = missing_context_engine.chat(
       "Where should that ticket's chatbot answers be escalated?"
   )
    
   print("Q: Who owns ticket 7421?")
   print(f"A: {first_response}")
   print("Q: Where should that ticket's chatbot answers be escalated?")
   print(f"A: {follow_up_response}")
   print("stored messages:", len(chat_engine.chat_history))
   print("missing history:", missing_history_response)
   print("missing context:", missing_context_response)
    
   controls_blocked = (
       str(missing_history_response) == NO_MATCH
       and str(missing_context_response) == NO_MATCH
   )
   print("controls blocked:", controls_blocked)
    
   assert str(first_response) == "Ticket 7421 belongs to Maya."
   assert str(follow_up_response) == (
       "Billing chatbot answers for ticket 7421 should be escalated "
       "to the docs-review queue."
   )
   assert controls_blocked

7. Run chat_engine_run.py to check the context-backed answers and fail-capable controls.

   $ python3 chat_engine_run.py
   Q: Who owns ticket 7421?
   A: Ticket 7421 belongs to Maya.
   Q: Where should that ticket's chatbot answers be escalated?
   A: Billing chatbot answers for ticket 7421 should be escalated to the docs-review queue.
   stored messages: 4
   missing history: No matching runbook entry was found.
   missing context: No matching runbook entry was found.
   controls blocked: True

   The follow-up supplies no ticket number. The positive engine extracts ticket 7421 from an earlier user turn and the escalation queue from retrieved runbook text, while each control removes one of those required inputs and reaches the non-answer path.