Агентський ШІ з LangChain — Частина 6: Агентський RAG на основі MCP

У цьому посібнику описано процес створення системи від сировини до готового продукту для: Агентного ШІ з LangChain — Частина 6: Агентний RAG з MCP та LangGraph. Основна увага приділяється крокам виконання, чітким перевіркам та коду, який можна просто додати до репозиторію без необхідності здогадуватися щодо його призначення. Для загального огляду необхідно визначити вхідні дані, виконавця кроку та критерії завершення перед зміною коду. Виконавці мають мати можливість перезапустити крок з відомої точки контролю, не намагаючись здогадатися про прихований стан системи. Розглядайте цю стадію як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте безповідомного часткового завершення роботи.

Технічні концепції

Під час вивчення технічних концепцій спочатку запишіть умови використання: необхідні вхідні дані, сигнал про успішне виконання та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Запишіть час виконання та витрати на токени або запити поруч із результатами функціональності. Чітке бачення витрат заздалегідь запобігає несподіваним рахункам, коли процес переходить від демо-версії до спільних середовищ. Вимірюйте рівень точності відповідей на фіксований набір запитань перед налаштуванням промптів. Часта зміна промптів рідко допомагає покращити якість пошуку інформації.

1. Конфігурації системи

Під час роботи над розділом «1. Конфігурації системи» спочатку запишіть умови використання: необхідні дані вхіду, сигнал про успішне виконання та наслідки часткової невдачі. Такий перелік допоможе зберегти чесність подальших змін у коді. Тримайте конфігурацію окремо від коду програми. Файли середовища, бази зберігання секретних даних та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь код. Перед налаштуванням запитів вимірюйте рівень відтворення інформації на фіксованому наборі запитань. Зміна формулювань запитів рідко вирішує проблеми слабкого пошуку даних.

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. Структура проекту

Під час роботи над розділом 2. «Структура проекту» спочатку запишіть умови виконання: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Одночасно задокументуйте шлях успішного виконання та шлях відновлення. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації. Перед налаштуванням запитів вимірюйте рівень точності відповідей на фіксованому наборі запитань. Зміна запитів рідко допомагає вирішити проблеми з низькою ефективністю пошуку інформації.

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)

Під час роботи над 3. MCP (Model Context Protocol) спочатку запишіть умови використання: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Віддавайте перевагу невеликим, тестованим одиницям коду перед об’ємними скриптами. Якщо якийсь крок зазнає невдачі, причина має вказувати на конкретну функцію, а не на складну послідовність операцій. Зберігайте у кеші стабільні інструкції системи та схеми інструментів. Повторна передача ідентичних даних є поширеною причиною непотрібних витрат ресурсів.

3.1. Концепція

Під час роботи над розділом 3.1. Концепція спочатку запишіть умови контракту: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Дайте назви елементам, визначте критерії успіху та не допускайте беззвучного часткового виконання завдань. Перед налаштуванням запитів вимірюйте рівень запам’ятовування на фіксованому наборі запитань. Часта зміна формулювань запитів рідко виправляє проблеми з недостатньою ефективністю пошуку.

3.2 Архітектура MCP

Під час роботи над розділом 3.2 «Архітектура MCP» спочатку запишіть умови взаємодії: необхідні вхідні дані, сигнал про успішне виконання та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Запишіть час виконання та витрати на токени або запити поруч із функціональними результатами. Чітке бачення витрат заздалегідь запобігає несподіваним рахункам, коли процес переходить від демо-середовища до спільних. Вимірюйте рівень точності відповідей на фіксований набір запитань перед налаштуванням підказок. Часта зміна підказок рідко допомагає покращити якість пошуку інформації.

3.3. Режими передачі даних MCP

Під час роботи над розділом 3.3. MCP Transport Modes спочатку запишіть умови взаємодії: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Тримайте конфігурацію окремо від коду додатку. Файли середовища, сховища секретних даних та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь код. Вимірюйте рівень відтворення інформації на фіксованому наборі запитань перед налаштуванням підказок. Зміна підказок рідко вирішує проблеми слабкого пошуку даних.

3.3.1. Режим STDIO

Під час роботи над розділом 3.3.1. Режим STDIO спочатку запишіть умови взаємодії: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Одночасно задокументуйте шлях успішного виконання та шлях відновлення. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації. Перед налаштуванням запитів вимірюйте рівень точності відповідей на фіксованому наборі запитань. Зміна запитів рідко допомагає покращити якість пошуку. Під час роботи над розділом 3.3.1. Режим STDIO спочатку запишіть умови взаємодії: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Розглядайте цей етап як угоду між вхідними даними та перевіреними результатами. Назвіть всі елементи, визначте критерії успіху та не допускайте безповідомного часткового виконання завдань.

3.3.2. Режим SSE

3.3.2. Режим SSE працює найкраще, коли його розглядають як вимірювану поверхню. Запишіть один ідеальний приклад роботи, один випадок збою та примітки щодо скасування змін перед розширенням обсягу. Записуйте час виконання та витрати на токени чи запити поруч із функціональними результатами. Візуалізація витрат заздалегідь запобігає несподіваним рахункам під час переходу з демо-середовища у спільні. Розділіть політику часткового оброблення даних від політики їх отримання. Зміна однієї з них не повинна змушувати переписувати іншу при зміні показників якості.

4. Впровадження MCP у режимі STDIO

1. Впровадження MCP у режимі STDIO найкраще функціонує, якщо його розглядати як вимірювану систему. Збережіть один ідеальний зразок виконання, один випадок збою та примітки щодо скасування змін перед розширенням обсягу роботи. Тримайте конфігурацію окремо від коду програми. Файли середовища, сховища конфіденційних даних та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь код. Розділіть політику розбиття на частини та політику отримання даних. Зміна однієї з них не повинна змушувати переписувати іншу при зміні показників якості.

4.1. Сервер MCP

4.1. MCP Server працює найкраще, коли його розглядають як вимірювану систему. Збережіть один ідеальний запис, один випадок збою та примітку щодо скасування змін перед розширенням обсягу роботи. Документуйте як успішний, так і відновлювальний сценарії роботи. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації. Розділіть політику часткової обробки даних від політики їх отримання. Зміна однієї з них не повинна змушувати переписувати іншу при зміні показників якості. 4.1. MCP Server працює найкраще, коли його розглядають як вимірювану систему. Збережіть один ідеальний запис, один випадок збою та примітку щодо скасування змін перед розширенням обсягу роботи. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте безповідомного часткового виконання завдань.

# 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

Для версії 4.2 клієнта MCP необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись визначити прихований стан. Записуйте час виконання та витрати на токени або запити поруч із функціональними результатами. Чітке відображення витрат заздалегідь запобігає несподіваним рахункам під час переходу з демо-середовища у спільні. Наводьте конкретні уривки тексту, які лягли в основу відповіді. Без посилань оператори не зможуть відрізнити галюцинації від проблем з індексуванням.

# 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. Використання інструментів MCP у агенті

Для розділу 4.3. При використанні інструментів MCP у агенті необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись визначити прихований стан. Конфігурацію слід тримати окремо від коду програми. Файли середовища, сховища конфіденційних даних та флаги функціоналу мають знаходитися в одному місці, де оператори можуть їх перевірити, не читаючи весь код. Необхідно наводити конкретні уривки тексту, на яких ґрунтується відповідь. Без посилань оператори не зможуть відрізнити галюцинації від проблем з індексуванням.

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. Тестування робочого процесу STDIO

Для розділу 4.4. Тестування робочого процесу STDIO необхідно визначити вхідні дані, відповідальну особу за кожен крок та критерії завершення перед зміною коду. Оператори повинні мати можливість знову виконати крок з відомої точки контролю, не намагаючись вгадати прихований стан. Необхідно документувати як успішний, так і відновлювальний сценарії роботи. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації. Наводьте конкретні уривки тексту, які лежать в основі відповіді. Без посилань оператори не зможуть відрізнити галюцинації від проблем з індексуванням. Для розділу 4.4. Тестування робочого процесу STDIO необхідно визначити вхідні дані, відповідальну особу за кожен крок та критерії завершення перед зміною коду. Оператори повинні мати можливість знову виконати крок з відомої точки контролю, не намагаючись вгадати прихований стан. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте мовчазного часткового виконання завдань.

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. Впровадження MCP у режимі SSE

Під час роботи над розділом 5. Впровадження MCP у режимі SSE спочатку запишіть контракт: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Записуйте час виконання та витрати на токени або запити поруч із функціональними результатами. Відображення витрат на ранньому етапі запобігає несподіваним рахункам, коли система переходить від демо-режиму до спільних середовищ. Перевірте ефективність пошуку на фіксованому наборі запитань перед налаштуванням підказок. Часта зміна підказок рідко допомагає покращити якість пошуку.

5.1. Сервер MCP

Під час роботи над розділом 5.1. MCP Server спочатку запишіть умови взаємодії: необхідні вхідні дані, сигнал про успішне виконання та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Зберігайте конфігурацію окремо від коду додатку. Файли середовища, храни секретів та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь код. Перед налаштуванням запитів вимірюйте рівень відтворення інформації на фіксованому наборі запитань. Зміна формулювань запитів рідко допомагає покращити якість пошуку.

# 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

Під час роботи над розділом 5.2. MCP Client спочатку запишіть умови взаємодії: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Одночасно задокументуйте шлях успішного виконання та шлях відновлення. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації. Перед налаштуванням запитів вимірюйте рівень точності відповідей на фіксованому наборі запитань. Зміна запитів рідко допомагає покращити якість отримання інформації. Під час роботи над розділом 5.2. MCP Client спочатку запишіть умови взаємодії: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Розглядайте цей етап як угоду між вхідними даними та перевіреними результатами. Назвіть всі елементи, визначте критерії успіху та не допускайте безповідомного часткового виконання завдань.

# 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. Використання інструментів MCP у StateGraph

5.3. Використання інструментів MCP у StateGraph найкраще функціонує, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу роботи. Записуйте час виконання та витрати на токени або запити поруч із функціональними результатами. Візуалізація витрат на ранньому етапі запобігає несподіваним рахункам, коли процес переходить від демо-середовища до спільних. Розділіть політику часткового оброблення даних від політики їх отримання. Зміна однієї з них не повинна змушувати переписувати іншу при зміні показників якості.

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. Чому реалізація SSE відрізняється від STDIO

5.4. Чому реалізація SSE відрізняється від STDIO: найкраще розглядати її як вимірювану поверхню. Запишіть один ідеальний приклад роботи, один випадок збою та примітки щодо скасування змін перед розширенням обсягу. Тримайте конфігурацію окремо від коду додатку. Файли середовища, сховища секретів та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь код.

6. Інтеграція клієнта SSE MCP з 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. Перебудова StateGraph у частині 5

7.1. Кінцева точка FastAPI

@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))

7.2. Фронтенд Streamlit

8. Інструменти, керовані MCP, проти локальних інструментів LangChain

9. Підсумковий огляд

Чек-лист для експлуатації