Агентскі ІІ з 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 Backend

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 польскі.

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. Складанне падсумку

Чакліст аператыўных задач