Hier is een voorbeeld van hoe je een aangepaste LangChain-agent kunt bouwen voor het behandelen van klantvragen over productproblemen en escalaties, met aandacht voor snelle reactietijden en laag geheugenverbruik. Stap 1: Installeer benodigde packages ```bash pip install langchain openai ``` Stap 2: Definieer prompttemplates Gebruik `PromptTemplate` om gestructureerde prompts te maken voor klantvragen en escalaties. ```python from langchain.prompts import PromptTemplate # Voor klantvragen over productproblemen product_issue_template = PromptTemplate( template=""" Jij bent een klantenservicemedewerker. Beantwoord de vraag van de klant over een productprobleem. Geef een duidelijke en beknopte oplossing of advies. Klantvraag: {question} Antwoord:""" ) # Voor escalaties escalation_template = PromptTemplate( template=""" Jij bent een klantenservicemedewerker die een escalatie moet afhandelen. Beoordeel of de kwestie moet worden doorverwezen naar een supervisor of directe oplossing kan bieden. Klantvraag: {question} Antwoord:""" ) ``` Stap 3: Configureer een snelle en efficiënte LLM (bijvoorbeeld OpenAI GPT-3.5 Turbo) met beperkte context en lage latentie. ```python from langchain.chat_models import ChatOpenAI llm = ChatOpenAI( model="gpt-3.5-turbo", temperature=0, max_tokens=300, # Beperk het aantal tokens voor snelheid request_timeout=2 # Tijdslimiet in seconden ) ``` Stap 4: Maak een eenvoudige keten die het prompttemplate gebruikt ```python from langchain.chains import LLMChain product_issue_chain = LLMChain( llm=llm, prompt=product_issue_template ) escalation_chain = LLMChain( llm=llm, prompt=escalation_template ) ``` Stap 5: Voeg logica toe om te bepalen welke keten wordt gebruikt, bijvoorbeeld op basis van sleutelwoorden ```python def behandel_klantvraag(question): # Eenvoudige keyword-analyse voor selectie escalatie_keywords = ["boos", "klacht", "terugbetaling", "reparatie"] if any(keyword in question.lower() for keyword in escalatie_keywords): response = escalation_chain.run(question=question) else: response = product_issue_chain.run(question=question) return response ``` Stap 6: Integratie met externe databronnen of API's Bijvoorbeeld, om productstatus te controleren: ```python import requests def controleer_product_status(product_id): # Dummy API call response = requests.get(f"https://api.example.com/product/{product_id}/status") if response.status_code == 200: return response.json().get("status") else: return "Onbekende status" ``` Gebruik dit binnen je functies waar nodig. --- **Beperkingen / Prestatie-eisen:** - Reactietijd onder 3 seconden vereist snelle API-responses en mogelijk caching. - Laag geheugenverbruik door beperking van tokens en gebruik van eenvoudige ketens. - Gebruik van een betaalde API zoals OpenAI GPT-3.5 Turbo met request timeout en throttling om latency te beperken. --- **Samenvatting:** - Gebruik `PromptTemplate` voor gestructureerde vragen. - Beperk tokens en timeout voor snelheid. - Implementeer eenvoudige tekstanalyse om de juiste keten te kiezen. - Integreer externe API's voor aanvullende data. Als je verdere hulp nodig hebt bij het implementeren of uitbreiden, laat het gerust weten!

Zeker, ik help je bij het bouwen van een aangepaste LangChain-agent voor klantenservice. Hier is een complete implementatie: ## 1. Basisconfiguratie en Imports ```python from langchain.agents import AgentType, initialize_agent, Tool from langchain.chains import LLMChain, RetrievalQA from langchain.llms import OpenAI from langchain.prompts import PromptTemplate from langchain.memory import ConversationBufferWindowMemory from langchain.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import FAISS from langchain.document_loaders import TextLoader from langchain.text_splitter import CharacterTextSplitter import requests import json from typing import Dict, Any import time ``` ## 2. Prompt Templates ```python # Hoofdprompt voor de agent AGENT_PROMPT = PromptTemplate( input_variables=["input", "agent_scratchpad", "history"], template="""Je bent een professionele klantenservice-agent gespecialiseerd in productproblemen en escalaties. Gespreksgeschiedenis: {history} Huidige vraag: {input} {agent_scratchpad} Handel de vraag af volgens dit protocol: 1. Identificeer het productprobleem 2. Controleer beschikbare oplossingen in kennisbank 3. Beoordeel urgentie en impact 4. Escaleer indien nodig volgens escalation matrix 5. Geef duidelijke volgende stappen Antwoord in het Nederlands en houd reacties onder 200 woorden.""" ) # Template voor escalatie-analyse ESCALATION_PROMPT = PromptTemplate( input_variables=["issue", "customer_tier"], template="""Analyseer deze melding voor escalatie: Probleem: {issue} Klanttype: {customer_tier} Beoordeel: - Urgentie (hoog/medium/laag) - Impact op klantoperaties - Benodigde expertise niveau - Geschatte resolutietijd Geef escalatie-advies:""" ) ``` ## 3. Externe Integraties (Tools) ```python class KnowledgeBaseTool: """Tool voor kennisbank-integratie""" def __init__(self): self.embeddings = HuggingFaceEmbeddings( model_name="sentence-transformers/all-MiniLM-L6-v2" ) self.vectorstore = self._setup_knowledge_base() def _setup_knowledge_base(self): # Laad kennisbank documenten (pas pad aan) loader = TextLoader("kennisbank.txt", encoding='utf-8') documents = loader.load() text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=100) docs = text_splitter.split_documents(documents) return FAISS.from_documents(docs, self.embeddings) def search_solutions(self, query: str) -> str: """Zoek oplossingen in kennisbank""" start_time = time.time() if time.time() - start_time > 2.5: # Timeout voor performance return "Kennisbank tijdelijk niet beschikbaar. Escaleer naar senior support." results = self.vectorstore.similarity_search(query, k=3) return "\n\n".join([doc.page_content for doc in results]) class APIIntegrationTool: """Tool voor externe API-integraties""" def check_ticket_status(self, ticket_id: str) -> str: """Controleer ticket status via API""" try: # Mock API call - vervang met echte integratie response = requests.get( f"https://api.example.com/tickets/{ticket_id}", timeout=2.0 # Timeout voor performance ) return response.json().get('status', 'Onbekend') except: return "API niet bereikbaar. Handmatige statuscontrole nodig." def create_escalation_ticket(self, issue_data: Dict[str, Any]) -> str: """Creëer escalatie ticket""" try: payload = { "issue": issue_data.get("issue"), "urgency": issue_data.get("urgency", "medium"), "customer_id": issue_data.get("customer_id") } response = requests.post( "https://api.example.com/escalations", json=payload, timeout=2.5 ) return f"Escalatie aangemaakt: {response.json().get('ticket_id')}" except: return "Escalatie systeem onbereikbaar. Neem contact op met teamleider." ``` ## 4. Agent Setup en Configuratie ```python def setup_agent(): # Lightweight LLM configuratie llm = OpenAI( temperature=0.1, # Lage temperature voor consistente antwoorden max_tokens=200, # Beperk output lengte model_name="gpt-3.5-turbo" # Kosten-effectief model ) # Initialize tools kb_tool = KnowledgeBaseTool() api_tool = APIIntegrationTool() # Definieer beschikbare tools tools = [ Tool( name="Kennisbank Zoeken", func=kb_tool.search_solutions, description="Gebruik dit voor het zoeken naar oplossingen in de kennisbank. Input is de zoekvraag." ), Tool( name="Ticket Status Check", func=api_tool.check_ticket_status, description="Gebruik dit om de status van bestaande tickets te controleren. Input is ticket ID." ), Tool( name="Escalatie Aanmaken", func=api_tool.create_escalation_ticket, description="Gebruik dit om een nieuwe escalatie aan te maken. Input is JSON met issue, urgency, customer_id." ) ] # Memory configuratie met beperkte window size memory = ConversationBufferWindowMemory( memory_key="history", k=3, # Houd alleen laatste 3 berichten return_messages=True ) # Initialize agent met performance constraints agent = initialize_agent( tools=tools, llm=llm, agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION, verbose=True, memory=memory, max_iterations=3, # Beperk iteraties voor performance early_stopping_method="generate", agent_kwargs={"prompt": AGENT_PROMPT} ) return agent # Initializeer de agent customer_agent = setup_agent() ``` ## 5. Gebruiksvoorbeeld ```python # Voorbeeld interactie def handle_customer_query(query: str, customer_context: Dict[str, Any] = None): start_time = time.time() try: # Voeg context toe aan query enhanced_query = f"{query} | Klanttype: {customer_context.get('tier', 'standard')}" if customer_context else query response = customer_agent.run(enhanced_query) # Performance monitoring processing_time = time.time() - start_time if processing_time > 2.8: print(f"Waarschuwing: Reactietijd {processing_time:.2f}s nadert limiet") return response except Exception as e: return f"Excuses, er is een technische storing. Probeer het later opnieuw. Error: {str(e)}" # Voorbeeld gebruik response = handle_customer_query( "Mijn product XYZ werkt niet meer na de update", {"tier": "premium", "customer_id": "12345"} ) print(response) ``` ## 6. Prestatiebeperkingen en Optimalisaties **Beperkingen:** - Maximale reactietijd: < 3 seconden - Geheugengebruik: < 512MB RAM - Maximaal 3 tool calls per interactie - Response lengte: max 200 tokens **Optimalisatiestrategieën:** ```python # Cache configuratie voor performance from langchain.cache import InMemoryCache import langchain langchain.llm_cache = InMemoryCache() # Batch processing voor kennisbank queries def batch_search(queries: List[str]) -> List[str]: """Batch verwerking voor efficiëntie""" return [kb_tool.search_solutions(q) for q in queries] # Memory management import gc def cleanup_memory(): """Periodieke memory cleanup""" gc.collect() ``` ## 7. Monitoring en Logging ```python class PerformanceMonitor: """Monitor agent performance""" def __init__(self): self.response_times = [] def log_performance(self, start_time: float, success: bool): duration = time.time() - start_time self.response_times.append(duration) if duration > 2.5: # Alert voor bijna-timeout print(f"PERFORMANCE WARNING: {duration:.2f}s") if len(self.response_times) > 100: # Bereken gemiddelde elke 100 requests avg_time = sum(self.response_times[-100:]) / 100 print(f"Gemiddelde reactietijd: {avg_time:.2f}s") ``` ## Installatievereisten ```bash pip install langchain openai faiss-cpu sentence-transformers requests ``` Deze implementatie voldoet aan je prestatie-eisen door: - Beperkte context window (laag geheugen) - Timeouts op externe calls - Efficiente embedding model - Beperkte iteraties en token usage - Caching mechanisme Pas de API endpoints en kennisbank paden aan naar je eigen omgeving.