LangChain is a framework for developing applications powered by language models. It provides tools for building chains, agents, memory systems, and RAG applications.

How do I integrate LangChain with Flutter?

Create a backend API using Python/FastAPI with LangChain, then connect your Flutter app via HTTP/WebSocket. Never expose LLM API keys in your Flutter app.

What is RAG in LangChain?

RAG (Retrieval Augmented Generation) combines retrieval systems with LLMs. It first retrieves relevant documents from a knowledge base, then uses them as context to generate accurate responses.

Creating AI Chatbots with Flutter and LangChain

Introduction to LangChain
Understanding LangChain Architecture
Setting Up LangChain with Flutter
Building Your First Chatbot
Implementing Memory Systems
Building RAG Applications
Creating Intelligent Agents
Function Calling and Tools
Real-Time Streaming Responses
Production Deployment

Introduction to LangChain

LangChain has revolutionized the way developers build applications with Large Language Models (LLMs). As a powerful framework for developing applications powered by language models, LangChain provides the building blocks needed to create sophisticated AI applications. When combined with Flutter's cross-platform capabilities, you can build intelligent chatbots that work seamlessly across mobile, web, and desktop platforms.

In this comprehensive guide, we'll explore how to leverage LangChain's powerful features to build advanced AI chatbots in Flutter. From simple conversational interfaces to complex AI agents capable of reasoning and using tools, you'll learn everything needed to create production-ready applications.

Understanding LangChain Architecture

Before diving into implementation, let's understand the core components of LangChain and how they work together to create intelligent applications.

LangChain Core Components

┌─────────────────────────────────────────────────┐
│                  Application Layer               │
│  ┌───────────┐  ┌──────────┐  ┌──────────────┐ │
│  │  Agents   │  │  Chains  │  │   Callbacks  │ │
│  └───────────┘  └──────────┘  └──────────────┘ │
├─────────────────────────────────────────────────┤
│                 Core Components                  │
│  ┌───────────┐  ┌──────────┐  ┌──────────────┐ │
│  │   LLMs    │  │  Memory  │  │   Prompts    │ │
│  └───────────┘  └──────────┘  └──────────────┘ │
├─────────────────────────────────────────────────┤
│                  Data Layer                      │
│  ┌───────────┐  ┌──────────┐  ┌──────────────┐ │
│  │ Embeddings│  │  Vector  │  │  Retrievers  │ │
│  │           │  │  Stores  │  │              │ │
│  └───────────┘  └──────────┘  └──────────────┘ │
└─────────────────────────────────────────────────┘

Key Components Explained

LLMs: Language models like GPT-4, Claude, or open-source alternatives
Chains: Combine multiple LLM calls and processing steps
Agents: LLMs that can use tools and make decisions
Memory: Maintain conversation context across interactions
Embeddings: Convert text to vector representations
Vector Stores: Store and retrieve embedded documents efficiently
Retrievers: Fetch relevant documents for context

Setting Up LangChain with Flutter

Let's start by setting up a Flutter project with LangChain integration. We'll create a service layer that bridges Flutter and LangChain's Python backend.

Project Structure

flutter_langchain_app/
├── lib/
│   ├── main.dart
│   ├── services/
│   │   ├── langchain_service.dart
│   │   ├── chat_service.dart
│   │   └── vector_store_service.dart
│   ├── models/
│   │   ├── message.dart
│   │   └── conversation.dart
│   └── screens/
│       ├── chat_screen.dart
│       └── settings_screen.dart
├── backend/
│   ├── app.py
│   ├── langchain_handler.py
│   └── requirements.txt
└── pubspec.yaml

Backend Setup (Python FastAPI)

# backend/requirements.txt
langchain==0.1.0
langchain-openai==0.0.2
langchain-community==0.0.10
fastapi==0.104.1
uvicorn==0.24.0
python-dotenv==1.0.0
chromadb==0.4.18
tiktoken==0.5.1
pydantic==2.5.0

# backend/app.py
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List, Optional, Dict, Any
import asyncio
from langchain_handler import LangChainHandler

app = FastAPI()

# Enable CORS for Flutter web
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Initialize LangChain handler
handler = LangChainHandler()

class ChatRequest(BaseModel):
    message: str
    conversation_id: str
    context: Optional[Dict[str, Any]] = None
    stream: bool = False

class ChatResponse(BaseModel):
    response: str
    conversation_id: str
    metadata: Optional[Dict[str, Any]] = None

@app.post("/chat")
async def chat(request: ChatRequest) -> ChatResponse:
    try:
        response = await handler.process_message(
            message=request.message,
            conversation_id=request.conversation_id,
            context=request.context,
            stream=request.stream
        )
        
        return ChatResponse(
            response=response["text"],
            conversation_id=request.conversation_id,
            metadata=response.get("metadata")
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/embeddings")
async def create_embeddings(texts: List[str]) -> List[List[float]]:
    return await handler.create_embeddings(texts)

@app.post("/rag/query")
async def rag_query(query: str, top_k: int = 5) -> Dict[str, Any]:
    return await handler.rag_query(query, top_k)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

LangChain Handler Implementation

# backend/langchain_handler.py
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationalRetrievalChain
from langchain.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import TextLoader
from langchain.agents import initialize_agent, Tool, AgentType
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from typing import Dict, List, Any, Optional
import os
from dotenv import load_dotenv

load_dotenv()

class LangChainHandler:
    def __init__(self):
        self.llm = ChatOpenAI(
            temperature=0.7,
            model="gpt-4-turbo-preview",
            streaming=True,
            callbacks=[StreamingStdOutCallbackHandler()]
        )
        
        self.embeddings = OpenAIEmbeddings()
        self.memories = {}  # Store conversation memories
        self.vector_store = None
        self._initialize_vector_store()
        
    def _initialize_vector_store(self):
        """Initialize vector store for RAG"""
        # In production, load from persistent storage
        self.vector_store = Chroma(
            embedding_function=self.embeddings,
            persist_directory="./chroma_db"
        )
    
    def get_or_create_memory(self, conversation_id: str) -> ConversationBufferWindowMemory:
        """Get or create conversation memory"""
        if conversation_id not in self.memories:
            self.memories[conversation_id] = ConversationBufferWindowMemory(
                k=10,  # Keep last 10 exchanges
                return_messages=True,
                memory_key="chat_history"
            )
        return self.memories[conversation_id]
    
    async def process_message(
        self,
        message: str,
        conversation_id: str,
        context: Optional[Dict[str, Any]] = None,
        stream: bool = False
    ) -> Dict[str, Any]:
        """Process a chat message with memory"""
        memory = self.get_or_create_memory(conversation_id)
        
        # Create conversational chain
        chain = ConversationalRetrievalChain.from_llm(
            llm=self.llm,
            retriever=self.vector_store.as_retriever(
                search_kwargs={"k": 3}
            ),
            memory=memory,
            return_source_documents=True,
            verbose=True
        )
        
        # Process the message
        result = await chain.ainvoke({"question": message})
        
        return {
            "text": result["answer"],
            "source_documents": [
                {
                    "content": doc.page_content,
                    "metadata": doc.metadata
                }
                for doc in result.get("source_documents", [])
            ],
            "metadata": {
                "model": self.llm.model_name,
                "temperature": self.llm.temperature
            }
        }
    
    async def create_embeddings(self, texts: List[str]) -> List[List[float]]:
        """Create embeddings for given texts"""
        return await self.embeddings.aembed_documents(texts)
    
    async def rag_query(self, query: str, top_k: int = 5) -> Dict[str, Any]:
        """Perform RAG query"""
        docs = self.vector_store.similarity_search(query, k=top_k)
        
        return {
            "query": query,
            "results": [
                {
                    "content": doc.page_content,
                    "metadata": doc.metadata,
                    "score": doc.metadata.get("score", 0)
                }
                for doc in docs
            ]
        }
    
    def add_documents(self, documents: List[str], metadata: Optional[List[Dict]] = None):
        """Add documents to vector store"""
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200
        )
        
        texts = text_splitter.create_documents(documents, metadatas=metadata)
        self.vector_store.add_documents(texts)

Flutter Service Layer

// lib/services/langchain_service.dart
import 'dart:convert';
import 'package:http/http.dart' as http;
import 'package:flutter_dotenv/flutter_dotenv.dart';

class LangChainService {
  late final String baseUrl;
  final Map headers = {
    'Content-Type': 'application/json',
  };
  
  LangChainService() {
    baseUrl = dotenv.env['LANGCHAIN_API_URL'] ?? 'http://localhost:8000';
  }
  
  Future sendMessage({
    required String message,
    required String conversationId,
    Map? context,
    bool stream = false,
  }) async {
    try {
      final response = await http.post(
        Uri.parse('$baseUrl/chat'),
        headers: headers,
        body: json.encode({
          'message': message,
          'conversation_id': conversationId,
          'context': context,
          'stream': stream,
        }),
      );
      
      if (response.statusCode == 200) {
        final data = json.decode(response.body);
        return ChatResponse.fromJson(data);
      } else {
        throw Exception('Failed to send message: ${response.body}');
      }
    } catch (e) {
      throw Exception('Network error: $e');
    }
  }
  
  Future>> createEmbeddings(List texts) async {
    final response = await http.post(
      Uri.parse('$baseUrl/embeddings'),
      headers: headers,
      body: json.encode(texts),
    );
    
    if (response.statusCode == 200) {
      final List data = json.decode(response.body);
      return data.map((e) => List.from(e)).toList();
    } else {
      throw Exception('Failed to create embeddings');
    }
  }
  
  Future performRagQuery(String query, {int topK = 5}) async {
    final response = await http.post(
      Uri.parse('$baseUrl/rag/query'),
      headers: headers,
      body: json.encode({
        'query': query,
        'top_k': topK,
      }),
    );
    
    if (response.statusCode == 200) {
      return RagQueryResult.fromJson(json.decode(response.body));
    } else {
      throw Exception('Failed to perform RAG query');
    }
  }
  
  Stream streamMessage({
    required String message,
    required String conversationId,
  }) async* {
    final request = http.Request(
      'POST',
      Uri.parse('$baseUrl/chat/stream'),
    );
    
    request.headers.addAll(headers);
    request.body = json.encode({
      'message': message,
      'conversation_id': conversationId,
      'stream': true,
    });
    
    final streamedResponse = await request.send();
    
    await for (final chunk in streamedResponse.stream.transform(utf8.decoder)) {
      yield chunk;
    }
  }
}

class ChatResponse {
  final String response;
  final String conversationId;
  final Map? metadata;
  
  ChatResponse({
    required this.response,
    required this.conversationId,
    this.metadata,
  });
  
  factory ChatResponse.fromJson(Map json) {
    return ChatResponse(
      response: json['response'],
      conversationId: json['conversation_id'],
      metadata: json['metadata'],
    );
  }
}

class RagQueryResult {
  final String query;
  final List results;
  
  RagQueryResult({required this.query, required this.results});
  
  factory RagQueryResult.fromJson(Map json) {
    return RagQueryResult(
      query: json['query'],
      results: (json['results'] as List)
          .map((e) => RagDocument.fromJson(e))
          .toList(),
    );
  }
}

class RagDocument {
  final String content;
  final Map metadata;
  final double score;
  
  RagDocument({
    required this.content,
    required this.metadata,
    required this.score,
  });
  
  factory RagDocument.fromJson(Map json) {
    return RagDocument(
      content: json['content'],
      metadata: json['metadata'],
      score: json['score'].toDouble(),
    );
  }
}

Building Your First Chatbot

Now let's create a basic chatbot interface that leverages LangChain's conversational capabilities.

// lib/screens/chat_screen.dart
import 'package:flutter/material.dart';
import 'package:uuid/uuid.dart';
import '../services/langchain_service.dart';
import '../models/message.dart';

class ChatScreen extends StatefulWidget {
  @override
  _ChatScreenState createState() => _ChatScreenState();
}

class _ChatScreenState extends State {
  final LangChainService _langChainService = LangChainService();
  final TextEditingController _controller = TextEditingController();
  final List _messages = [];
  final String _conversationId = Uuid().v4();
  bool _isLoading = false;
  
  @override
  void initState() {
    super.initState();
    _addWelcomeMessage();
  }
  
  void _addWelcomeMessage() {
    setState(() {
      _messages.add(Message(
        id: Uuid().v4(),
        text: 'Hello! I\'m your AI assistant powered by LangChain. How can I help you today?',
        isUser: false,
        timestamp: DateTime.now(),
      ));
    });
  }
  
  Future _sendMessage() async {
    if (_controller.text.trim().isEmpty) return;
    
    final userMessage = Message(
      id: Uuid().v4(),
      text: _controller.text,
      isUser: true,
      timestamp: DateTime.now(),
    );
    
    setState(() {
      _messages.add(userMessage);
      _isLoading = true;
    });
    
    _controller.clear();
    
    try {
      final response = await _langChainService.sendMessage(
        message: userMessage.text,
        conversationId: _conversationId,
      );
      
      setState(() {
        _messages.add(Message(
          id: Uuid().v4(),
          text: response.response,
          isUser: false,
          timestamp: DateTime.now(),
          metadata: response.metadata,
        ));
      });
    } catch (e) {
      _showError('Failed to get response: $e');
    } finally {
      setState(() {
        _isLoading = false;
      });
    }
  }
  
  void _showError(String message) {
    ScaffoldMessenger.of(context).showSnackBar(
      SnackBar(
        content: Text(message),
        backgroundColor: Colors.red,
      ),
    );
  }
  
  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: Text('LangChain AI Chat'),
        backgroundColor: Theme.of(context).colorScheme.inversePrimary,
        actions: [
          IconButton(
            icon: Icon(Icons.delete_outline),
            onPressed: () {
              setState(() {
                _messages.clear();
                _addWelcomeMessage();
              });
            },
            tooltip: 'Clear conversation',
          ),
        ],
      ),
      body: Column(
        children: [
          Expanded(
            child: ListView.builder(
              padding: EdgeInsets.all(16),
              itemCount: _messages.length,
              itemBuilder: (context, index) {
                final message = _messages[index];
                return MessageBubble(message: message);
              },
            ),
          ),
          if (_isLoading)
            LinearProgressIndicator(),
          _buildInputArea(),
        ],
      ),
    );
  }
  
  Widget _buildInputArea() {
    return Container(
      padding: EdgeInsets.all(16),
      decoration: BoxDecoration(
        color: Colors.grey[100],
        boxShadow: [
          BoxShadow(
            offset: Offset(0, -2),
            blurRadius: 4,
            color: Colors.black.withOpacity(0.1),
          ),
        ],
      ),
      child: Row(
        children: [
          Expanded(
            child: TextField(
              controller: _controller,
              decoration: InputDecoration(
                hintText: 'Type your message...',
                filled: true,
                fillColor: Colors.white,
                border: OutlineInputBorder(
                  borderRadius: BorderRadius.circular(24),
                  borderSide: BorderSide.none,
                ),
                contentPadding: EdgeInsets.symmetric(
                  horizontal: 20,
                  vertical: 12,
                ),
              ),
              onSubmitted: (_) => _sendMessage(),
              maxLines: null,
              textInputAction: TextInputAction.send,
            ),
          ),
          SizedBox(width: 8),
          CircleAvatar(
            backgroundColor: Theme.of(context).primaryColor,
            child: IconButton(
              icon: Icon(Icons.send, color: Colors.white),
              onPressed: _sendMessage,
            ),
          ),
        ],
      ),
    );
  }
}

class MessageBubble extends StatelessWidget {
  final Message message;
  
  const MessageBubble({Key? key, required this.message}) : super(key: key);
  
  @override
  Widget build(BuildContext context) {
    return Align(
      alignment: message.isUser ? Alignment.centerRight : Alignment.centerLeft,
      child: Container(
        margin: EdgeInsets.symmetric(vertical: 8),
        padding: EdgeInsets.symmetric(horizontal: 16, vertical: 12),
        constraints: BoxConstraints(
          maxWidth: MediaQuery.of(context).size.width * 0.75,
        ),
        decoration: BoxDecoration(
          color: message.isUser
              ? Theme.of(context).primaryColor
              : Colors.grey[200],
          borderRadius: BorderRadius.circular(20).copyWith(
            bottomRight: message.isUser ? Radius.circular(4) : null,
            bottomLeft: !message.isUser ? Radius.circular(4) : null,
          ),
        ),
        child: Column(
          crossAxisAlignment: CrossAxisAlignment.start,
          children: [
            Text(
              message.text,
              style: TextStyle(
                color: message.isUser ? Colors.white : Colors.black87,
                fontSize: 16,
              ),
            ),
            if (message.metadata != null) ...[
              SizedBox(height: 8),
              Text(
                'Model: ${message.metadata!['model'] ?? 'Unknown'}',
                style: TextStyle(
                  fontSize: 12,
                  color: message.isUser
                      ? Colors.white70
                      : Colors.black54,
                ),
              ),
            ],
          ],
        ),
      ),
    );
  }
}

Chat Interface Demo

What is LangChain and how does it work?

LangChain is a framework for developing applications powered by language models. It provides: 1. **Chains**: Combine multiple LLM calls 2. **Agents**: LLMs that can use tools 3. **Memory**: Maintain conversation context 4. **Retrievers**: Access external data It works by orchestrating these components to create intelligent applications.

Implementing Memory Systems

Memory is crucial for creating coherent conversations. LangChain provides several memory types that we can leverage in our Flutter applications.

Memory Type	Description	Use Case	Storage
ConversationBufferMemory	Stores entire conversation	Short conversations	High
ConversationBufferWindowMemory	Stores last K interactions	Long conversations	Medium
ConversationSummaryMemory	Stores summarized conversation	Very long conversations	Low
VectorStoreRetrieverMemory	Stores in vector database	Semantic search	Efficient

Enhanced Memory Implementation

# backend/memory_manager.py
from langchain.memory import (
    ConversationBufferMemory,
    ConversationBufferWindowMemory,
    ConversationSummaryMemory,
    VectorStoreRetrieverMemory
)
from langchain.schema import BaseMessage
from typing import Dict, List, Optional
import json

class MemoryManager:
    def __init__(self, llm, embeddings, vector_store):
        self.llm = llm
        self.embeddings = embeddings
        self.vector_store = vector_store
        self.memories: Dict[str, Dict] = {}
        
    def get_memory(
        self,
        conversation_id: str,
        memory_type: str = "buffer_window"
    ) -> BaseMemory:
        """Get or create memory for conversation"""
        
        if conversation_id not in self.memories:
            self.memories[conversation_id] = {
                "type": memory_type,
                "memory": self._create_memory(memory_type)
            }
        
        return self.memories[conversation_id]["memory"]
    
    def _create_memory(self, memory_type: str) -> BaseMemory:
        """Create memory based on type"""
        
        if memory_type == "buffer":
            return ConversationBufferMemory(
                return_messages=True,
                memory_key="chat_history"
            )
        
        elif memory_type == "buffer_window":
            return ConversationBufferWindowMemory(
                k=10,  # Keep last 10 exchanges
                return_messages=True,
                memory_key="chat_history"
            )
        
        elif memory_type == "summary":
            return ConversationSummaryMemory(
                llm=self.llm,
                return_messages=True,
                memory_key="chat_history"
            )
        
        elif memory_type == "vector_store":
            return VectorStoreRetrieverMemory(
                retriever=self.vector_store.as_retriever(
                    search_kwargs={"k": 5}
                ),
                memory_key="chat_history",
                return_messages=True
            )
        
        else:
            raise ValueError(f"Unknown memory type: {memory_type}")
    
    def save_conversation(self, conversation_id: str) -> Dict:
        """Save conversation to persistent storage"""
        
        if conversation_id not in self.memories:
            return {"error": "Conversation not found"}
        
        memory_data = self.memories[conversation_id]
        memory = memory_data["memory"]
        
        # Extract messages
        messages = memory.chat_memory.messages
        
        # Convert to serializable format
        serialized_messages = [
            {
                "type": msg.__class__.__name__,
                "content": msg.content,
                "additional_kwargs": msg.additional_kwargs
            }
            for msg in messages
        ]
        
        return {
            "conversation_id": conversation_id,
            "memory_type": memory_data["type"],
            "messages": serialized_messages,
            "timestamp": datetime.now().isoformat()
        }
    
    def load_conversation(
        self,
        conversation_id: str,
        saved_data: Dict
    ) -> None:
        """Load conversation from saved data"""
        
        memory_type = saved_data["memory_type"]
        memory = self._create_memory(memory_type)
        
        # Restore messages
        for msg_data in saved_data["messages"]:
            if msg_data["type"] == "HumanMessage":
                memory.chat_memory.add_user_message(msg_data["content"])
            elif msg_data["type"] == "AIMessage":
                memory.chat_memory.add_ai_message(msg_data["content"])
        
        self.memories[conversation_id] = {
            "type": memory_type,
            "memory": memory
        }

Flutter Memory Service

// lib/services/memory_service.dart
import 'package:shared_preferences/shared_preferences.dart';
import 'dart:convert';

class MemoryService {
  static const String _conversationsKey = 'conversations';
  
  Future saveConversation(
    String conversationId,
    List messages,
  ) async {
    final prefs = await SharedPreferences.getInstance();
    
    // Get existing conversations
    final conversationsJson = prefs.getString(_conversationsKey) ?? '{}';
    final conversations = json.decode(conversationsJson) as Map;
    
    // Save current conversation
    conversations[conversationId] = {
      'messages': messages.map((m) => m.toJson()).toList(),
      'timestamp': DateTime.now().toIso8601String(),
    };
    
    // Save back to preferences
    await prefs.setString(_conversationsKey, json.encode(conversations));
  }
  
  Future?> loadConversation(String conversationId) async {
    final prefs = await SharedPreferences.getInstance();
    final conversationsJson = prefs.getString(_conversationsKey) ?? '{}';
    final conversations = json.decode(conversationsJson) as Map;
    
    if (conversations.containsKey(conversationId)) {
      final messagesData = conversations[conversationId]['messages'] as List;
      return messagesData
          .map((m) => Message.fromJson(m))
          .toList();
    }
    
    return null;
  }
  
  Future> getConversationSummaries() async {
    final prefs = await SharedPreferences.getInstance();
    final conversationsJson = prefs.getString(_conversationsKey) ?? '{}';
    final conversations = json.decode(conversationsJson) as Map;
    
    return conversations.entries.map((entry) {
      final messages = (entry.value['messages'] as List)
          .map((m) => Message.fromJson(m))
          .toList();
      
      return ConversationSummary(
        id: entry.key,
        title: _generateTitle(messages),
        lastMessage: messages.isNotEmpty ? messages.last.text : '',
        timestamp: DateTime.parse(entry.value['timestamp']),
        messageCount: messages.length,
      );
    }).toList()
      ..sort((a, b) => b.timestamp.compareTo(a.timestamp));
  }
  
  String _generateTitle(List messages) {
    // Use first user message as title
    final firstUserMessage = messages.firstWhere(
      (m) => m.isUser,
      orElse: () => messages.first,
    );
    
    return firstUserMessage.text.length > 50
        ? '${firstUserMessage.text.substring(0, 50)}...'
        : firstUserMessage.text;
  }
}

class ConversationSummary {
  final String id;
  final String title;
  final String lastMessage;
  final DateTime timestamp;
  final int messageCount;
  
  ConversationSummary({
    required this.id,
    required this.title,
    required this.lastMessage,
    required this.timestamp,
    required this.messageCount,
  });
}

Building RAG Applications

Retrieval Augmented Generation (RAG) allows your chatbot to access and use external knowledge. This is crucial for building chatbots that can answer questions about specific domains or datasets.

💡 What is RAG? RAG combines the power of retrieval systems with generative models. It first retrieves relevant documents from a knowledge base, then uses them as context to generate more accurate and relevant responses.

Document Processing Pipeline

# backend/rag_pipeline.py
from langchain.document_loaders import (
    TextLoader,
    PDFLoader,
    UnstructuredMarkdownLoader,
    CSVLoader
)
from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter,
    TokenTextSplitter
)
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma, Pinecone, Weaviate
from typing import List, Dict, Any
import hashlib

class RAGPipeline:
    def __init__(self, vector_store_type: str = "chroma"):
        self.embeddings = OpenAIEmbeddings()
        self.vector_store = self._initialize_vector_store(vector_store_type)
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            length_function=len,
            separators=["\n\n", "\n", " ", ""]
        )
    
    def _initialize_vector_store(self, store_type: str):
        """Initialize vector store based on type"""
        if store_type == "chroma":
            return Chroma(
                embedding_function=self.embeddings,
                persist_directory="./chroma_db"
            )
        elif store_type == "pinecone":
            import pinecone
            pinecone.init(
                api_key=os.getenv("PINECONE_API_KEY"),
                environment=os.getenv("PINECONE_ENV")
            )
            return Pinecone.from_existing_index(
                index_name="langchain-chatbot",
                embedding=self.embeddings
            )
        else:
            raise ValueError(f"Unsupported vector store: {store_type}")
    
    async def process_documents(
        self,
        file_paths: List[str],
        metadata: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """Process documents and add to vector store"""
        
        all_documents = []
        processed_files = []
        errors = []
        
        for file_path in file_paths:
            try:
                # Load document based on file type
                documents = self._load_document(file_path)
                
                # Add metadata
                for doc in documents:
                    doc.metadata.update(metadata or {})
                    doc.metadata["file_path"] = file_path
                    doc.metadata["file_hash"] = self._calculate_file_hash(file_path)
                
                all_documents.extend(documents)
                processed_files.append(file_path)
                
            except Exception as e:
                errors.append({
                    "file": file_path,
                    "error": str(e)
                })
        
        # Split documents
        split_docs = self.text_splitter.split_documents(all_documents)
        
        # Add to vector store
        self.vector_store.add_documents(split_docs)
        
        return {
            "processed_files": processed_files,
            "total_chunks": len(split_docs),
            "errors": errors
        }
    
    def _load_document(self, file_path: str) -> List[Document]:
        """Load document based on file extension"""
        
        ext = file_path.lower().split('.')[-1]
        
        if ext == 'txt':
            loader = TextLoader(file_path)
        elif ext == 'pdf':
            loader = PDFLoader(file_path)
        elif ext == 'md':
            loader = UnstructuredMarkdownLoader(file_path)
        elif ext == 'csv':
            loader = CSVLoader(file_path)
        else:
            raise ValueError(f"Unsupported file type: {ext}")
        
        return loader.load()
    
    def _calculate_file_hash(self, file_path: str) -> str:
        """Calculate file hash for deduplication"""
        with open(file_path, 'rb') as f:
            return hashlib.sha256(f.read()).hexdigest()
    
    async def semantic_search(
        self,
        query: str,
        k: int = 5,
        filter_dict: Optional[Dict] = None
    ) -> List[Document]:
        """Perform semantic search on vector store"""
        
        if filter_dict:
            return self.vector_store.similarity_search(
                query,
                k=k,
                filter=filter_dict
            )
        else:
            return self.vector_store.similarity_search(query, k=k)
    
    async def hybrid_search(
        self,
        query: str,
        k: int = 5,
        alpha: float = 0.5
    ) -> List[Document]:
        """Perform hybrid search combining semantic and keyword search"""
        
        # Semantic search
        semantic_results = await self.semantic_search(query, k=k*2)
        
        # Keyword search (using metadata)
        keyword_results = self.vector_store.similarity_search(
            query,
            k=k*2,
            search_type="mmr"  # Maximum Marginal Relevance
        )
        
        # Combine and deduplicate
        seen = set()
        combined_results = []
        
        for doc in semantic_results + keyword_results:
            doc_hash = hashlib.md5(doc.page_content.encode()).hexdigest()
            if doc_hash not in seen:
                seen.add(doc_hash)
                combined_results.append(doc)
        
        # Re-rank based on hybrid score
        # Implementation depends on your ranking algorithm
        
        return combined_results[:k]

Flutter RAG Interface

// lib/screens/rag_chat_screen.dart
import 'package:flutter/material.dart';
import 'package:file_picker/file_picker.dart';
import '../services/rag_service.dart';

class RAGChatScreen extends StatefulWidget {
  @override
  _RAGChatScreenState createState() => _RAGChatScreenState();
}

class _RAGChatScreenState extends State {
  final RAGService _ragService = RAGService();
  final TextEditingController _controller = TextEditingController();
  List _documentSources = [];
  bool _isProcessing = false;
  
  Future _uploadDocuments() async {
    final result = await FilePicker.platform.pickFiles(
      allowMultiple: true,
      type: FileType.custom,
      allowedExtensions: ['txt', 'pdf', 'md', 'csv'],
    );
    
    if (result != null) {
      setState(() {
        _isProcessing = true;
      });
      
      try {
        final response = await _ragService.uploadDocuments(
          result.files.map((f) => f.path!).toList(),
        );
        
        setState(() {
          _documentSources.addAll(response.sources);
        });
        
        _showSuccess('Documents processed successfully');
      } catch (e) {
        _showError('Failed to process documents: $e');
      } finally {
        setState(() {
          _isProcessing = false;
        });
      }
    }
  }
  
  Future _performRAGQuery() async {
    if (_controller.text.trim().isEmpty) return;
    
    final query = _controller.text;
    _controller.clear();
    
    setState(() {
      _isProcessing = true;
    });
    
    try {
      final result = await _ragService.performRAGQuery(
        query: query,
        topK: 5,
      );
      
      // Show results in a dialog
      _showRAGResults(result);
    } catch (e) {
      _showError('Query failed: $e');
    } finally {
      setState(() {
        _isProcessing = false;
      });
    }
  }
  
  void _showRAGResults(RAGQueryResult result) {
    showDialog(
      context: context,
      builder: (context) => AlertDialog(
        title: Text('Search Results'),
        content: Container(
          width: double.maxFinite,
          constraints: BoxConstraints(maxHeight: 400),
          child: ListView.builder(
            shrinkWrap: true,
            itemCount: result.results.length,
            itemBuilder: (context, index) {
              final doc = result.results[index];
              return Card(
                margin: EdgeInsets.symmetric(vertical: 8),
                child: Padding(
                  padding: EdgeInsets.all(12),
                  child: Column(
                    crossAxisAlignment: CrossAxisAlignment.start,
                    children: [
                      Text(
                        'Score: ${(doc.score * 100).toStringAsFixed(1)}%',
                        style: TextStyle(
                          fontWeight: FontWeight.bold,
                          color: Theme.of(context).primaryColor,
                        ),
                      ),
                      SizedBox(height: 8),
                      Text(
                        doc.content,
                        maxLines: 4,
                        overflow: TextOverflow.ellipsis,
                      ),
                      SizedBox(height: 8),
                      Text(
                        'Source: ${doc.metadata['file_path'] ?? 'Unknown'}',
                        style: TextStyle(
                          fontSize: 12,
                          color: Colors.grey[600],
                        ),
                      ),
                    ],
                  ),
                ),
              );
            },
          ),
        ),
        actions: [
          TextButton(
            onPressed: () => Navigator.pop(context),
            child: Text('Close'),
          ),
        ],
      ),
    );
  }
  
  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: Text('RAG-Enhanced Chat'),
        actions: [
          IconButton(
            icon: Icon(Icons.upload_file),
            onPressed: _isProcessing ? null : _uploadDocuments,
            tooltip: 'Upload documents',
          ),
        ],
      ),
      body: Column(
        children: [
          // Document sources indicator
          if (_documentSources.isNotEmpty)
            Container(
              height: 50,
              padding: EdgeInsets.symmetric(horizontal: 16),
              child: ListView.builder(
                scrollDirection: Axis.horizontal,
                itemCount: _documentSources.length,
                itemBuilder: (context, index) {
                  final source = _documentSources[index];
                  return Chip(
                    label: Text(source.name),
                    deleteIcon: Icon(Icons.close, size: 18),
                    onDeleted: () {
                      setState(() {
                        _documentSources.removeAt(index);
                      });
                    },
                  );
                },
              ),
            ),
          
          // Chat area
          Expanded(
            child: Center(
              child: Text('RAG-enhanced conversations'),
            ),
          ),
          
          // Input area
          Container(
            padding: EdgeInsets.all(16),
            child: Row(
              children: [
                Expanded(
                  child: TextField(
                    controller: _controller,
                    decoration: InputDecoration(
                      hintText: 'Ask about your documents...',
                      border: OutlineInputBorder(
                        borderRadius: BorderRadius.circular(24),
                      ),
                    ),
                    onSubmitted: (_) => _performRAGQuery(),
                  ),
                ),
                SizedBox(width: 8),
                IconButton(
                  icon: Icon(Icons.search),
                  onPressed: _isProcessing ? null : _performRAGQuery,
                ),
              ],
            ),
          ),
          
          if (_isProcessing)
            LinearProgressIndicator(),
        ],
      ),
    );
  }
}

class DocumentSource {
  final String id;
  final String name;
  final String path;
  final int chunkCount;
  
  DocumentSource({
    required this.id,
    required this.name,
    required this.path,
    required this.chunkCount,
  });
}

Creating Intelligent Agents

LangChain agents are LLMs that can use tools to accomplish tasks. They can reason about what actions to take and in what order, making them incredibly powerful for complex applications.

Agent Implementation

# backend/agents.py
from langchain.agents import initialize_agent, Tool, AgentType
from langchain.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from langchain.utilities import WikipediaAPIWrapper
from langchain.tools.python.tool import PythonREPLTool
from langchain.tools import StructuredTool
from typing import Dict, List, Any
import requests

class AgentManager:
    def __init__(self, llm):
        self.llm = llm
        self.tools = self._initialize_tools()
        self.agent = self._initialize_agent()
    
    def _initialize_tools(self) -> List[Tool]:
        """Initialize available tools for the agent"""
        
        # Web search tool
        search = DuckDuckGoSearchRun()
        
        # Wikipedia tool
        wikipedia = WikipediaQueryRun(
            api_wrapper=WikipediaAPIWrapper()
        )
        
        # Python REPL for calculations
        python_repl = PythonREPLTool()
        
        # Custom weather tool
        weather_tool = StructuredTool.from_function(
            func=self._get_weather,
            name="get_weather",
            description="Get current weather for a location"
        )
        
        # Custom database query tool
        db_tool = StructuredTool.from_function(
            func=self._query_database,
            name="query_database",
            description="Query the application database"
        )
        
        return [
            Tool(
                name="Search",
                func=search.run,
                description="Search the web for current information"
            ),
            Tool(
                name="Wikipedia",
                func=wikipedia.run,
                description="Search Wikipedia for encyclopedic information"
            ),
            Tool(
                name="Calculator",
                func=python_repl.run,
                description="Execute Python code for calculations"
            ),
            weather_tool,
            db_tool
        ]
    
    def _initialize_agent(self):
        """Initialize the agent with tools"""
        
        return initialize_agent(
            tools=self.tools,
            llm=self.llm,
            agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
            verbose=True,
            max_iterations=5,
            early_stopping_method="generate"
        )
    
    async def _get_weather(self, location: str) -> str:
        """Get weather for a location"""
        # In production, use a real weather API
        api_key = os.getenv("WEATHER_API_KEY")
        url = f"https://api.openweathermap.org/data/2.5/weather"
        
        response = requests.get(url, params={
            "q": location,
            "appid": api_key,
            "units": "metric"
        })
        
        if response.status_code == 200:
            data = response.json()
            return f"Weather in {location}: {data['weather'][0]['description']}, {data['main']['temp']}°C"
        else:
            return f"Could not get weather for {location}"
    
    async def _query_database(self, query: str) -> str:
        """Query application database"""
        # Implement your database query logic
        # This is a placeholder
        return f"Database query result for: {query}"
    
    async def process_with_tools(
        self,
        user_input: str,
        conversation_history: List[Dict] = None
    ) -> Dict[str, Any]:
        """Process user input with agent and tools"""
        
        try:
            # Add conversation history if available
            if conversation_history:
                # Format history for agent
                history_text = "\n".join([
                    f"{msg['role']}: {msg['content']}"
                    for msg in conversation_history[-5:]  # Last 5 messages
                ])
                
                full_input = f"Previous conversation:\n{history_text}\n\nUser: {user_input}"
            else:
                full_input = user_input
            
            # Run agent
            response = await self.agent.ainvoke({"input": full_input})
            
            return {
                "response": response["output"],
                "intermediate_steps": response.get("intermediate_steps", []),
                "tools_used": [
                    step[0].tool for step in response.get("intermediate_steps", [])
                ]
            }
            
        except Exception as e:
            return {
                "response": f"An error occurred: {str(e)}",
                "error": True
            }

Flutter Agent Interface

// lib/screens/agent_chat_screen.dart
class AgentChatScreen extends StatefulWidget {
  @override
  _AgentChatScreenState createState() => _AgentChatScreenState();
}

class _AgentChatScreenState extends State {
  final AgentService _agentService = AgentService();
  final List _messages = [];
  bool _isProcessing = false;
  
  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: Text('AI Agent Assistant'),
        actions: [
          PopupMenuButton(
            onSelected: (value) {
              // Handle tool selection
              _showToolInfo(value);
            },
            itemBuilder: (context) => [
              PopupMenuItem(
                value: 'search',
                child: Row(
                  children: [
                    Icon(Icons.search, size: 20),
                    SizedBox(width: 8),
                    Text('Web Search'),
                  ],
                ),
              ),
              PopupMenuItem(
                value: 'calculator',
                child: Row(
                  children: [
                    Icon(Icons.calculate, size: 20),
                    SizedBox(width: 8),
                    Text('Calculator'),
                  ],
                ),
              ),
              PopupMenuItem(
                value: 'weather',
                child: Row(
                  children: [
                    Icon(Icons.cloud, size: 20),
                    SizedBox(width: 8),
                    Text('Weather'),
                  ],
                ),
              ),
            ],
          ),
        ],
      ),
      body: Column(
        children: [
          Expanded(
            child: ListView.builder(
              padding: EdgeInsets.all(16),
              itemCount: _messages.length,
              itemBuilder: (context, index) {
                final message = _messages[index];
                return AgentMessageWidget(message: message);
              },
            ),
          ),
          if (_isProcessing)
            Container(
              padding: EdgeInsets.all(16),
              child: Column(
                children: [
                  CircularProgressIndicator(),
                  SizedBox(height: 8),
                  Text('Agent is thinking...'),
                ],
              ),
            ),
          _buildInputArea(),
        ],
      ),
    );
  }
}

class AgentMessageWidget extends StatelessWidget {
  final AgentMessage message;
  
  @override
  Widget build(BuildContext context) {
    return Card(
      margin: EdgeInsets.symmetric(vertical: 8),
      child: Padding(
        padding: EdgeInsets.all(16),
        child: Column(
          crossAxisAlignment: CrossAxisAlignment.start,
          children: [
            Row(
              children: [
                Icon(
                  message.isUser ? Icons.person : Icons.smart_toy,
                  color: message.isUser
                      ? Theme.of(context).primaryColor
                      : Colors.orange,
                ),
                SizedBox(width: 8),
                Text(
                  message.isUser ? 'You' : 'AI Agent',
                  style: TextStyle(fontWeight: FontWeight.bold),
                ),
                Spacer(),
                Text(
                  _formatTime(message.timestamp),
                  style: TextStyle(fontSize: 12, color: Colors.grey),
                ),
              ],
            ),
            SizedBox(height: 12),
            Text(message.text),
            if (message.toolsUsed.isNotEmpty) ...[
              SizedBox(height: 12),
              Wrap(
                spacing: 8,
                children: message.toolsUsed.map((tool) {
                  return Chip(
                    label: Text(tool),
                    avatar: Icon(_getToolIcon(tool), size: 16),
                    backgroundColor: Colors.blue.shade50,
                  );
                }).toList(),
              ),
            ],
            if (message.intermediateSteps.isNotEmpty) ...[
              SizedBox(height: 12),
              ExpansionTile(
                title: Text('View reasoning steps'),
                children: message.intermediateSteps.map((step) {
                  return ListTile(
                    leading: Icon(Icons.arrow_right),
                    title: Text(step['action'] ?? ''),
                    subtitle: Text(step['observation'] ?? ''),
                  );
                }).toList(),
              ),
            ],
          ],
        ),
      ),
    );
  }
  
  IconData _getToolIcon(String tool) {
    switch (tool.toLowerCase()) {
      case 'search':
        return Icons.search;
      case 'calculator':
        return Icons.calculate;
      case 'weather':
        return Icons.cloud;
      case 'wikipedia':
        return Icons.book;
      default:
        return Icons.build;
    }
  }
}

Function Calling and Tools

Function calling allows your chatbot to interact with external systems and APIs, making it truly useful for real-world applications.

✅ Best Practice: Always validate function inputs and outputs to ensure safety and reliability. Implement proper error handling for all external API calls.

Function Definitions

# backend/functions.py
from typing import Dict, List, Any, Optional
from pydantic import BaseModel, Field
from langchain.tools import tool
import json

class FlightSearchParams(BaseModel):
    origin: str = Field(description="Origin airport code")
    destination: str = Field(description="Destination airport code")
    date: str = Field(description="Travel date in YYYY-MM-DD format")
    passengers: int = Field(description="Number of passengers", default=1)

class EmailParams(BaseModel):
    recipient: str = Field(description="Email recipient")
    subject: str = Field(description="Email subject")
    body: str = Field(description="Email body")

class FunctionRegistry:
    def __init__(self):
        self.functions = {}
        self._register_functions()
    
    def _register_functions(self):
        """Register all available functions"""
        
        @tool
        def search_flights(params: FlightSearchParams) -> Dict[str, Any]:
            """Search for available flights"""
            # Implement flight search logic
            return {
                "flights": [
                    {
                        "airline": "Flutter Airways",
                        "flight_number": "FL123",
                        "departure": f"{params.origin} 10:00",
                        "arrival": f"{params.destination} 14:00",
                        "price": 299.99
                    }
                ],
                "search_params": params.dict()
            }
        
        @tool
        def send_email(params: EmailParams) -> Dict[str, Any]:
            """Send an email"""
            # Implement email sending logic
            return {
                "status": "sent",
                "message_id": "msg_123",
                "timestamp": datetime.now().isoformat()
            }
        
        @tool
        def get_user_calendar(date: str) -> List[Dict[str, Any]]:
            """Get user's calendar events for a specific date"""
            # Implement calendar retrieval
            return [
                {
                    "time": "10:00",
                    "title": "Team Meeting",
                    "duration": 60
                },
                {
                    "time": "14:00",
                    "title": "Client Call",
                    "duration": 30
                }
            ]
        
        @tool
        def create_reminder(
            title: str,
            datetime_str: str,
            description: Optional[str] = None
        ) -> Dict[str, Any]:
            """Create a reminder"""
            return {
                "reminder_id": "rem_456",
                "title": title,
                "datetime": datetime_str,
                "description": description,
                "status": "created"
            }
        
        # Register functions
        self.functions = {
            "search_flights": search_flights,
            "send_email": send_email,
            "get_user_calendar": get_user_calendar,
            "create_reminder": create_reminder
        }
    
    def get_tools(self) -> List[Tool]:
        """Get all registered tools"""
        return list(self.functions.values())
    
    def get_tool_descriptions(self) -> List[Dict[str, Any]]:
        """Get descriptions of all tools"""
        return [
            {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.args_schema.schema() if tool.args_schema else {}
            }
            for tool in self.functions.values()
        ]

Real-Time Streaming Responses

Streaming responses provide a better user experience by showing the AI's response as it's being generated, similar to ChatGPT.

Streaming Backend

# backend/streaming.py
from fastapi import WebSocket
from langchain.callbacks.base import BaseCallbackHandler
import asyncio
import json

class StreamingCallbackHandler(BaseCallbackHandler):
    def __init__(self, websocket: WebSocket):
        self.websocket = websocket
    
    async def on_llm_new_token(self, token: str, **kwargs) -> None:
        """Send each token as it's generated"""
        await self.websocket.send_json({
            "type": "token",
            "content": token
        })
    
    async def on_llm_end(self, response, **kwargs) -> None:
        """Signal end of generation"""
        await self.websocket.send_json({
            "type": "end"
        })
    
    async def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs) -> None:
        """Notify when tool execution starts"""
        await self.websocket.send_json({
            "type": "tool_start",
            "tool": serialized.get("name", "unknown"),
            "input": input_str
        })
    
    async def on_tool_end(self, output: str, **kwargs) -> None:
        """Send tool execution result"""
        await self.websocket.send_json({
            "type": "tool_end",
            "output": output
        })

@app.websocket("/ws/chat/{conversation_id}")
async def websocket_chat(websocket: WebSocket, conversation_id: str):
    await websocket.accept()
    
    try:
        handler = LangChainHandler()
        
        while True:
            # Receive message from client
            data = await websocket.receive_json()
            
            # Create streaming callback
            streaming_callback = StreamingCallbackHandler(websocket)
            
            # Process with streaming
            llm_with_streaming = handler.llm.copy()
            llm_with_streaming.callbacks = [streaming_callback]
            
            # Get memory
            memory = handler.get_or_create_memory(conversation_id)
            
            # Create chain with streaming LLM
            chain = ConversationalRetrievalChain.from_llm(
                llm=llm_with_streaming,
                retriever=handler.vector_store.as_retriever(),
                memory=memory,
                return_source_documents=True
            )
            
            # Process message
            await chain.ainvoke({"question": data["message"]})
            
    except Exception as e:
        await websocket.send_json({
            "type": "error",
            "message": str(e)
        })
    finally:
        await websocket.close()

Flutter Streaming Client

// lib/services/streaming_service.dart
import 'package:web_socket_channel/web_socket_channel.dart';
import 'dart:convert';

class StreamingService {
  WebSocketChannel? _channel;
  final String baseUrl;
  
  StreamingService({required this.baseUrl});
  
  Stream streamChat({
    required String message,
    required String conversationId,
  }) async* {
    // Connect to WebSocket
    final wsUrl = baseUrl.replaceFirst('http', 'ws');
    _channel = WebSocketChannel.connect(
      Uri.parse('$wsUrl/ws/chat/$conversationId'),
    );
    
    // Send message
    _channel!.sink.add(json.encode({
      'message': message,
    }));
    
    // Listen to stream
    await for (final data in _channel!.stream) {
      final decoded = json.decode(data);
      
      switch (decoded['type']) {
        case 'token':
          yield ChatToken(decoded['content']);
          break;
        case 'end':
          yield ChatEnd();
          break;
        case 'tool_start':
          yield ToolStart(
            tool: decoded['tool'],
            input: decoded['input'],
          );
          break;
        case 'tool_end':
          yield ToolEnd(output: decoded['output']);
          break;
        case 'error':
          yield ChatError(decoded['message']);
          break;
      }
    }
  }
  
  void dispose() {
    _channel?.sink.close();
  }
}

// Event types
abstract class ChatEvent {}

class ChatToken extends ChatEvent {
  final String token;
  ChatToken(this.token);
}

class ChatEnd extends ChatEvent {}

class ToolStart extends ChatEvent {
  final String tool;
  final String input;
  ToolStart({required this.tool, required this.input});
}

class ToolEnd extends ChatEvent {
  final String output;
  ToolEnd({required this.output});
}

class ChatError extends ChatEvent {
  final String message;
  ChatError(this.message);
}

// Usage in UI
class StreamingChatWidget extends StatefulWidget {
  @override
  _StreamingChatWidgetState createState() => _StreamingChatWidgetState();
}

class _StreamingChatWidgetState extends State {
  final StreamingService _streamingService = StreamingService(
    baseUrl: 'http://localhost:8000',
  );
  String _currentResponse = '';
  bool _isStreaming = false;
  
  Future _sendMessage(String message) async {
    setState(() {
      _currentResponse = '';
      _isStreaming = true;
    });
    
    await for (final event in _streamingService.streamChat(
      message: message,
      conversationId: 'conv_123',
    )) {
      if (event is ChatToken) {
        setState(() {
          _currentResponse += event.token;
        });
      } else if (event is ChatEnd) {
        setState(() {
          _isStreaming = false;
        });
      } else if (event is ToolStart) {
        // Show tool usage indicator
      } else if (event is ChatError) {
        _showError(event.message);
      }
    }
  }
  
  @override
  Widget build(BuildContext context) {
    return Column(
      children: [
        if (_currentResponse.isNotEmpty)
          Container(
            padding: EdgeInsets.all(16),
            child: Text(_currentResponse),
          ),
        if (_isStreaming)
          LinearProgressIndicator(),
      ],
    );
  }
}

Production Deployment

Deploying LangChain applications to production requires careful consideration of security, scalability, and monitoring.

Security Best Practices

⚠️ Security Warning: Never expose API keys in your Flutter app. Always use a backend service to handle LLM interactions.

# backend/security.py
from fastapi import Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from datetime import datetime, timedelta
import os

security = HTTPBearer()

class AuthHandler:
    secret = os.getenv("JWT_SECRET")
    algorithm = "HS256"
    
    def encode_token(self, user_id: str) -> str:
        payload = {
            "user_id": user_id,
            "exp": datetime.utcnow() + timedelta(hours=24),
            "iat": datetime.utcnow()
        }
        return jwt.encode(payload, self.secret, algorithm=self.algorithm)
    
    def decode_token(self, token: str) -> Dict:
        try:
            payload = jwt.decode(token, self.secret, algorithms=[self.algorithm])
            return payload
        except jwt.ExpiredSignatureError:
            raise HTTPException(status_code=401, detail="Token expired")
        except jwt.InvalidTokenError:
            raise HTTPException(status_code=401, detail="Invalid token")

auth_handler = AuthHandler()

async def get_current_user(credentials: HTTPAuthorizationCredentials = Security(security)):
    token = credentials.credentials
    payload = auth_handler.decode_token(token)
    return payload["user_id"]

# Rate limiting
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.post("/chat", dependencies=[Depends(get_current_user)])
@limiter.limit("10/minute")
async def chat_endpoint(request: ChatRequest):
    # Your chat logic here
    pass

Monitoring and Analytics

# backend/monitoring.py
from prometheus_client import Counter, Histogram, Gauge
import time
from functools import wraps

# Metrics
chat_requests = Counter('langchain_chat_requests_total', 'Total chat requests')
chat_errors = Counter('langchain_chat_errors_total', 'Total chat errors')
response_time = Histogram('langchain_response_time_seconds', 'Response time')
active_conversations = Gauge('langchain_active_conversations', 'Active conversations')
tokens_used = Counter('langchain_tokens_used_total', 'Total tokens used')

def track_metrics(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start_time = time.time()
        chat_requests.inc()
        
        try:
            result = await func(*args, **kwargs)
            return result
        except Exception as e:
            chat_errors.inc()
            raise
        finally:
            response_time.observe(time.time() - start_time)
    
    return wrapper

# Usage analytics
class AnalyticsService:
    def __init__(self):
        self.db = get_database()
    
    async def log_conversation(
        self,
        user_id: str,
        conversation_id: str,
        message: str,
        response: str,
        tokens_used: int,
        tools_used: List[str],
        response_time: float
    ):
        await self.db.conversations.insert_one({
            "user_id": user_id,
            "conversation_id": conversation_id,
            "message": message,
            "response": response,
            "tokens_used": tokens_used,
            "tools_used": tools_used,
            "response_time": response_time,
            "timestamp": datetime.utcnow()
        })
    
    async def get_user_stats(self, user_id: str) -> Dict:
        pipeline = [
            {"$match": {"user_id": user_id}},
            {"$group": {
                "_id": "$user_id",
                "total_messages": {"$sum": 1},
                "total_tokens": {"$sum": "$tokens_used"},
                "avg_response_time": {"$avg": "$response_time"},
                "unique_conversations": {"$addToSet": "$conversation_id"}
            }}
        ]
        
        result = await self.db.conversations.aggregate(pipeline).to_list(1)
        return result[0] if result else {}

Conclusion

Building AI chatbots with Flutter and LangChain opens up incredible possibilities for creating intelligent, context-aware applications. By leveraging LangChain's powerful abstractions and Flutter's cross-platform capabilities, you can build sophisticated conversational interfaces that work seamlessly across all platforms.

Key Takeaways

LangChain provides powerful abstractions for building LLM applications
Memory systems enable coherent, context-aware conversations
RAG allows chatbots to access and use external knowledge
Agents can use tools to accomplish complex tasks
Streaming responses improve user experience
Production deployment requires careful attention to security and monitoring

Next Steps

Experiment with Different LLMs: Try Claude, GPT-4, or open-source models
Build Custom Tools: Create tools specific to your domain
Implement Advanced RAG: Use hybrid search and re-ranking
Add Voice Integration: Combine with speech-to-text and text-to-speech
Deploy to Production: Set up proper infrastructure and monitoring

About the Author

Utsav Savani is the founder of Innovative Flutter AI and a full-stack product engineer with 8+ years of experience. He designs in Figma, builds in Flutter and ships the backend on Supabase, Firebase and AWS, and leads a 50-person team. Find him on GitHub and LinkedIn.

Creating AI Chatbots with Flutter and LangChain

Table of Contents

Introduction to LangChain

Understanding LangChain Architecture

LangChain Core Components

Key Components Explained

Setting Up LangChain with Flutter

Project Structure

Backend Setup (Python FastAPI)

LangChain Handler Implementation

Flutter Service Layer

Building Your First Chatbot

Chat Interface Demo

Implementing Memory Systems

Enhanced Memory Implementation

Flutter Memory Service

Building RAG Applications

Document Processing Pipeline

Flutter RAG Interface

Creating Intelligent Agents

Agent Implementation

Flutter Agent Interface

Function Calling and Tools

Function Definitions

Real-Time Streaming Responses

Streaming Backend

Flutter Streaming Client

Production Deployment

Security Best Practices

Monitoring and Analytics

Conclusion

Key Takeaways

Next Steps

About the Author

Building something like this?

Creating AI Chatbots with Flutter and LangChain

Table of Contents

Introduction to LangChain

Understanding LangChain Architecture

LangChain Core Components

Key Components Explained

Setting Up LangChain with Flutter

Project Structure

Backend Setup (Python FastAPI)

LangChain Handler Implementation

Flutter Service Layer

Building Your First Chatbot

Chat Interface Demo

Implementing Memory Systems

Enhanced Memory Implementation

Flutter Memory Service

Building RAG Applications

Document Processing Pipeline

Flutter RAG Interface

Creating Intelligent Agents

Agent Implementation

Flutter Agent Interface

Function Calling and Tools

Function Definitions

Real-Time Streaming Responses

Streaming Backend

Flutter Streaming Client

Production Deployment

Security Best Practices

Monitoring and Analytics

Conclusion

Key Takeaways

Next Steps

About the Author

Related Articles

Building AI-Powered Flutter Apps with Google Gemini

TensorFlow Lite in Flutter

ML Kit Vision APIs in Flutter

Building something like this?