Агентные ИИ с LangChain Часть 1

В этом руководстве показано, как построить цепочку от сырья до функционирующей системы для: Агентного ИИ с LangChain — Часть 1: Создание семантической пайплайны RAG. Основное внимание уделяется практическим шагам, четкой проверке и коду, который можно просто добавить в репозиторий без необходимости угадывать намерения автора. Для получения обзора определите входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не пытаясь угадать скрытое состояние. Документируйте как успешный ход выполнения, так и способы восстановления после сбоев. Повторные попытки, проверка человеком и обработка неработоспособных сообщений являются частью продукта, а не этапом последующей доработки.

Технологическая стек

При работе над технологической стеком сначала запишите условия работы: необходимые входные данные, сигнал успешного выполнения и что происходит при частичной неудаче. Такой чек-лист помогает сохранять честность при последующих изменениях кода. Предпочитайте небольшие, тестируемые модули большим скриптам. Когда какой-то шаг терпит неудачу, ошибка должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Оцените уровень воспроизводимости ответов на фиксированном наборе вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска.

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

При работе над разделом «1. Конфигурации системы» сначала запишите условия соглашения: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Рассматривайте этот этап как соглашение между входными данными и проверенными выходными результатами. Дайте названия соответствующим элементам, определите критерии успешности и не допускайте молчаливого частичного выполнения задачи. Оцените уровень воспроизводимости на фиксированном наборе вопросов перед настройкой подсказок. Частая смена подсказок редко помогает улучшить качество поиска.

OPENAI_API_KEY="<Your OpenAI API Key>"
VECTOR_DB=pinecone
Hugging_Face_TOKEN=<your huggingface token>
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]

    """Loads the dotenv file. Including this is necessary to get
    pydantic to load a .env file."""
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

2. Структура проекта

При работе над разделом 2. Структура проекта сначала запишите условия взаимодействия: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Рядом с функциональными результатами записывайте время выполнения и стоимость токенов или запросов. Очевидность затрат с самого начала предотвращает неожиданные расходы при переходе с демо-среды в общедоступные среды. Перед настройкой подсказок измерьте точность воспроизведения ответов на фиксированный набор вопросов. Частая смена подсказок редко помогает улучшить качество поиска.

project/
├── rag_process/
│   ├── __init__.py
│   ├── extractor.py
|   └── service.py
│
├── chains.py           # llm chains
├── nodes.py            # python functions utlizing llm chains
|── schemas.py          # pydantic data models
|── upload.py           # python functions to upload files asynchronously
├── main.py             # fastAPI app
├── config.py           # pydantic BaseConfig
│── dependencies.py     # dependency inject to fastAPI endpoint function
├── test_chains.py      # pytests
├── .env                # environment variable definition
├── client.py           # streamlit client
├── .gitignore
└── requirements.txt     # package requirements

При работе над разделом 2. Структура проекта сначала запишите условия взаимодействия: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Одновременно задокументируйте успешный сценарий работы и сценарий восстановления. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки.

2.1. Цепочки LLM

2.1. Цепочки LLM работают наилучшим образом, когда их рассматривают как измеримую структуру. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема задачи. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. При сбое какого-либо шага причина должна быть связана с конкретной ответственностью, а не с запутанной цепочкой операций. Установите лимит токенов на один ход и на всю сессию. Инструменты агентного типа активно расширяют объем контекста; жесткие ограничения предотвращают появление неожиданных счетов.

2.2. Symantic RAG

2.2. Семантический RAG работает наилучшим образом, когда рассматривается как измеримая структура. Соберите один идеальный пример обработки данных, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объема работ. Рассматривайте этот этап как контракт между входными данными и проверенными результатами. Дайте названия элементам данных, определите критерии успешности и не соглашайтесь на молчаливое частичное выполнение задачи. Разделяйте политику разбиения данных на части и политику их поиска. Изменение одной из них не должно приводить к переписыванию другой при изменении показателей качества.

2.3. Бэкенд FastAPI

2.3. FastAPI Backend работает наилучшим образом, когда рассматривается как объект с измеримыми показателями. Перед расширением объема работы необходимо зафиксировать один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию. Регистрируйте время выполнения операций, а также стоимость токенов или запросов вместе с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные расходы при переходе с демо-среды в общедоступные среды. Разделяйте политику разбиения данных на части и политику их извлечения. Изменение одной из них не должно приводить к необходимости переписывания другой при изменении показателей качества. 2.3. FastAPI Backend работает наилучшим образом, когда рассматривается как объект с измеримыми показателями. Перед расширением объема работы необходимо зафиксировать один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию. Документируйте одновременно успешный сценарий работы и сценарий восстановления. Повторные попытки, проверка человеком и обработка неработающих сообщений являются частью продукта, а не элементами последующей доработки.

2.4. Клиент Streamlit

Для версии 2.4 клиента Streamlit необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не пытаясь угадать скрытое состояние. Лучше использовать небольшие, тестируемые модули вместо обширных скриптов. При сбое шага причина должна быть связана с конкретной функцией, а не с запутанной структурой обработки данных. Указывайте те части текста, на которых основан ответ. Без цитат операторы не смогут отличить галлюцинации от проблем с индексацией.

3. Цепочки LLM с LCEL

Для цепочек LLM версии 3 с использованием LCEL необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг, исходя из известной точки контроля, без необходимости угадывать скрытое состояние. Рассматривайте этот этап как контракт между входными данными и проверенными результатами. Укажите названия элементов, определите критерии успеха и не допускайте молчаливого частичного выполнения задачи. При следующем шаге, представляющем собой код или вызов инструмента, предпочтительнее использовать структурированные результаты с проверкой по шаблону вместо свободного текста.

3.1. Язык выражений LangChain (LCEL)

Для 3.1. Языка выражений LangChain (LCEL) необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Записывайте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости заранее предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. Указывайте те участки текста, которые фактически легли в основу ответа. Без цитат операторы не могут отличить галлюцинации от пробелов в индексации. Для 3.1. Языка выражений LangChain (LCEL) необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Документируйте одновременно успешный путь выполнения и пути восстановления. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не чем-то, что добавляется позже.

р польский.

class GradeDocuments(BaseModel):
    """Binary score for relevance check on retrieved documents."""

    binary_score: bool = Field(
        description="Documents are relevant to the question, 'yes' or 'no'"
    )


async def retrieval_grader(document: Document, question: str):
    llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
    structured_llm_grader = llm.with_structured_output(GradeDocuments)

    system = """You are a grader assessing relevance of a retrieved document to a user question. \n
    If the document contains keyword(s) or semantic meaning related to the question, grade it as relevant.\n
    Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question. \n
    if the document is relevant, give it a score 'yes', otherwise give it a score 'no'
    """

    grade_prompt = ChatPromptTemplate.from_messages(
        [
            ("system", system),
            ("human", "Retrieved document: \n\n {document} User question: {question}"),

        ]
    )

    retrieval_grader = grade_prompt | structured_llm_grader
    grade = await retrieval_grader.ainvoke({"document": document, "question": question})

    return grade.binary_score

4. Семантический RAG с PineCone

4.1. Процесс ввода данных

4.2. Процесс поиска

4.3. Класс VectorService

4.3.1. Инициализация

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore

class VectorService:
    def __init__(self, index_name: str = "knowledgebase"):
        self.index_name = index_name

        # Embedding model
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small", api_key=api_key)

        # Pinecone vectorstore wrapper
        self.vectorstore = PineconeVectorStore(
            index_name=index_name,
            embedding=self.embeddings,
            pinecone_api_key=pinecone_key,
        )

        # Retriever
        self.retriever = self.vectorstore.as_retriever(
            search_type="similarity",
            search_kwargs={"k": 3}
        )

        # Text splitter (LangChain-native)
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=512,
            chunk_overlap=50,
            separators=["\n\n", "\n", ".", " ", ""]
        )

4.3.2. Хранение содержимого файла

    async def store_file_content_in_db(self, filepath: str) -> None:
        """
        Read the uploaded text file, split the file, embed chunks and save chunks to vector database
        :param filepath: filepath to load to vectordb
        :return: None
        """
        logger.debug(f"Loading file: {filepath}")

        # Load entire file (safe for most RAG use cases)
        async with aiofiles.open(filepath, "r", encoding="utf-8") as f:
            raw_text = await f.read()

        # Minimal cleaning (preserve structure)
        raw_text = raw_text.replace("\r", "")

        # Split into LangChain Documents
        docs = self.splitter.create_documents(
            [raw_text],
            metadatas=[{"source": os.path.basename(filepath)}]
        )

        logger.debug(f"Storing {len(docs)} chunks into Pinecone")

        # Ingest into Pinecone
        PineconeVectorStore.from_documents(
            documents=docs,
            embedding=self.embeddings,
            index_name=self.index_name,
            pinecone_api_key=pinecone_key,
        )

4.3.3. Поиск документов

    async def search_documents(self, query: str) -> list[Document]:
        logger.debug(f"Searching for: {query}")
        return self.retriever.invoke(query)

5. Бэкенд FastAPI

5.1. Загрузчик файлов

async def save_file(file: UploadFile) -> str:
    """
    accept a UploadFile from FastAPI and save the file in specified location
    :param file: UploadFile specified by users from fastAPI interface
    :return: path of the saved file
    """
    await makedirs("uploads", exist_ok=True)
    filepath = os.path.join("uploads", file.filename)
    async with aiofiles.open(filepath, "wb") as f:
        while chunk := await file.read(DEFAULT_CHUNK_SIZE):
            await f.write(chunk)
    return filepath

5.2. Модели запросов/ответов

class RAGRequest(BaseModel):
    prompt: str

class RAGResponse(BaseModel):
    answer: str
    num_original_documents: int

5.3. Конечная точка генерации ответов

# dependencies.py
from fastapi import Body, HTTPException
from schemas import RAGRequest
from nodes import generate

async def get_generation(body: RAGRequest=Body(...)) -> dict:
    try:
        generation = await generate(body.prompt)

        return {
            "answer": generation.get("generation", ""),
            "num_original_documents": generation.get("num_original_documents", 0)
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# main.py
@app.post("/generate_text", response_model=RAGResponse)
async def query_by_RAG_controller(generation: dict = Depends(get_generation)) -> RAGResponse:
    return RAGResponse(**generation)

5.4. Фоновая обработка

@app.post("/upload")
async def file_upload_controller(
        file: Annotated[UploadFile, File(description="Uploaded PDF documents")],
        bg_text_processor: BackgroundTasks,
):
    if file.content_type != "application/pdf":
        raise HTTPException(
            detail=f"Only uploading PDF documents are supported",
            status_code=status.HTTP_400_BAD_REQUEST,
        )
    try:
        filepath = await save_file(file)
        bg_text_processor.add_task(pdf_text_extractor, filepath)
        bg_text_processor.add_task(
            vector_service.store_file_content_in_db,
            filepath.replace("pdf", "txt")
        )
    except Exception as e:
        raise HTTPException(
            detail=f"An error occurred while saving file - Error: {e}",
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        )
    return {"filename": file.filename, "message": "File uploaded successfully"}

6. Фронтенд Streamlit

6.1. Загрузка файлов

st.write("Upload a file to FastAPI")
file = st.file_uploader("Choose a file", type=["pdf"])

if st.button("Submit"):
    if file is not None:
        files = {"file": (file.name, file, file.type)}
        response = requests.post("http://localhost:8000/upload", files=files)
        st.write(response.text)
    else:
        st.write("No file uploaded.")

6.2. История разговора

if "messages" not in st.session_state:
    st.session_state.messages = []

for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

6.3. Процесс чата

if prompt := st.chat_input("Write your prompt in this input field"):
    st.session_state.messages.append({"role": "user", "content": prompt})

    with st.chat_message("user"):
        st.text(prompt)

    question = prompt

    response = requests.post(
        f"http://localhost:8000/generate_text",
        json={"prompt": question}
    )
    response.raise_for_status()

    response_json = response.json()
    answer = response_json["answer"]
    st.session_state.messages.append({"role": "assistant", "content": answer})

    with st.chat_message("assistant"):
        st.markdown(answer)

7. Составление краткого резюме

Чек-лист операций