​​Erstellung eines RAG-Workflows mit LangGraph und Elasticsearch

Die LangGraph Retrieval Agent-Vorlage ist ein von LangChain entwickeltes Starterprojekt, das die Erstellung abrufbasierter Frage-Antwort-Systeme mit LangGraph in LangGraph Studio erleichtern soll. Diese Vorlage ist für die nahtlose Integration mit Elasticsearch vorkonfiguriert und ermöglicht Entwicklern die schnelle Erstellung von Agenten, die Dokumente effizient indizieren und abrufen können.

In diesem Blog geht es um das Ausführen und Anpassen der LangChain Retrieval Agent-Vorlage mit LangGraph Studio und LangGraph CLI. Die Vorlage bietet ein Framework zum Erstellen von Retrieval-Augmented Generation (RAG)-Anwendungen und nutzt verschiedene Retrieval-Backends wie Elasticsearch.

Wir führen Sie durch die Einrichtung, Konfiguration der Umgebung und effiziente Ausführung der Vorlage mit Elastic, während wir den Agentenfluss anpassen.

Voraussetzungen

Bevor Sie fortfahren, stellen Sie sicher, dass Folgendes installiert ist:

• Elasticsearch Cloud-Bereitstellung oder lokale Elasticsearch-Bereitstellung (oder erstellen Sie eine 14-tägige kostenlose Testversion auf Elastic Cloud) – Version 8.0.0 oder höher

• Python 3.9+

• Zugriff auf einen LLM-Anbieter wie Cohere (in diesem Handbuch verwendet), OpenAI oder Anthropic/Claude

Erstellen der LangGraph-App

1. Installieren Sie die LangGraph CLI

pip install --upgrade "langgraph-cli[inmem]"

2. Erstellen Sie eine LangGraph-App aus der Retrieval-Agent-Vorlage

mkdir lg-agent-demo
cd lg-agent-demo
langgraph new lg-agent-demo

Ihnen wird ein interaktives Menü angezeigt, in dem Sie aus einer Liste verfügbarer Vorlagen auswählen können. Wählen Sie 4 für Retrieval Agent und 1 für Python, wie unten gezeigt:

• Fehlerbehebung: Wenn der Fehler „urllib.error.URLError: <urlopen error [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1000)>“ auftritt, „

Führen Sie den Python-Befehl „Zertifikat installieren“ aus, um das Problem wie unten gezeigt zu beheben.

3. Abhängigkeiten installieren

Erstellen Sie im Stammverzeichnis Ihrer neuen LangGraph-App eine virtuelle Umgebung und installieren Sie die Abhängigkeiten im Modus edit , damit Ihre lokalen Änderungen vom Server verwendet werden:

#For Mac
python3 -m venv lg-demo
source lg-demo/bin/activate 
pip install -e .

#For Windows
python3 -m venv lg-demo
lg-demo\Scripts\activate 
pip install -e .

Einrichten der Umgebung

1. Erstellen Sie eine .environment-Umgebung Datei

Die Datei .env enthält API-Schlüssel und Konfigurationen, damit die App eine Verbindung mit dem von Ihnen gewählten LLM- und Abrufanbieter herstellen kann. Generieren Sie eine neue .env -Datei, indem Sie die Beispielkonfiguration duplizieren:

cp .env.example .env

2. Konfigurieren Sie die .env-Datei. Datei

Die Datei .env enthält eine Reihe von Standardkonfigurationen. Sie können es aktualisieren, indem Sie basierend auf Ihrem Setup die erforderlichen API-Schlüssel und -Werte hinzufügen. Alle Schlüssel, die für Ihren Anwendungsfall nicht relevant sind, können unverändert bleiben oder entfernt werden.

# To separate your traces from other applications
LANGSMITH_PROJECT=retrieval-agent

# LLM choice (set the API key for your selected provider):
ANTHROPIC_API_KEY=your_anthropic_api_key
FIREWORKS_API_KEY=your_fireworks_api_key
OPENAI_API_KEY=your_openai_api_key

# Retrieval provider (configure based on your chosen service):

## Elastic Cloud:
ELASTICSEARCH_URL=https://your_elastic_cloud_url
ELASTICSEARCH_API_KEY=your_elastic_api_key

## Elastic Local:
ELASTICSEARCH_URL=http://host.docker.internal:9200
ELASTICSEARCH_USER=elastic
ELASTICSEARCH_PASSWORD=changeme

## Pinecone:
PINECONE_API_KEY=your_pinecone_api_key
PINECONE_INDEX_NAME=your_pinecone_index_name

## MongoDB Atlas:
MONGODB_URI=your_mongodb_connection_string

# Cohere API key:
COHERE_API_KEY=your_cohere_api_key

• Beispieldatei .env (unter Verwendung von Elastic Cloud und Cohere)

Nachfolgend finden Sie eine Beispielkonfiguration .env für die Verwendung von Elastic Cloud als Abrufanbieter und Cohere als LLM, wie in diesem Blog gezeigt:

# To separate your traces from other applications
LANGSMITH_PROJECT=retrieval-agent
#Retrieval Provider
# Elasticsearch configuration
ELASTICSEARCH_URL=elastic-url:443
ELASTICSEARCH_API_KEY=elastic_api_key
# Cohere API key
COHERE_API_KEY=cohere_api_key

Hinweis: In dieser Anleitung wird Cohere sowohl für die Antwortgenerierung als auch für die Einbettung verwendet. Sie können je nach Anwendungsfall jedoch auch andere LLM-Anbieter wie OpenAI, Claude oder sogar ein lokales LLM-Modell verwenden. Stellen Sie sicher, dass alle .env Schlüssel, die Sie verwenden möchten, in der Datei vorhanden und korrekt festgelegt sind.

3. Aktualisieren Sie die Konfigurationsdatei -configuration.py

Nachdem Sie Ihre .env -Datei mit den entsprechenden API-Schlüsseln eingerichtet haben, besteht der nächste Schritt darin, die Standardmodellkonfiguration Ihrer Anwendung zu aktualisieren. Durch die Aktualisierung der Konfiguration wird sichergestellt, dass das System die Dienste und Modelle verwendet, die Sie in Ihrer .env -Datei angegeben haben.

Navigieren Sie zur Konfigurationsdatei:

 cd src/retrieval_graph

Die Datei configuration.py enthält die Standardmodelleinstellungen, die vom Abrufagenten für drei Hauptaufgaben verwendet werden:

• Einbettungsmodell – konvertiert Dokumente in Vektordarstellungen

• Abfragemodell – verarbeitet die Abfrage des Benutzers in einen Vektor

• Antwortmodell – generiert die endgültige Antwort

Standardmäßig verwendet der Code Modelle von OpenAI (z. B. openai/text-embedding-3-small) und Anthropic (z. anthropic/claude-3-5-sonnet-20240620 and anthropic/claude-3-haiku-20240307). In diesem Blog wechseln wir zur Verwendung von Cohere-Modellen. Wenn Sie bereits OpenAI oder Anthropic verwenden, sind keine Änderungen erforderlich.

Beispieländerungen (mit Cohere):

Öffnen Sie configuration.py und ändern Sie die Modellstandards wie unten gezeigt:

…
 embedding_model: Annotated[
       str,
       {"__template_metadata__": {"kind": "embeddings"}},
   ] = field(
       default="cohere/embed-english-v3.0",
…
response_model: Annotated[str, {"__template_metadata__": {"kind": "llm"}}] = field(
       default="cohere/command-r-08-2024",
…
query_model: Annotated[str, {"__template_metadata__": {"kind": "llm"}}] = field(
       default="cohere/command-r-08-2024",
       metadata={

Ausführen des Abrufagenten mit der LangGraph-Befehlszeilenschnittstelle

1. Starten Sie den LangGraph-Server

cd lg-agent-demo
langgraph dev

Dadurch wird der LangGraph-API-Server lokal gestartet. Wenn dies erfolgreich ausgeführt wird, sollten Sie etwa Folgendes sehen:

Öffnen Sie die URL der Studio-Benutzeroberfläche.

Es stehen zwei Diagramme zur Verfügung:

• Retrieval-Graph: Ruft Daten aus Elasticsearch ab und beantwortet die Anfrage mithilfe eines LLM.

• Indexer-Graph: Indiziert Dokumente in Elasticsearch und generiert Einbettungen mithilfe eines LLM.

2. Konfigurieren des Indexergraphen

• Öffnen Sie den Indexergraphen.

• Klicken Sie auf „Assistenten verwalten“.

  • Klicken Sie auf „Neuen Assistenten hinzufügen“, geben Sie die Benutzerdaten wie angegeben ein und schließen Sie anschließend das Fenster.

{"user_id": "101"}

3. Beispieldokumente indexieren

• Indizieren Sie die folgenden Beispieldokumente, die einen hypothetischen Quartalsbericht für die Organisation NoveTech darstellen:

[
  {    "page_content": "NoveTech Solutions Q1 2025 Report - Revenue: $120.5M, Net Profit: $18.2M, EPS: $2.15. Strong AI software launch and $50M government contract secured."
  },
  {
    "page_content": "NoveTech Solutions Business Highlights - AI-driven analytics software gained 15% market share. Expansion into Southeast Asia with two new offices. Cloud security contract secured."
  },
  {
    "page_content": "NoveTech Solutions Financial Overview - Operating expenses at $85.3M, Gross Margin 29.3%. Stock price rose from $72.5 to $78.3. Market Cap reached $5.2B."
  },
  {
    "page_content": "NoveTech Solutions Challenges - Rising supply chain costs impacting hardware production. Regulatory delays slowing European expansion. Competitive pressure in cybersecurity sector."
  },
  {
    "page_content": "NoveTech Solutions Future Outlook - Expected revenue for Q2 2025: $135M. New AI chatbot and blockchain security platform launch planned. Expansion into Latin America."
  },
  {
    "page_content": "NoveTech Solutions Market Performance - Year-over-Year growth at 12.7%. Stock price increase reflects investor confidence. Cybersecurity and AI sectors remain competitive."
  },
  {
    "page_content": "NoveTech Solutions Strategic Moves - Investing in R&D to enhance AI-driven automation. Strengthening partnerships with enterprise cloud providers. Focusing on data privacy solutions."
  },
  {
    "page_content": "NoveTech Solutions CEO Statement - 'NoveTech Solutions continues to innovate in AI and cybersecurity. Our growth strategy remains strong, and we foresee steady expansion in the coming quarters.'"
  }
]

Sobald die Dokumente indiziert sind, wird im Thread eine Löschmeldung angezeigt, wie unten dargestellt.

4. Ausführen des Abrufgraphen

• Wechseln Sie zum Abrufgraphen.

• Geben Sie die folgende Suchanfrage ein:

What was NovaTech Solutions total revenue in Q1 2025?

Das System gibt relevante Dokumente zurück und liefert eine genaue Antwort basierend auf den indizierten Daten.

Passen Sie den Abrufagenten an.

Um das Benutzererlebnis zu verbessern, führen wir einen Anpassungsschritt im Abrufgraphen ein, um die nächsten drei Fragen vorherzusagen, die ein Benutzer stellen könnte. Diese Vorhersage basiert auf Folgendem:

• Kontext aus den abgerufenen Dokumenten

• Vorherige Benutzerinteraktionen

• Letzte Benutzerabfrage

Zur Implementierung der Abfragevorhersagefunktion sind die folgenden Codeänderungen erforderlich:

1. Aktualisiere graph.py

• Funktion predict_query hinzufügen:

async def predict_query(
   state: State, *, config: RunnableConfig
) -> dict[str, list[BaseMessage]]:
   logger.info(f"predict_query predict_querypredict_query predict_query predict_query predict_query")  # Log the query

   configuration = Configuration.from_runnable_config(config)
   prompt = ChatPromptTemplate.from_messages(
       [
           ("system", configuration.predict_next_question_prompt),
           ("placeholder", "{messages}"),
       ]
   )
   model = load_chat_model(configuration.response_model)
   user_query = state.queries[-1] if state.queries else "No prior query available"
   logger.info(f"user_query: {user_query}")
   logger.info(f"statemessage: {state.messages}")
   #human_messages = [msg for msg in state.message if isinstance(msg, HumanMessage)]

   message_value = await prompt.ainvoke(
       {
           "messages": state.messages,
           "user_query": user_query,  # Use the most recent query as primary input
           "system_time": datetime.now(tz=timezone.utc).isoformat(),
       },
       config,
   )

   next_question = await model.ainvoke(message_value, config)
   return {"next_question": [next_question]}

• Ändern Sie die Funktion respond , um das Objekt response anstelle der Nachricht zurückzugeben:

async def respond(
   state: State, *, config: RunnableConfig
) -> dict[str, list[BaseMessage]]:
   """Call the LLM powering our "agent"."""
   configuration = Configuration.from_runnable_config(config)
   # Feel free to customize the prompt, model, and other logic!
   prompt = ChatPromptTemplate.from_messages(
       [
           ("system", configuration.response_system_prompt),
           ("placeholder", "{messages}"),
       ]
   )
   model = load_chat_model(configuration.response_model)

   retrieved_docs = format_docs(state.retrieved_docs)
   message_value = await prompt.ainvoke(
       {
           "messages": state.messages,
           "retrieved_docs": retrieved_docs,
           "system_time": datetime.now(tz=timezone.utc).isoformat(),
       },
       config,
   )
   response = await model.ainvoke(message_value, config)
   # We return a list, because this will get added to the existing list
   return {"response": [response]}

• Aktualisieren Sie die Graphstruktur, um einen neuen Knoten und eine neue Kante für predict_query hinzuzufügen:

builder.add_node(generate_query)
builder.add_node(retrieve)
builder.add_node(respond)
builder.add_node(predict_query)
builder.add_edge("__start__", "generate_query")
builder.add_edge("generate_query", "retrieve")
builder.add_edge("retrieve", "respond")
builder.add_edge("respond", "predict_query")

2. Aktualisiere prompts.py

• Erstelle eine Eingabeaufforderung für eine Vorhersage in prompts.py:

PREDICT_NEXT_QUESTION_PROMPT = """Given the user query and the retrieved documents, suggest the most likely next question the user might ask.

**Context:**
- Previous Queries:
{previous_queries}

- Latest User Query: {user_query}

- Retrieved Documents:
{retrieved_docs}

**Guidelines:**
1. Do not suggest a question that has already been asked in previous queries.
2. Consider the retrieved documents when predicting the next logical question.
3. If the user's query is already fully answered, suggest a relevant follow-up question.
4. Keep the suggested question natural and conversational.
5. Suggest at least 3 question

System time: {system_time}"""

3. Aktualisieren Sie die Datei configuration.py

• Fügen Sie predict_next_question_prompt hinzu:

predict_next_question_prompt: str = field(
       default=prompts.PREDICT_NEXT_QUESTION_PROMPT,
       metadata={"description": "The system prompt used for generating responses."},
   )

4. Aktualisiere state.py

• Fügen Sie die folgenden Attribute hinzu:

response: Annotated[Sequence[AnyMessage], add_messages]
next_question : Annotated[Sequence[AnyMessage], add_messages]

5. Führen Sie den Abrufgraphen erneut aus.

• Geben Sie die folgende Suchanfrage erneut ein:

What was NovaTech Solutions total revenue in Q1 2025?

Das System verarbeitet die Eingabe und sagt drei verwandte Fragen voraus, die Benutzer stellen könnten, wie unten gezeigt.

Fazit

Die Integration der Retrieval Agent-Vorlage in LangGraph Studio und CLI bietet mehrere wichtige Vorteile:

• Beschleunigte Entwicklung: Die Vorlagen- und Visualisierungstools optimieren die Erstellung und das Debuggen von Abruf-Workflows und verkürzen so die Entwicklungszeit.

• Nahtlose Bereitstellung: Integrierte Unterstützung für APIs und automatische Skalierung gewährleistet eine reibungslose Bereitstellung in allen Umgebungen.

• Einfache Updates: Das Ändern von Arbeitsabläufen, das Hinzufügen neuer Funktionen und die Integration zusätzlicher Knoten ist einfach, wodurch der Abrufprozess leichter skaliert und verbessert werden kann.

• Permanenter Speicher: Das System behält Agentenzustände und -wissen bei und verbessert so Konsistenz und Zuverlässigkeit.

• Flexible Workflow-Modellierung: Entwickler können Abruflogik und Kommunikationsregeln für bestimmte Anwendungsfälle anpassen.

• Interaktion und Debugging in Echtzeit: Die Möglichkeit zur Interaktion mit laufenden Agenten ermöglicht effizientes Testen und Problemlösen.

Durch die Nutzung dieser Funktionen können Unternehmen leistungsstarke, effiziente und skalierbare Abrufsysteme erstellen, die die Datenzugänglichkeit und das Benutzererlebnis verbessern.

Der vollständige Quellcode für dieses Projekt ist auf GitHub verfügbar.