Implementation:Infiniflow Ragflow ConversationService Save And Chat

Knowledge Sources	RAGFlow
Domains	RAG, Conversational_AI
Last Updated	2026-02-12 06:00 GMT

Overview

Concrete tool for creating conversations and streaming RAG-powered responses provided by RAGFlow ConversationService and async_chat.

Description

ConversationService.save creates a Conversation record linked to a Dialog. The async_chat async generator in DialogService orchestrates retrieval, prompt building, LLM streaming, and citation insertion, yielding SSE events.

Usage

Called via POST /v1/conversation/set (create) and POST /v1/conversation/completion (chat SSE).

Code Reference

Source Location

• Repository: ragflow
• File: api/apps/conversation_app.py (L39-78 create, L168-251 completion), api/db/services/dialog_service.py (L411-733 async_chat)

Signature

class ConversationService(CommonService):
    model = Conversation

    @classmethod
    def save(cls, **kwargs) -> Conversation:
        """Create a conversation.

        Args:
            dialog_id: str - Parent dialog ID.
            name: str - Conversation name.
            is_new: bool - Whether this is a new conversation.
        """

async def async_chat(dialog, messages, stream=True, **kwargs):
    """RAG chat pipeline as async generator.

    Args:
        dialog: Dialog - Chat application config.
        messages: list[dict] - Message history [{role, content}].
        stream: bool - Enable streaming.

    Yields:
        dict - SSE events with answer chunks and references.
    """

Import

from api.db.services.conversation_service import ConversationService
from api.db.services.dialog_service import async_chat

I/O Contract

Inputs

Name	Type	Required	Description
conversation_id	str	Yes	Conversation ID (for chat)
messages	list[dict]	Yes	Message history with role/content

Outputs

Name	Type	Description
SSE stream	text/event-stream	data:{code:0, data:{answer, reference:{chunks, doc_aggs}, final:bool}}

Usage Examples

import requests
import json

# Create conversation
url = "http://localhost:9380/v1/conversation/set"
payload = {"dialog_id": "dialog-uuid-123", "name": "Session 1"}
response = requests.post(url, json=payload, headers=headers)
conv_id = response.json()["data"]["id"]

# Chat with SSE streaming
url = "http://localhost:9380/v1/conversation/completion"
payload = {
    "conversation_id": conv_id,
    "messages": [{"role": "user", "content": "What is the refund policy?"}]
}
with requests.post(url, json=payload, headers=headers, stream=True) as r:
    for line in r.iter_lines():
        if line.startswith(b"data:"):
            data = json.loads(line[5:])
            if data["data"] is True:
                break
            print(data["data"].get("answer", ""), end="")

Related Pages

Implements Principle

• Principle:Infiniflow_Ragflow_Conversation_Management