Агентський ШІ з 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. Symantic RAG працює найкраще, коли його розглядають як вимірювану поверхню. Зафіксуйте один ідеальний запис, один випадок невдачі та примітку про скасування змін перед розширенням обсягу роботи. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте мовчазного часткового виконання завдань. Розділіть політику розбиття на частини та політику пошуку інформації. Зміна однієї з них не повинна змушувати переписувати іншу при зміні показників якості.

2.3. Backend FastAPI

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

2.4. Клієнт Streamlit

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

3. Ланцюги LLM з LCEL

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

3.1. Мова виразів LangChain (LCEL)

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

r polish.

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. Створення резюме

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