Skip to content

Index

ChatResponseMode #

Bases: str, Enum

Flag toggling waiting/streaming in Agent._chat.

Source code in llama-index-core/llama_index/core/chat_engine/types.py
39
40
41
42
43
class ChatResponseMode(str, Enum):
    """Flag toggling waiting/streaming in `Agent._chat`."""

    WAIT = "wait"
    STREAM = "stream"

AgentChatResponse dataclass #

Agent chat response.

Source code in llama-index-core/llama_index/core/chat_engine/types.py
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@dataclass
class AgentChatResponse:
    """Agent chat response."""

    response: str = ""
    sources: List[ToolOutput] = field(default_factory=list)
    source_nodes: List[NodeWithScore] = field(default_factory=list)
    is_dummy_stream: bool = False

    def __post_init__(self) -> None:
        if self.sources and not self.source_nodes:
            for tool_output in self.sources:
                if isinstance(tool_output.raw_output, (Response, StreamingResponse)):
                    self.source_nodes.extend(tool_output.raw_output.source_nodes)

    def __str__(self) -> str:
        return self.response

    @property
    def response_gen(self) -> Generator[str, None, None]:
        """Used for fake streaming, i.e. with tool outputs."""
        if not self.is_dummy_stream:
            raise ValueError(
                "response_gen is only available for streaming responses. "
                "Set is_dummy_stream=True if you still want a generator."
            )

        for token in self.response.split(" "):
            yield token + " "
            time.sleep(0.1)

    async def async_response_gen(self) -> AsyncGenerator[str, None]:
        """Used for fake streaming, i.e. with tool outputs."""
        if not self.is_dummy_stream:
            raise ValueError(
                "response_gen is only available for streaming responses. "
                "Set is_dummy_stream=True if you still want a generator."
            )

        for token in self.response.split(" "):
            yield token + " "
            await asyncio.sleep(0.1)

response_gen property #

response_gen: Generator[str, None, None]

Used for fake streaming, i.e. with tool outputs.

async_response_gen async #

async_response_gen() -> AsyncGenerator[str, None]

Used for fake streaming, i.e. with tool outputs.

Source code in llama-index-core/llama_index/core/chat_engine/types.py
77
78
79
80
81
82
83
84
85
86
87
async def async_response_gen(self) -> AsyncGenerator[str, None]:
    """Used for fake streaming, i.e. with tool outputs."""
    if not self.is_dummy_stream:
        raise ValueError(
            "response_gen is only available for streaming responses. "
            "Set is_dummy_stream=True if you still want a generator."
        )

    for token in self.response.split(" "):
        yield token + " "
        await asyncio.sleep(0.1)

StreamingAgentChatResponse dataclass #

Streaming chat response to user and writing to chat history.

Source code in llama-index-core/llama_index/core/chat_engine/types.py
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
@dataclass
class StreamingAgentChatResponse:
    """Streaming chat response to user and writing to chat history."""

    response: str = ""
    sources: List[ToolOutput] = field(default_factory=list)
    chat_stream: Optional[ChatResponseGen] = None
    achat_stream: Optional[ChatResponseAsyncGen] = None
    source_nodes: List[NodeWithScore] = field(default_factory=list)
    _unformatted_response: str = ""
    _queue: queue.Queue = field(default_factory=queue.Queue)
    _aqueue: Optional[asyncio.Queue] = None
    # flag when chat message is a function call
    _is_function: Optional[bool] = None
    # flag when processing done
    _is_done = False
    # signal when a new item is added to the queue
    _new_item_event: Optional[asyncio.Event] = None
    # NOTE: async code uses two events rather than one since it yields
    # control when waiting for queue item
    # signal when the OpenAI functions stop executing
    _is_function_false_event: Optional[asyncio.Event] = None
    # signal when an OpenAI function is being executed
    _is_function_not_none_thread_event: Event = field(default_factory=Event)

    def __post_init__(self) -> None:
        if self.sources and not self.source_nodes:
            for tool_output in self.sources:
                if isinstance(tool_output.raw_output, (Response, StreamingResponse)):
                    self.source_nodes.extend(tool_output.raw_output.source_nodes)

    def __str__(self) -> str:
        if self._is_done and not self._queue.empty() and not self._is_function:
            while self._queue.queue:
                delta = self._queue.queue.popleft()
                self._unformatted_response += delta
            self.response = self._unformatted_response.strip()
        return self.response

    def _ensure_async_setup(self) -> None:
        if self._aqueue is None:
            self._aqueue = asyncio.Queue()
        if self._new_item_event is None:
            self._new_item_event = asyncio.Event()
        if self._is_function_false_event is None:
            self._is_function_false_event = asyncio.Event()

    def put_in_queue(self, delta: Optional[str]) -> None:
        self._queue.put_nowait(delta)
        self._is_function_not_none_thread_event.set()

    def aput_in_queue(self, delta: Optional[str]) -> None:
        self._aqueue.put_nowait(delta)
        self._new_item_event.set()

    @dispatcher.span
    def write_response_to_history(
        self,
        memory: BaseMemory,
        on_stream_end_fn: Optional[callable] = None,
        raise_error: bool = False,
    ) -> None:
        if self.chat_stream is None:
            raise ValueError(
                "chat_stream is None. Cannot write to history without chat_stream."
            )
        dispatch_event = dispatcher.get_dispatch_event()

        # try/except to prevent hanging on error
        dispatch_event(StreamChatStartEvent())
        try:
            final_text = ""
            for chat in self.chat_stream:
                self._is_function = is_function(chat.message)
                if chat.delta:
                    dispatch_event(
                        StreamChatDeltaReceivedEvent(
                            delta=chat.delta,
                        )
                    )
                    self.put_in_queue(chat.delta)
                final_text += chat.delta or ""
            if self._is_function is not None:  # if loop has gone through iteration
                # NOTE: this is to handle the special case where we consume some of the
                # chat stream, but not all of it (e.g. in react agent)
                chat.message.content = final_text.strip()  # final message
                memory.put(chat.message)
        except Exception as e:
            dispatch_event(StreamChatErrorEvent())
            if not raise_error:
                logger.warning(
                    f"Encountered exception writing response to history: {e}"
                )
            else:
                raise
        dispatch_event(StreamChatEndEvent())

        self._is_done = True

        # This act as is_done events for any consumers waiting
        self._is_function_not_none_thread_event.set()
        if on_stream_end_fn is not None and not self._is_function:
            on_stream_end_fn()

    @dispatcher.span
    async def awrite_response_to_history(
        self,
        memory: BaseMemory,
        on_stream_end_fn: Optional[callable] = None,
    ) -> None:
        self._ensure_async_setup()
        dispatch_event = dispatcher.get_dispatch_event()

        if self.achat_stream is None:
            raise ValueError(
                "achat_stream is None. Cannot asynchronously write to "
                "history without achat_stream."
            )

        # try/except to prevent hanging on error
        dispatch_event(StreamChatStartEvent())
        try:
            final_text = ""
            async for chat in self.achat_stream:
                self._is_function = is_function(chat.message)
                if chat.delta:
                    dispatch_event(
                        StreamChatDeltaReceivedEvent(
                            delta=chat.delta,
                        )
                    )
                    self.aput_in_queue(chat.delta)
                final_text += chat.delta or ""
                self._new_item_event.set()
                if self._is_function is False:
                    self._is_function_false_event.set()
            if self._is_function is not None:  # if loop has gone through iteration
                # NOTE: this is to handle the special case where we consume some of the
                # chat stream, but not all of it (e.g. in react agent)
                chat.message.content = final_text.strip()  # final message
                memory.put(chat.message)
        except Exception as e:
            dispatch_event(StreamChatErrorEvent())
            logger.warning(f"Encountered exception writing response to history: {e}")
        dispatch_event(StreamChatEndEvent())
        self._is_done = True

        # These act as is_done events for any consumers waiting
        self._is_function_false_event.set()
        self._new_item_event.set()
        if on_stream_end_fn is not None and not self._is_function:
            on_stream_end_fn()

    @property
    def response_gen(self) -> Generator[str, None, None]:
        while not self._is_done or not self._queue.empty():
            try:
                delta = self._queue.get(block=False)
                self._unformatted_response += delta
                yield delta
            except queue.Empty:
                # Queue is empty, but we're not done yet. Sleep for 0 secs to release the GIL and allow other threads to run.
                time.sleep(0)
        self.response = self._unformatted_response.strip()

    async def async_response_gen(self) -> AsyncGenerator[str, None]:
        self._ensure_async_setup()
        while True:
            if not self._aqueue.empty() or not self._is_done:
                try:
                    delta = await asyncio.wait_for(self._aqueue.get(), timeout=0.1)
                except asyncio.TimeoutError:
                    if self._is_done:
                        break
                    continue
                if delta is not None:
                    self._unformatted_response += delta
                    yield delta
            else:
                break
        self.response = self._unformatted_response.strip()

    def print_response_stream(self) -> None:
        for token in self.response_gen:
            print(token, end="", flush=True)

    async def aprint_response_stream(self) -> None:
        async for token in self.async_response_gen():
            print(token, end="", flush=True)

BaseChatEngine #

Bases: ABC

Base Chat Engine.

Source code in llama-index-core/llama_index/core/chat_engine/types.py
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
class BaseChatEngine(ABC):
    """Base Chat Engine."""

    @abstractmethod
    def reset(self) -> None:
        """Reset conversation state."""

    @abstractmethod
    def chat(
        self, message: str, chat_history: Optional[List[ChatMessage]] = None
    ) -> AGENT_CHAT_RESPONSE_TYPE:
        """Main chat interface."""

    @abstractmethod
    def stream_chat(
        self, message: str, chat_history: Optional[List[ChatMessage]] = None
    ) -> StreamingAgentChatResponse:
        """Stream chat interface."""

    @abstractmethod
    async def achat(
        self, message: str, chat_history: Optional[List[ChatMessage]] = None
    ) -> AGENT_CHAT_RESPONSE_TYPE:
        """Async version of main chat interface."""

    @abstractmethod
    async def astream_chat(
        self, message: str, chat_history: Optional[List[ChatMessage]] = None
    ) -> StreamingAgentChatResponse:
        """Async version of main chat interface."""

    def chat_repl(self) -> None:
        """Enter interactive chat REPL."""
        print("===== Entering Chat REPL =====")
        print('Type "exit" to exit.\n')
        self.reset()
        message = input("Human: ")
        while message != "exit":
            response = self.chat(message)
            print(f"Assistant: {response}\n")
            message = input("Human: ")

    def streaming_chat_repl(self) -> None:
        """Enter interactive chat REPL with streaming responses."""
        print("===== Entering Chat REPL =====")
        print('Type "exit" to exit.\n')
        self.reset()
        message = input("Human: ")
        while message != "exit":
            response = self.stream_chat(message)
            print("Assistant: ", end="", flush=True)
            response.print_response_stream()
            print("\n")
            message = input("Human: ")

    @property
    @abstractmethod
    def chat_history(self) -> List[ChatMessage]:
        pass

reset abstractmethod #

reset() -> None

Reset conversation state.

Source code in llama-index-core/llama_index/core/chat_engine/types.py
287
288
289
@abstractmethod
def reset(self) -> None:
    """Reset conversation state."""

chat abstractmethod #

chat(message: str, chat_history: Optional[List[ChatMessage]] = None) -> AGENT_CHAT_RESPONSE_TYPE

Main chat interface.

Source code in llama-index-core/llama_index/core/chat_engine/types.py
291
292
293
294
295
@abstractmethod
def chat(
    self, message: str, chat_history: Optional[List[ChatMessage]] = None
) -> AGENT_CHAT_RESPONSE_TYPE:
    """Main chat interface."""

stream_chat abstractmethod #

stream_chat(message: str, chat_history: Optional[List[ChatMessage]] = None) -> StreamingAgentChatResponse

Stream chat interface.

Source code in llama-index-core/llama_index/core/chat_engine/types.py
297
298
299
300
301
@abstractmethod
def stream_chat(
    self, message: str, chat_history: Optional[List[ChatMessage]] = None
) -> StreamingAgentChatResponse:
    """Stream chat interface."""

achat abstractmethod async #

achat(message: str, chat_history: Optional[List[ChatMessage]] = None) -> AGENT_CHAT_RESPONSE_TYPE

Async version of main chat interface.

Source code in llama-index-core/llama_index/core/chat_engine/types.py
303
304
305
306
307
@abstractmethod
async def achat(
    self, message: str, chat_history: Optional[List[ChatMessage]] = None
) -> AGENT_CHAT_RESPONSE_TYPE:
    """Async version of main chat interface."""

astream_chat abstractmethod async #

astream_chat(message: str, chat_history: Optional[List[ChatMessage]] = None) -> StreamingAgentChatResponse

Async version of main chat interface.

Source code in llama-index-core/llama_index/core/chat_engine/types.py
309
310
311
312
313
@abstractmethod
async def astream_chat(
    self, message: str, chat_history: Optional[List[ChatMessage]] = None
) -> StreamingAgentChatResponse:
    """Async version of main chat interface."""

chat_repl #

chat_repl() -> None

Enter interactive chat REPL.

Source code in llama-index-core/llama_index/core/chat_engine/types.py
315
316
317
318
319
320
321
322
323
324
def chat_repl(self) -> None:
    """Enter interactive chat REPL."""
    print("===== Entering Chat REPL =====")
    print('Type "exit" to exit.\n')
    self.reset()
    message = input("Human: ")
    while message != "exit":
        response = self.chat(message)
        print(f"Assistant: {response}\n")
        message = input("Human: ")

streaming_chat_repl #

streaming_chat_repl() -> None

Enter interactive chat REPL with streaming responses.

Source code in llama-index-core/llama_index/core/chat_engine/types.py
326
327
328
329
330
331
332
333
334
335
336
337
def streaming_chat_repl(self) -> None:
    """Enter interactive chat REPL with streaming responses."""
    print("===== Entering Chat REPL =====")
    print('Type "exit" to exit.\n')
    self.reset()
    message = input("Human: ")
    while message != "exit":
        response = self.stream_chat(message)
        print("Assistant: ", end="", flush=True)
        response.print_response_stream()
        print("\n")
        message = input("Human: ")

ChatMode #

Bases: str, Enum

Chat Engine Modes.

Source code in llama-index-core/llama_index/core/chat_engine/types.py
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
class ChatMode(str, Enum):
    """Chat Engine Modes."""

    SIMPLE = "simple"
    """Corresponds to `SimpleChatEngine`.

    Chat with LLM, without making use of a knowledge base.
    """

    CONDENSE_QUESTION = "condense_question"
    """Corresponds to `CondenseQuestionChatEngine`.

    First generate a standalone question from conversation context and last message,
    then query the query engine for a response.
    """

    CONTEXT = "context"
    """Corresponds to `ContextChatEngine`.

    First retrieve text from the index using the user's message, then use the context
    in the system prompt to generate a response.
    """

    CONDENSE_PLUS_CONTEXT = "condense_plus_context"
    """Corresponds to `CondensePlusContextChatEngine`.

    First condense a conversation and latest user message to a standalone question.
    Then build a context for the standalone question from a retriever,
    Then pass the context along with prompt and user message to LLM to generate a response.
    """

    REACT = "react"
    """Corresponds to `ReActAgent`.

    Use a ReAct agent loop with query engine tools.
    """

    OPENAI = "openai"
    """Corresponds to `OpenAIAgent`.

    Use an OpenAI function calling agent loop.

    NOTE: only works with OpenAI models that support function calling API.
    """

    BEST = "best"
    """Select the best chat engine based on the current LLM.

    Corresponds to `OpenAIAgent` if using an OpenAI model that supports
    function calling API, otherwise, corresponds to `ReActAgent`.
    """

SIMPLE class-attribute instance-attribute #

SIMPLE = 'simple'

Corresponds to SimpleChatEngine.

Chat with LLM, without making use of a knowledge base.

CONDENSE_QUESTION class-attribute instance-attribute #

CONDENSE_QUESTION = 'condense_question'

Corresponds to CondenseQuestionChatEngine.

First generate a standalone question from conversation context and last message, then query the query engine for a response.

CONTEXT class-attribute instance-attribute #

CONTEXT = 'context'

Corresponds to ContextChatEngine.

First retrieve text from the index using the user's message, then use the context in the system prompt to generate a response.

CONDENSE_PLUS_CONTEXT class-attribute instance-attribute #

CONDENSE_PLUS_CONTEXT = 'condense_plus_context'

Corresponds to CondensePlusContextChatEngine.

First condense a conversation and latest user message to a standalone question. Then build a context for the standalone question from a retriever, Then pass the context along with prompt and user message to LLM to generate a response.

REACT class-attribute instance-attribute #

REACT = 'react'

Corresponds to ReActAgent.

Use a ReAct agent loop with query engine tools.

OPENAI class-attribute instance-attribute #

OPENAI = 'openai'

Corresponds to OpenAIAgent.

Use an OpenAI function calling agent loop.

NOTE: only works with OpenAI models that support function calling API.

BEST class-attribute instance-attribute #

BEST = 'best'

Select the best chat engine based on the current LLM.

Corresponds to OpenAIAgent if using an OpenAI model that supports function calling API, otherwise, corresponds to ReActAgent.

is_function #

is_function(message: ChatMessage) -> bool

Utility for ChatMessage responses from OpenAI models.

Source code in llama-index-core/llama_index/core/chat_engine/types.py
34
35
36
def is_function(message: ChatMessage) -> bool:
    """Utility for ChatMessage responses from OpenAI models."""
    return "tool_calls" in message.additional_kwargs