Ragent: یک زمزمه PDF چند عامل ساخته شده بر روی Langchain + Langgraph

پس از نصب موفقیت آمیز ، ما تمام بسته های لازم را برای ایجاد اولین عامل retriever وارد خواهیم کرد.

کدگذاری retrieveragent:

from langchain_openai import ChatOpenAI
from langchain_community.vectorstores.faiss import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain.docstore.document import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from pypdf import PdfReader
import re
from dotenv import load_dotenv
import streamlit as st
load_dotenv()
LLM = ChatOpenAI(model_name="gpt-4o", temperature=0.0)
def extract_text_from_pdf(pdf_path):
try:
pdf = PdfReader(pdf_path)
output = []
for i, page in enumerate(pdf.pages, 1):
text = page.extract_text()
text = re.sub(r"(\w+)-\n(\w+)", r"\1\2", text)
text = re.sub(r"(?, " ", text.strip())
text = re.sub(r"\n\s*\n", "\n\n", text)
output.append((text, i)) # Tuple of (text, page number)
return output
except Exception as e:
st.error(f"Error reading PDF: {e}")
return []
def text_to_docs(text_with_pages):
docs = []
text_splitter = RecursiveCharacterTextSplitter(chunk_size=4000, chunk_overlap=200)
for text, page_num in text_with_pages:
chunks = text_splitter.split_text(text)
for i, chunk in enumerate(chunks):
doc = Document(
page_content=chunk,
metadata={"source": f"page-{page_num}", "page_num": page_num}
)
docs.append(doc)
return docs
def create_vectordb(pdf_path):
text_with_pages = extract_text_from_pdf(pdf_path)
if not text_with_pages:
raise ValueError("No text extracted from PDF.")
docs = text_to_docs(text_with_pages)
embeddings = OpenAIEmbeddings()
return FAISS.from_documents(docs, embeddings)
# Define Tools
def retrieve_from_pdf(query: str, vectordb) -> dict:
"""Retrieve the most relevant text and page number using similarity search."""
# Use similarity_search to get the top result
docs = vectordb.similarity_search(query, k=3) # k=1 for single most relevant result
if docs:
doc = docs[0]
content = f"Page {doc.metadata['page_num']}: {doc.page_content}"
page_num = doc.metadata["page_num"]
return {"content": content, "page_num": page_num}
return {"content": "No content retrieved.", "page_num": None}
RETRIEVE_PROMPT = ChatPromptTemplate.from_messages([
("system", """
You are the Retrieve Agent. Your task is to fetch the most relevant text from a PDF based on the user's query.
- Use the provided retrieval function to get content and a single page number.
- Return the content directly with the page number included (e.g., 'Page X: text').
- If no content is found, return "No content retrieved."
"""),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{query}"),
])

توضیح کد –

در این کد عامل Retriever ، اولا ، ما تمام ماژول ها و کلاس های لازم را وارد می کنیم ، ما اعتبار خود را مانند OpenAi API Key در یک ذخیره می کنیم. .env پرونده ، به همین دلیل از ماژول dotenv در کنار تماس عملکرد Load_dotenv استفاده شده است. در مرحله بعد ، ما با ارائه آرگومان های مورد نیاز مانند نام مدل ، دما و غیره ، LLM را آغاز می کنیم.

توضیحات توابع

Extract_Text_from_pdf برای خواندن و استخراج محتوای PDF و تمیز کردن آن کمی با رفع خطوط خط هیپنوته شده ، که باعث می شود یک کلمه به دو قطعه تبدیل شود ، تبدیل یک خط جدید به فضاها می شود ، مگر اینکه آنها بخشی از فاصله پاراگراف باشند ، و غیره. فرایند تمیز کردن صفحه عاقلانه انجام می شود ، به همین دلیل یک حلقه با استفاده از عملکرد enumerate استفاده می شود. سرانجام ، از این عملکرد ، محتوای استخراج شده تمیز شده در کنار صفحه Pagenumber به عنوان نوعی لیست از Tuples بازگردانده می شود. در صورت بروز هرگونه خطای ناخواسته ، می توان از طریق آن نیز اداره شد از پیش امتحان کردن بلوک استفاده شده ؛ این تضمین می کند که کد بدون شکستن به دلیل خطاها ، یکپارچه کار می کند.

text_to_docs برای انجام استفاده می شود چاک دهی، در اینجا RevursiveCharacterTextSplitter کلاس ماژول Langchain مورد استفاده قرار می گیرد ، هر اندازه اندازه از 4000 خواهد بود و همپوشانی 200 خواهد بود. سپس یک حلقه بر روی آرگومان text_with_pages انجام می شود ، که خروجی را از عملکرد قبلی دریافت می کند ، یعنی Extract_Text_from_pdf ، همانطور که خروجی را در لیستی از قالب Tuples برمی گرداند. از دو متغیر در حلقه استفاده می شود تا هر دو مورد از Tuple را در نظر بگیریم. سپس متن پاک شده به تکه ها تقسیم می شود و به یک شیء سند تبدیل می شود که بیشتر برای تبدیل آنها به تعبیه ها استفاده می شود. جدا از محتوای صفحه ، شیء سند شماره صفحه و یک برچسب رشته را شامل شماره صفحه به عنوان ابرداده نگه می دارد. سپس هر سند به یک لیست اضافه می شود و باز می گردد.

Create_vectordb این تابع از دو عملکرد فوق برای ایجاد تعبیه با استفاده از FAISS (Facebook AI Searchity Searchity) استفاده می کند. این یک فروشگاه وکتور سبک است که شاخص را به صورت محلی ذخیره می کند و به انجام جستجوهای شباهت با سهولت کمک می کند. این عملکرد فقط بانک اطلاعاتی بردار را ایجاد و باز می گرداند. این است

retrieve_from_pdf در این عملکرد ، ما در حال انجام جستجوی شباهت و گرفتن 3 قطعه برتر هستیم و در صورت یافتن ، ما فقط اولین تکه را در نظر می گیریم به طوری که از همان محتوای مشابه تشکیل شده و آن را به همراه شماره صفحه خود به عنوان یک فرهنگ لغت بازگرداند.

بازیابی_prompt یک ChatpromptTemplate متشکل از دستورالعمل ، یعنی پیام سیستم برای LLM است ، کار به عنوان یک عامل رتریور. همچنین کل تاریخ چت یک جلسه خاص را در نظر می گیرد و پرس و جو کاربر را به عنوان ورودی انسانی می پذیرد.

برنامه نویسی عامل تقویت کننده

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from typing import Optional
def augment_with_context(content: str, page_num: Optional[int]) -> str:
"""Augment retrieved content with source context."""
if content != "No content retrieved." and page_num:
return f"{content}\n\nAdditional context: Sourced from page {page_num}."
return f"{content}\n\nAdditional context: No specific page identified."
AUGMENT_PROMPT = ChatPromptTemplate.from_messages([
("system", """
You are the Augment Agent. Enhance the retrieved content with additional context.
- If content is available, append a note with the single page number.
- If no content is retrieved, return "No augmented content."
"""),
MessagesPlaceholder(variable_name="chat_history"),
("human", "Retrieved content: {retrieved_content}\nPage number: {page_num}"),
])

توضیح توابع

augment_with_context این یک رویکرد بسیار ساده است که در آن ما به دنبال اطلاعات اضافی از PDF ارائه شده برای تحکیم اطلاعات بازیابی شده توسط عامل بازیابی هستیم. در صورت یافتن ، محتوای اضافی ، در کنار شماره صفحه خود ، به محتوای اصلی بازیابی شده اضافه می شود. در غیر این صورت ، اگر هر دو یافت نشوند ، به سادگی همان محتوای اصلی را بدون هیچ گونه اصلاح بازگرداند

ugment_prompt مجدداً بسیار ساده است ، فقط اطلاعاتی در مورد LLM است که به دنبال اطلاعاتی باشد که محتوای آن توسط عامل بازیابی را تقویت کند ، که به عنوان chat_history نیز در نظر گرفته می شود ، و متغیرهای بازیابی شده_content و page_num به طور خودکار در طول زمان اجرا جمع می شوند.

کدگذاری ژنراتور

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
GENERATE_PROMPT = ChatPromptTemplate.from_messages([
("system", """
You are the Generate Agent. Create a detailed response based on the augmented content.
- Focus on DBMS and SQL content.
- Append "Source: Page X" at the end if a page number is available.
- If the user query consists of terms like "explain", "simple", "simplify" etc. or relatable, then do not return any page number, otherwise return the proper page number.
- If the question is not DBMS-related, reply "Not applicable."
- Use the chat history to maintain context.
"""),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{query}\nAugmented content: {augmented_content}"),
])

نماینده ژنراتور فقط از دستورالعمل نحوه تولید پاسخ نهایی بر اساس محتوای بازیابی شده و همچنین اطلاعات اضافی افزوده از دو مرحله قبلی تشکیل شده است.

پس از ایجاد همه این عوامل جداگانه ، وقت آن است که آنها را در زیر یک چتر واحد ذخیره کرده و کل گردش کار پایان به پایان را با استفاده از Langgraph تشکیل دهید.

کد برای ایجاد نمودار با استفاده از Langgraph

import streamlit as st
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Optional
import re
from IPython.display import display, Image
from retriever import (LLM,extract_text_from_pdf,text_to_docs,create_vectordb,retrieve_from_pdf,RETRIEVE_PROMPT)
from augmentation import augment_with_context,AUGMENT_PROMPT
from generation import GENERATE_PROMPT
from dotenv import load_dotenv
load_dotenv()
PDF_FILE_PATH = "dbms_notes.pdf"
# Define the Agent State
class AgentState(TypedDict):
query: str
chat_history: List[dict]
retrieved_content: Optional[str]
page_num: Optional[int] # Single page number instead of a list
augmented_content: Optional[str]
response: Optional[str]
def format_for_display(text):
def replace_latex(match):
latex_expr = match.group(1)
return f"$${latex_expr}$$" # Use $$ for Streamlit Markdown to render LaTeX
text = re.sub(r'\\frac\{([^}]+)\}\{([^}]+)\}', r'$\\frac{\1}{\2}$', text)
return text
# Define Multi-Agent Nodes
def retrieve_agent(state: AgentState) -> AgentState:
chain = RETRIEVE_PROMPT | LLM
retrieved = retrieve_from_pdf(state["query"], st.session_state.vectordb)
response = chain.invoke({"query": state["query"], "chat_history": state["chat_history"]})
#print(retrieved)
return {
"retrieved_content": retrieved['content'],
"page_num": retrieved["page_num"]
}
def augment_agent(state: AgentState) -> AgentState:
chain = AUGMENT_PROMPT | LLM
if state["retrieved_content"] and state["retrieved_content"] != "No content retrieved.":
# Prepare input for the LLM
input_data = {
"retrieved_content": state["retrieved_content"],
"page_num": str(state["page_num"]) if state["page_num"] else "None",
"chat_history": state["chat_history"]
}
# Invoke the LLM to generate augmented content
response = chain.invoke(input_data)
augmented_content = response.content # Use the LLM's output
else:
augmented_content = "No augmented content."
return {"augmented_content": augmented_content}
def generate_agent(state: AgentState) -> AgentState:
chain = GENERATE_PROMPT | LLM
response = chain.invoke({
"query": state["query"],
"augmented_content": state["augmented_content"] or "No augmented content.",
"chat_history": state["chat_history"]
})
return {"response": response.content}
# Define Conditional Edge Logic
def decide_augmentation(state: AgentState) -> str:
if state["retrieved_content"] and state["retrieved_content"] != "No content retrieved.":
return "augmentation"
return "generation"
workflow = StateGraph(AgentState)
workflow.add_node("retrieve_agent", retrieve_agent)
workflow.add_node("augment_agent", augment_agent)
workflow.add_node("generate_agent", generate_agent)
workflow.set_entry_point("retrieve_agent")
workflow.add_conditional_edges(
"retrieve_agent",
decide_augmentation,
{
"augmentation": "augment_agent",
"generation": "generate_agent"
}
)
workflow.add_edge("augment_agent", "generate_agent")
workflow.add_edge("generate_agent", END)
agent = workflow.compile()
# display(Image(agent.get_graph().draw_mermaid_png(output_file_path="tutor_agent.png")))
st.set_page_config(page_title="🤖 RAGent", layout="wide")
st.title("🤖 RAGent : Your Personal Teaching Assistant")
st.markdown("Ask any question from your book and get detailed answers with a single source page!")
# Initialize session state for vector database
if "vectordb" not in st.session_state:
with st.spinner("Loading PDF content... This may take a minute."):
try:
st.session_state.vectordb = create_vectordb(PDF_FILE_PATH)
except Exception as e:
st.error(f"Failed to load PDF: {e}")
st.stop()
# Initialize chat history in session state
if "messages" not in st.session_state:
st.session_state.messages = []
# Display chat history
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
# User input
user_input = st.chat_input("Ask anything from the PDF")
if user_input:
# Add user message to chat history
st.session_state.messages.append({"role": "user", "content": user_input})
with st.chat_message("user"):
st.markdown(user_input)
# Display assistant response
with st.chat_message("assistant"):
message_placeholder = st.empty()
# Prepare chat history for the agent
chat_history = [
{"type": "human", "content": msg["content"]} if msg["role"] == "user" else
{"type": "ai", "content": msg["content"]}
for msg in st.session_state.messages[:-1] # Exclude current input
]
# Prepare initial state
initial_state = {
"query": user_input,
"chat_history": chat_history,
"retrieved_content": None,
"page_num": None,
"augmented_content": None,
"response": None, # Add field for Ragas sample
}
# Run the agent with a spinner
with st.spinner("Processing..."):
final_state = agent.invoke(initial_state)
answer = final_state["response"]
formatted_answer = format_for_display(answer)
# Display response
message_placeholder.markdown(formatted_answer)
# Update chat history
st.session_state.messages.append({
"role": "assistant",
"content": formatted_answer
})

توضیح کد

کلاس AgentState – در این کلاس ، ما در حال تعریف یک طرحواره هستیم که در بالای پاسخ LLM اجرا شود و کل “حالت” این ساختار مشابه را در کل گردش کار حمل می کند. این به عنوان استدلال در هنگام ایجاد دولت گراف تصویب می شود.

FORMAT_FOR_DISPLAY – این عملکرد دارای یک عملکرد تو در تو است که برای کنترل خروجی های مبتنی بر لاتکس استفاده می شود. ما از این استفاده می کنیم زیرا این سند ممکن است حاوی کسری باشد که ممکن است توسط جریان صحیح انجام نشود ، بنابراین استفاده از این به عنوان احتیاط اضافی.

عملکرد treive_agent – با این کار از عملکرد RETRIEVE_FROM_PDF که قبلاً تعریف کردیم استفاده می شود. در مرحله اول ، ما با استفاده از بازیابی سریع و LLM یک زنجیره ایجاد خواهیم کرد. سپس با استفاده از پرس و جو ارائه شده توسط کاربر ، آن را فراخوانی کنید ، که چیزی جز سوال کاربر نیست ، و همچنین کل chat_history را نیز در نظر بگیرید و در آخر آن محتوا و شماره صفحه را برگردانید.

عملکرد augment_agent – در اینجا ، ما دوباره با استفاده از Agment_Prompt ، زنجیره ای ایجاد خواهیم کرد و بررسی خواهیم کرد که آیا عامل Retriever هر محتوا را برگردانده است یا خیر. اگر هر محتوا را برگرداند ، ما با تابع AGEMENT_WOTH_CONTEXT تماس خواهیم گرفت و محتوای بازیابی شده ، شماره صفحه و همچنین Chat_history را منتقل می کنیم ، سپس محتوای ارائه شده توسط پاسخ را برگردانیم.

function function_agent – در اینجا ، سرانجام ، ما در حال عبور از محتوای افزوده ، پرس و جو کاربر و تاریخچه گپ هستیم تا LLM بتواند از محتوای افزوده استفاده کند و پاسخ نهایی را بر اساس اطلاعات افزوده ایجاد کند و آن را به کاربر نمایش دهد.

DETIC_AUGMENTATION – این یک مرحله اختیاری است که برای بررسی اینکه آیا عامل تقویت کننده لازم است یا خیر ، لازم است.

پس از ایجاد همه عوامل لازم ، زمان آن رسیده است که آنها را برای ایجاد یک گردش کار پایان به پایان ، که با استفاده از کلاس StateGraph Langgraph انجام می شود ، ترکیب کنیم. در حین اولیه سازی کلاس StateGraph ، ما کلاس AgentState را که قبلاً به عنوان پارامتر آن تعریف کردیم ، عبور خواهیم داد تا نشان دهد که در کل گردش کار ، این تنها کلیدهایی هستند که در پاسخ وجود خواهند داشت ، چیز دیگری نیست. سپس ما گره ها را به StateGraph اضافه می کنیم تا کل گردش کار را ایجاد کنیم ، نقطه ورود را به صورت دستی تنظیم کنیم تا درک کند که کدام گره ابتدا اجرا می شود ، لبه هایی را بین گره ها اضافه می کند تا بیان شود که چگونه گردش کار مانند به نظر می رسد ، اضافه کردن لبه مشروط در بین این نشان می دهد که گره متصل به لبه شرطی ، ممکن است یا ممکن است در طول جریان کار فراخوانده نشود.

سرانجام ، تدوین کل گردش کار برای بررسی اینکه آیا همه چیز خوب کار می کند و نمودار ایجاد شده مناسب است یا خیر. ما می توانیم نمودار را با استفاده از ماژول IPython و روش جوهر پری دریایی نمایش دهیم. اگر همه چیز به درستی پیش برود ، نمودار زیر به نظر می رسد.

سپس ، بقیه کد کاملاً مبتنی بر جریان است. کاربر می تواند UI را با توجه به انتخاب خود طراحی کند. ما در طراحی UI یک رویکرد بسیار اساسی اتخاذ کرده ایم ، به طوری که کاربر پسند باقی بماند. ما همچنین برخی از حالت های جلسه را در نظر می گیریم ، بنابراین برای حفظ تاریخچه گپ ، پرس و جو کاربر و غیره. این بدون ورودی کاربر شروع نمی شود ، به این معنی که تا زمانی که کاربر پرس و جو ارائه دهد ، گردش کار شروع نمی شود.

تصاویر برنامه در شرایط کار –

این مقاله با همکاری نوشته شده است بیسواجیت داس

منتشر شده از طریق به سمت هوش مصنوعی