Agentenbasierte KI mit LangChain – Teil 6: MCP-betriebene agentenbasierte RAG
Teil 6 von agiler KI mit LangChain: agiler RAG, angetrieben von MCP, wobei die Abrufwerkzeuge von einem MCP-Server stammen und LangGraph den Agentenfluss orchestriert.
Quellen: Agentic AI with LangChain — Part 6: MCP Powered Agentic RAG with LangGraph
Dieser Leitfaden zeigt Schritt für Schritt den Weg von Rohstoffen bis zu einem funktionsfähigen System für: Agentic AI mit LangChain – Teil 6: MCP-basiertes agitives RAG mit LangGraph. Der Schwerpunkt liegt auf ausführbaren Schritten, expliziten Überprüfungen sowie Code, den Sie ohne Rätseln über die Absicht direkt in ein Repository einfügen können. Zur Übersicht sollten Sie vor dem Ändern des Codes die Eingaben, den Verantwortlichen für den Schritt sowie die Abbruchkriterien definieren. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Erzeugnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab.
Technische Konzepte
Beim Arbeiten an Technischen Konzepten sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Notieren Sie die Laufzeiten sowie die Kosten für Tokens oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Weg von einer Demo in gemeinsame Umgebungen wechselt. Messen Sie die Trefferquote anhand einer festgelegten Fragestellung, bevor Sie die Anfragen anpassen. Ein häufiges Wechseln der Anfragen behebt selten ein schwaches Suchverhalten.
1. Systemkonfigurationen
Beim Arbeiten an „1. Systemkonfigurationen“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort zusammengefasst sein, den Betreiber ohne das Durchlesen des gesamten Systems überprüfen können. Messen Sie die Trefferquote anhand eines festgelegten Fragekatalogs, bevor Sie die Anfragen anpassen. Eine häufige Änderung der Anfragen behebt selten ein schwaches Abrufverhalten.
OPENAI_API_KEY="<Your OpenAI API Key>"
VECTOR_DB=pinecone
PINECONE_API_KEY=<your Pinecone API Key>
PINECONE_ENVIRONMENT=gcp-starter
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"
TAVILY_API_KEY=<Your TAVILY API Key>
class BaseConfig(BaseSettings):
OPENAI_API_KEY: Optional[str]
PINECONE_API_KEY: Optional[str]
TAVILY_API_KEY: Optional[str]
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
2. Projektstruktur
Beim Arbeiten am Abschnitt „2. Projektstruktur“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikator sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie gleichzeitig den erfolgreichen Ablauf sowie den Notfallweg. Wiederholte Versuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst, nicht zu späteren Optimierungen. Messen Sie die Trefferquote anhand eines festgelegten Fragekatalogs, bevor Sie die Anfragenanweisungen anpassen – eine häufige Änderung dieser Anweisungen behebt in der Regel nicht ein schwaches Suchverhalten.
project/
├── rag_process/ # Pinecone RAG operations
│ ├── __init__.py
│ ├── extractor.py
| └── service.py
│
├── graphs # implement ReAct and Reflection StateGraph
│ ├── nodes.py # graph nodes. Defined as Python functions
│ ├── graph.py # graph workflow for MCP in SES mode
│ ├── graph_stdio.py # graph workflow for MCP in STDIO mode
│ ├── consts.py # define node names as constants
│ └── state.py # typed dictionary to store graph states
│
├── mcp_client_sse.py # mcp client in sse mode
├── mcp_client_stdio.py # mcp client in stdio mode
├── server_sse.py # mcp server in sse mode
│── server_stdio.py # mcp server in stdio mode
|── schemas.py # pydantic data models
|── upload.py # python functions to upload files asynchronously
├── main.py # fastAPI app
├── config.py # pydantic BaseConfig
│── utils.py # utility extracting question/docs from message list
├── .env # environment variable definition
├── client.py # streamlit client
├── .gitignore
└── requirements.txt # package requirements
3. MCP (Model Context Protocol)
Beim Arbeiten mit dem 3. MCP (Model Context Protocol) sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Cachen Sie stabile Systemanweisungen sowie Tool-Schemata. Das erneute Senden identischer Preambles ist eine häufige Ursache für Ressourcenverschwendung.
3.1. Konzept
Beim Arbeiten an Abschnitt 3.1 „Konzept“ sollten Sie zunächst einen Vertrag aufschreiben: die erforderlichen Eingaben, das Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Messen Sie die Erinnerungsfähigkeit anhand eines festgelegten Fragekatalogs, bevor Sie die Anfragen anpassen. Eine häufige Änderung der Anfragen behebt selten ein schwaches Suchverhalten.
3.2 MCP-Architektur
Beim Arbeiten an der 3.2 MCP-Architektur sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Notieren Sie außerdem die Laufzeiten sowie die Kosten für Tokens oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Einsatzbereich von Demoumgebungen in gemeinsam genutzte Umgebungen wechselt. Messen Sie außerdem die Trefferquote anhand eines festgelegten Fragekatalogs, bevor Sie die Anfragen anpassen – eine häufige Änderung der Anfragemuster behebt in der Regel nicht ein schwaches Suchverhalten.
3.3. MCP-Transportmodi
Beim Arbeiten an Abschnitt 3.3. MCP-Transportmodi sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreiber überprüfen können, ohne den gesamten Code durchzulesen. Messen Sie die Trefferquote anhand einer festgelegten Fragestellung, bevor Sie die Prompts anpassen. Eine häufige Änderung der Prompts behebt selten ein schwaches Suchverhalten.
3.3.1. STDIO-Modus
Beim Arbeiten an Abschnitt 3.3.1, STDIO-Modus, sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Wiederherstellungsprozess gemeinsam. Wiederholte Versuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen. Messen Sie die Trefferquote anhand eines festgelegten Fragekatalogs, bevor Sie die Anfragenanpassungen vornehmen – eine häufige Änderung der Anfragen löst in der Regel kein schwaches Abrufverhalten aus. Beim Arbeiten an Abschnitt 3.3.1, STDIO-Modus, sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisdokumente, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab.
3.3.2. SSE-Modus
3.3.2. Der SSE-Modus funktioniert am besten, wenn er als messbare Oberfläche betrachtet wird. Erfassen Sie ein erfolgreiches Beispiel, einen Fehlerfall sowie die Notizen zur Rücksetzung, bevor Sie den Umfang erweitern. Erfassen Sie außerdem die Laufzeiten sowie die Kosten für Token oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn sich der Einsatzbereich von einer Demo-Umgebung in gemeinsam genutzte Umgebungen verschiebt. Trennen Sie die Strategie zur Aufteilung in Blöcke von der Strategie zum Abrufen. Ein Änderungsbedarf bei einer dieser Strategien sollte nicht dazu führen, dass die andere neu geschrieben werden muss, wenn sich die Qualitätsmetriken ändern.
4. Implementierung von MCP im STDIO-Modus
- Die Implementierung von MCP im STDIO-Modus funktioniert am besten, wenn sie als messbarer Ansatz betrachtet wird. Erfassen Sie ein optimales Beispiel, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreiber ohne das Durchlesen des gesamten Systems prüfen können. Trennen Sie die Chunking-Strategie von der Abrufstrategie. Ein Änderung in einer sollte nicht dazu führen, dass die andere neu geschrieben werden muss, wenn sich die Qualitätsmetriken ändern.
4.1. MCP-Server
4.1. Der MCP Server funktioniert am besten, wenn er als messbarer Bereich betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Dokumentieren Sie gleichzeitig den erfolgreichen Ablauf sowie den Wiederherstellungsprozess. Wiederholversuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen. Trennen Sie die Strategie zur Aufteilung in Blöcke von der Strategie zum Abrufen. Ein Veränderungsbedarf bei einer dieser Strategien sollte nicht dazu führen, dass die andere neu geschrieben werden muss, wenn sich die Qualitätsmetriken ändern. 4.1. Der MCP Server funktioniert am besten, wenn er als messbarer Bereich betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die relevanten Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab.
# initialize an STDIO FastMCP instance
mcp = FastMCP("Search Server")
# create mcp tools
@mcp.tool()
async def search_agentic_docs(query: str) -> str:
"""
Good for questions about generative agents, prompt engineering, and adversarial attacks.
Pass a natural language search query to retrieve context from the agentic ai database.
"""
# Simply invoke your existing vector service retriever
docs = await vector_service.retriever.ainvoke(query)
# Flatten the document content into a single string for the LLM
return "\n\n".join([d.page_content for d in docs])
# start MCP server in stdio mode
if __name__ == "__main__":
mcp.run(transport="stdio")
4.2. MCP Client
Für den 4.2. MCP Client sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Codeändern definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Erfassen Sie die Laufzeiten sowie die Kosten für Token oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Weg von Demonstrumgebungen in gemeinsam genutzte Umgebungen wechselt. Zitieren Sie die Passagen, die tatsächlich die Antwort untermauern. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden.
# define location of server file
BASE_DIR = Path(__file__).parent
SERVER_SCRIPT = BASE_DIR / "server.py"
# Local server running directly via Python over stdio
SERVER_CONFIG = {
"local_search_server": {
"transport": "stdio",
"command": "python",
"args": [str(SERVER_SCRIPT)],
},
}
# create a MCP client and return
def get_mcp_client() :
"""Works for stdio, sse, or a mix of both!"""
client = MultiServerMCPClient(SERVER_CONFIG)
return client
4.3. Verwendung von MCP-Tools im Agenten
Für 4.3: Bei der Verwendung von MCP-Tools im Agenten sollten Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Die Konfiguration sollte außerhalb des Anwendungscode gespeichert werden. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort zusammengefasst sein, den die Operator überprüfen können, ohne den gesamten Ablauf durchzulesen. Zitieren Sie die Passagen, die tatsächlich die Antwort untermauern. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden.
from mcp_client_stdio import get_mcp_client
async def build_graph():
"""
build StateGraph
:return:
"""
# initialize a MCP client instance
client = get_mcp_client()
# retrieve MCP tools
mcp_tools = await client.get_tools()
# create an llM
llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
tools = [TavilySearch(max_results=3, tavily_api_key=TAVILY_API_KEY)] + mcp_tools
# bind tools to ToolNode and llm
tool_node = ToolNode(tools)
workflow = StateGraph(GraphState)
llm_with_tools = llm.bind_tools(tools)
4.4. Testen des STDIO-Ablaufs
Für 4.4. Das Testen des STDIO-Ablaufs: Definieren Sie die Eingaben, den Verantwortlichen für den Schritt sowie die Abbruchkriterien, bevor Sie Code ändern. Die Bediener sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallweg. Wiederholungsversuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen. Zitieren Sie die Passagen, die tatsächlich die Antwort begründen. Ohne Zitate können die Bediener nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden. Für 4.4. Das Testen des STDIO-Ablaufs: Definieren Sie die Eingaben, den Verantwortlichen für den Schritt sowie die Abbruchkriterien, bevor Sie Code ändern. Die Bediener sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab.
question_1 = "what is prompt engineer ?"
async def main():
print("hello ReAct agent by Langraph")
message = HumanMessage(content=question_1)
app = await build_graph()
res = await app.ainvoke({"messages": [message]})
if __name__ == "__main__":
asyncio.run(main())
5. Implementierung von MCP im SSE-Modus
Beim Arbeiten an 5. Implementierung von MCP im SSE-Modus sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Notieren Sie die Laufzeiten sowie die Kosten für Token oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Einsatzbereich von einer Demo in gemeinsame Umgebungen wechselt. Messen Sie die Trefferquote anhand einer festgelegten Fragestellung, bevor Sie die Anfragen anpassen. Häufige Änderungen der Anfragen beheben selten ein schwaches Retrieval-System.
5.1. MCP-Server
Beim Arbeiten an 5.1. MCP Server sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreiber ohne das Durchlesen des gesamten Systems überprüfen können. Messen Sie die Trefferquote anhand eines festgelegten Fragekatalogs, bevor Sie die Prompts anpassen. Eine häufige Änderung der Prompts behebt selten ein schwaches Suchsystem.
# for SSE
mcp = FastMCP(
name="My MCP Server",
host="0.0.0.0",
port=8001
)
# implement mcp tools here
if __name__ == "__main__":
mcp.run(transport="sse")
5.2. MCP Client
Beim Arbeiten an 5.2. MCP Client sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Wiederherstellungsprozess gemeinsam. Wiederholte Versuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen. Messen Sie die Trefferquote anhand eines festgelegten Fragekatalogs, bevor Sie die Anfragen anpassen – ein häufiges Wechseln der Anfragen löst in der Regel kein schwaches Suchverhalten aus. Beim Arbeiten an 5.2. MCP Client sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab.
# specify config of the SSE MCP server that the client will connect to
SERVER_CONFIG = {
# config for mcp by sse transportation
"fastapi_mcp_server": {
"transport": "sse",
"url": "http://localhost:8001/sse",
}
}
# create a client
def get_mcp_client() :
client = MultiServerMCPClient(SERVER_CONFIG)
return client
5.3. Verwendung von MCP-Tools in StateGraph
5.3. Die Verwendung von MCP-Tools in StateGraph funktioniert am besten, wenn sie als messbare Oberfläche betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein optimales Beispiel, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Erfassen Sie außerdem die Laufzeiten sowie die Kosten für Tokens oder Abfragen zusammen mit den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn sich der Einsatzbereich von einer Demo-Umgebung in gemeinsam genutzte Umgebungen verschiebt. Trennen Sie die Strategie zur Aufteilung in Blöcke von der Strategie zur Abrufung. Ein Änderungsbedarf bei einer dieser Strategien sollte nicht dazu führen, dass die andere neu geschrieben werden muss, wenn sich die Qualitätsmetriken ändern.
async def build_graph(mcp_tools: list):
"""
build StateGraph
:return:
"""
# create an llm
llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
tools = [TavilySearch(max_results=3, tavily_api_key=TAVILY_API_KEY)] + mcp_tools
# bind tools to ToolNode and llm
tool_node = ToolNode(tools)
llm_with_tools = llm.bind_tools(tools)
5.4. Warum die SSE-Implementierung sich von STDIO unterscheidet
5.4. Warum die SSE-Implementierung sich von STDIO unterscheidet: Am besten funktioniert sie, wenn sie als messbarer Ansatz betrachtet wird. Erfassen Sie ein gelungenes Beispiel, einen Fehlerfall sowie die Notizen zur Rücksetzung, bevor Sie den Umfang erweitern. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreiber ohne das Durchlesen des gesamten Systems überprüfen können.
6. Integration des SSE MCP-Clients mit FastAPI
from mcp_client_sse import get_mcp_client
# CREATES a manager for FastAPI startup/shutdown
@asynccontextmanager
async def lifespan(fastapi_app: FastAPI):
# create an MCP client, and retrieve MCP tools
mcp_client = get_mcp_client()
mcp_tools = await mcp_client.get_tools()
# Compile graph dynamically with tools and store it on app state
fastapi_app.state.agent = await build_graph(mcp_tools)
# PAUSES here while FastAPI runs and handles requests
yield
app = FastAPI(lifespan=lifespan)
7. Wiederherstellung des StateGraph aus Teil 5
7.1. FastAPI-Endpunkt
@app.post("/generate_text", response_model=RAGResponse)
async def query_by_RAG_controller(request: Request, body: RAGRequest=Body(...)) -> RAGResponse:
try:
# wrap input query in a HumanMessage
message = HumanMessage(content=body.question)
# loads the StateGraph agent from the request
# this agent is created and maintained by lifespan context
agent = request.app.state.agent
# Async invokes the agent and returns the response
generation = await agent.ainvoke({"messages": [message]})
response = {"answer": generation.get("generation", "")}
return RAGResponse(**response)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))