Практычныя прытамулкі: Прыем дадзей для RAG у працоўнай суперактуальнасці: стварэнне надзеямогучага
Практычныя прыказкі: Абробка дадзейна для стварэння RAG у працоўным режыме: стварэнне надзяйных рашэнняў — кантракты, перакрыцчы і слоты для коду для команд, якія выкарыстоўваюць гэты патэрн.
У гэтым керавану практычна перакладзець шлях ад сыр'ёў да рабочай системы для: Прыема дадзэнняў для вырабаткавага RAG: стварэнне надзеяных корпоратыўскіх каналаводаў дадзэнняў. Акцэнт ставіцца на крокі, якія можна выконваць, чыстае перакананне і код, які можна падставіць у репазітарый без неабязковасці з'ясоўваць мету. У стадзіі агляду неабходна праказаць вхідныя даны, адпаведальнага за крок і критэрыя завершэння прычымоўкі перад зменай коду. Аператары должны магчымае перадзваніць крок з вядомай точкі контролю без неабязковасці з'ясоўваць схованы стан. Спрыйміце гэтую стадзію як кантракт межа вхіднымі данымі і падтвердзенымі выходнымі рэзультатамі. Праказаць назвы артыфактаў, праказаць критэрыяі успеху і адмовіцца ад тыхнай частковай завершэнні без паведамлення.
Што на самай працэ практычна ёсь це — прыем дадзэнняў
Калі працюеце над стадзіяй «Што на самай працэ ўваймае данні», спачатку запісуйце умовы кантракту: неабяжныя данні, сигнал працэйскага успеху і тое, што выканаецца у разе частковага нявыполнення. Такі список контроля дапамагае заліцьваты змяны ў кодзе пасля таго, як яны былі здабыты.
Проблема формата
Калі працюеце над стадзіяй «Проблема формата», спачатку запісайце угоду: неабяжлівыя даны, сігнал успеху і тое, што выходзіць у разе частковага нявыпання. Такі список пераконвае ў тым, што пазнейшыя змены коду будуць чыстымі. Зберагаеце настройкі паза кодам прыемліка. Файлы сераўнавання, хранальнікі секрэтных данных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куды аператары можаць адбавіць аудыт без неабяжлівага чытання всей структуры. Перад налаштаваннем запитоў пераканайцеся ў рэверыі на фіксаванай сэтке запытаў. Частае змена запытаў рэдка калі вялікі эфект дае на слабую систему пошуку.
Документныя загрузчыкі: па форматах
Калі працуеце з форматам Document Loaders паэтапна, спачатку запісайце умовы кантракту: неабяжлівыя даны, сігнал успеху і тое, што выканаецца у разе частковага нявыполнення. Такі список пераканаецца зберагчы чыстасць пазнейшых змян у кодзе. Документавайце як шлях успеху, так і шлях вярнення да стану нормальна. Перапрыбуткі, людзкія контралі і обработка некоректных паведамленняў ёсцю частью продукту, а не пазнейшым дапрацоўкам. Змяржыце рэт кантрактування на фіксаванай сэтке запытаў прычым падбіру запытанняў. Частае змена запытанняў рэдка калі вярнайце слабкую способнасць да выкарыстоўвання інформацыі.
Документы PDF
Калі працуеце з этапам дакументаў у формате PDF, спачатку запісайце контракт: неабходныя даны, сігнал успеху і тое, што выходзіць, калі адбываецца частковая нявыплата. Такі список пераканаець у тым, што пазнейшыя змены коду будуць чыстымі. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выйшае, нявыплата должна вказваць на адну адпаведальнасць, а не на заплутаны ланцюг задач. Перад налаштаваннем запитоў пераканайцеся ў рэкале на фіксаваным наборы пытанняў. Частае змена запитоў рэдка калі-небудзь выправляе слабую систему пошуку.
import fitz # PyMuPDF
from pathlib import Path
def classify_pdf(pdf_path: str) -> str:
"""
Classify a PDF as text-native, scanned, or hybrid.
Returns: 'text', 'scanned', or 'hybrid'
"""
doc = fitz.open(pdf_path)
total_pages = len(doc)
pages_with_text = 0
total_text_length = 0
for page in doc:
text = page.get_text("text").strip()
if text:
pages_with_text += 1
total_text_length += len(text)
doc.close()
# No text layer at all: scanned PDF
if pages_with_text == 0:
return "scanned"
# Text exists but average is suspiciously short: likely hybrid with poor OCR
avg_text_per_page = total_text_length / total_pages
if avg_text_per_page < 100 and pages_with_text < total_pages * 0.5:
return "hybrid"
return "text"
def load_text_pdf(pdf_path: str) -> list[dict]:
"""
Extract text from a text-native PDF using PyMuPDF.
Returns a list of page dicts with text and metadata.
"""
doc = fitz.open(pdf_path)
pages = []
for page_num, page in enumerate(doc, start=1):
text = page.get_text("text")
pages.append({
"page_number": page_num,
"text": text.strip(),
"source": pdf_path,
"total_pages": len(doc),
})
doc.close()
return pages
from azure.ai.formrecognizer import DocumentAnalysisClient
from azure.core.credentials import AzureKeyCredential
def load_scanned_pdf_azure(
pdf_path: str,
endpoint: str,
api_key: str,
) -> list[dict]:
"""
Extract text from a scanned PDF using Azure Document Intelligence.
Returns pages with text, confidence scores, and table data.
"""
client = DocumentAnalysisClient(
endpoint=endpoint,
credential=AzureKeyCredential(api_key),
)
with open(pdf_path, "rb") as f:
poller = client.begin_analyze_document("prebuilt-read", f)
result = poller.result()
pages = []
for page in result.pages:
page_text = ""
low_confidence_words = []
for line in page.lines:
page_text += line.content + "\n"
# Flag words with confidence below threshold
for word in page.words:
if word.confidence < 0.85:
low_confidence_words.append({
"word": word.content,
"confidence": word.confidence,
})
pages.append({
"page_number": page.page_number,
"text": page_text.strip(),
"low_confidence_words": low_confidence_words,
"needs_review": len(low_confidence_words) > 5,
"source": pdf_path,
})
return pages
DOCX: Внутрашняя дакументацыя політык
Калі працюеце над стадзіяй DOCX Internal Policy Documents, спачатку запісайце угоду: неабяжлівыя даны, сігнал успеху і тое, што выходзіць у разе частковага нявыпання. Такі список пераконтроўвае чыстасць пазнейшых змян у кодзе. Спрыятлівае ставленне да гэтай стадзіі як да угоды межа данымі і перакананымі выходамі. Дайце назву рэзультатам, задаце критэрыя успеху і адмовіцеся ад тыхоўага частковага завершэння. Замерьце рівень запам’ятовання на фіксаванай сэтцы пытанняў прычым перад налаштаваннем запитоў. Частае зміненне запитоў рэдка калі вярна выправляе слабыя аспекты адзысквання інформаціі.
from docx import Document as DocxDocument
from docx.oxml.ns import qn
def load_docx(docx_path: str) -> list[dict]:
"""
Load a DOCX file preserving heading structure and tables.
Returns a list of content blocks with type and text.
"""
doc = DocxDocument(docx_path)
blocks = []
current_section = "Introduction"
for element in doc.element.body:
# Paragraphs
if element.tag == qn("w:p"):
para = element
style = para.style.name if hasattr(para, "style") else ""
text = para.text.strip()
if not text:
continue
if style.startswith("Heading"):
level = int(style.replace("Heading ", "")) if style != "Heading" else 1
current_section = text
blocks.append({
"type": "heading",
"level": level,
"text": text,
"section": current_section,
"source": docx_path,
})
else:
blocks.append({
"type": "paragraph",
"text": text,
"section": current_section,
"source": docx_path,
})
# Tables
elif element.tag == qn("w:tbl"):
table_data = []
for row in element.findall(f".//{qn('w:tr')}"):
row_data = []
for cell in row.findall(f".//{qn('w:tc')}"):
cell_text = " ".join(
p.text for p in cell.findall(f".//{qn('w:t')}")
if p.text
)
row_data.append(cell_text.strip())
table_data.append(row_data)
blocks.append({
"type": "table",
"data": table_data,
"section": current_section,
"source": docx_path,
})
return blocks
HTML: Порталы са адпаведнасцю і внутрашнія Вікі
Калі працуеце з порталамі і стадіямі санкцыяў HTML, спачатку запісайце умовы контракту: неабходныя даны, сігнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список контроля дапамагае заліцварыць будучыя змены коду. Запісуйце час выканання і вартасьць токенаў або запытак палягліва да рэзультатаў функцыянальнасці. Візуабельнае паказанне вартасцей з самага пачатку запобегае неспакойным рахункам, калі праця пераходзіць з дэмаверсіі ў спакульнаныя сераўысы. Замерайце рэткі адзыв на фіксаваныя наборы запытанняў прычымо да налаштавання запрасаў. Частыя змены запрасаў рэдка калі вядуць да павышэння якосці адзыва.
import httpx
from bs4 import BeautifulSoup
def load_html_page(url: str, headers: dict | None = None) -> dict:
"""
Load an HTML page and extract main content, stripping boilerplate.
"""
response = httpx.get(url, headers=headers or {}, follow_redirects=True)
response.raise_for_status()
soup = BeautifulSoup(response.text, "html.parser")
# Remove boilerplate elements common in compliance portals
for tag in soup(["nav", "header", "footer", "script", "style",
"aside", "advertisement", ".cookie-banner",
".sidebar", ".breadcrumb"]):
tag.decompose()
# Find main content area (adjust selectors for your portal)
main_content = (
soup.find("main")
or soup.find("article")
or soup.find(id="content")
or soup.find(class_="content")
or soup.find("body")
)
text = main_content.get_text(separator="\n", strip=True) if main_content else ""
return {
"url": url,
"title": soup.title.string if soup.title else "",
"text": text,
"source": url,
}
JSON і реляцыйныя базы дадзеных
Калі працуеце з стадзіям JSON і реляцыйных баз дадзэнняў, спачатку запісайце умовы варакту: неабходныя данні, сігнал успеху і тое, што выходзіць на частым неудачам. Такі список контроля дапамагае заліцьваваць змяны ў кодзе пазнейша. Зберагайце настройкі за межамі коду аплікацыі. Файлы сераўнавання, хранільнікі секрэтных дадзэнняў і флагі функцыйяй должны знаходзіцца ў аднам месцы, куды аператары можаюць адбавіць аудыт без неабяжнага чытання всіх элементаў. Перад налагоджэньем запитоў пераканайцеся ў рівень вярнага аднаходжэння дадзэнняў на фіксаваным наборе запитаў. Частая змена запітоў рэдка калі вярнайце слабкую систему аднаходжэння дадзэнняў.
import json
import psycopg2
from typing import Any
def load_json_policies(json_path: str, text_fields: list[str]) -> list[dict]:
"""
Load a JSON array of policy documents, extracting specified text fields.
Banking example: a JSON export from a policy management system.
"""
with open(json_path, "r", encoding="utf-8") as f:
policies = json.load(f)
documents = []
for policy in policies:
# Combine relevant text fields into a single document
text_parts = []
for field in text_fields:
if field in policy and policy[field]:
text_parts.append(f"{field.replace('_', ' ').title()}: {policy[field]}")
documents.append({
"text": "\n\n".join(text_parts),
"policy_id": policy.get("id"),
"policy_type": policy.get("type"),
"effective_date": policy.get("effective_date"),
"jurisdiction": policy.get("jurisdiction"),
"source": json_path,
})
return documents
def load_from_postgres(
connection_string: str,
query: str,
text_column: str,
metadata_columns: list[str],
) -> list[dict]:
"""
Load documents from a PostgreSQL database.
Banking example: regulatory requirement records from a compliance database.
"""
conn = psycopg2.connect(connection_string)
cursor = conn.cursor()
cursor.execute(query)
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
cursor.close()
conn.close()
documents = []
for row in rows:
row_dict = dict(zip(columns, row))
documents.append({
"text": str(row_dict.get(text_column, "")),
**{col: row_dict.get(col) for col in metadata_columns},
"source": "postgresql",
})
return documents
SharePoint і Google Docs: Складзішчы дакументаў для падпрыемства
Калі працюеце над этапамі SharePoint і Google Docs, спачатку запісайце угоду: неабяжлівыя даны, сигнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список пераконтроўвае чыстасць пазнейшых змян у кодзе. Документавайце як шлях успеху, так і шлях вярнення да нормы. Перапрыбуткі, людзкія контралі і обработка некоректных паведамленняў ёсць частью продукту, а не пазнейшым дапрацоўкам. Змяркуйце рэткасць адпаведзенняў на фіксованы набор запитанняў прычымо да налаштавання підказак. Частае змяненне підказак рэдка калі выправляе слабкую систему адзысквання інформаціі. Калі працюеце над этапамі SharePoint і Google Docs, спачатку запісайце угоду: неабяжлівыя даны, сигнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список пераконтроўвае чыстасць пазнейшых змян у кодзе. Спрытывайце гэты этап як угоду межа данымі і перакананымі выходамі. Дайце назвы элементам, задаце критэрыя успеху і адмовіцеся ад тыхоўага частковага завершэння.
import httpx
from typing import Generator
def load_sharepoint_library(
tenant_id: str,
client_id: str,
client_secret: str,
site_id: str,
drive_id: str,
) -> Generator[dict, None, None]:
"""
Iterate over all DOCX and PDF files in a SharePoint document library.
Yields file metadata dicts; caller is responsible for downloading and parsing.
"""
# Get access token
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
token_response = httpx.post(token_url, data={
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret,
"scope": "https://graph.microsoft.com/.default",
})
access_token = token_response.json()["access_token"]
headers = {"Authorization": f"Bearer {access_token}"}
base_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives/{drive_id}"
# List files recursively
next_link = f"{base_url}/root/children"
while next_link:
response = httpx.get(next_link, headers=headers)
data = response.json()
for item in data.get("value", []):
if item.get("file") and item["name"].endswith((".pdf", ".docx")):
yield {
"name": item["name"],
"download_url": item.get("@microsoft.graph.downloadUrl"),
"created": item.get("createdDateTime"),
"modified": item.get("lastModifiedDateTime"),
"size": item.get("size"),
"web_url": item.get("webUrl"),
}
next_link = data.get("@odata.nextLink")
Стрімаванне проты пакетнага приймання
Этап стрімінгавання проты батчовага практыкавання дае найлепыя рэзультаты, калі яго спрыяваць як меркаваную плошчу. Запісайце адны ідеальны прыклад, адну ситуацыю неудачы і прыметкі па поверненню да пярвоначальнага стану, прычым расширюючы сферу дзеяння. Запісвайце часы выконання і косты токеноў або запытак праза функцыйнае рэзультат. Відразлівасць костаў з самага пачатку запобегае неспакойным рахункам, калі процес пераходзіць з дэмовай среды ў спяльнаваныя сераўеры. Раздзеліце правілы часткавага обробкі дадзеных ад правіл ўтрымання іх. Змена адных не павінна вымагаць перапісвання іншых, калі змянююцыся паказнікі якосці.
import time
import httpx
from pathlib import Path
class RateLimitedConnector:
"""
A base connector with retry logic and rate limiting.
Use this pattern for any external API that imposes rate limits.
Banking example: connecting to an RBI document portal or a Bloomberg data feed.
"""
def __init__(
self,
base_url: str,
api_key: str,
requests_per_minute: int = 60,
max_retries: int = 3,
backoff_factor: float = 2.0,
):
self.base_url = base_url
self.api_key = api_key
self.min_interval = 60.0 / requests_per_minute
self.last_request_time = 0.0
self.max_retries = max_retries
self.backoff_factor = backoff_factor
def _wait_for_rate_limit(self):
elapsed = time.monotonic() - self.last_request_time
if elapsed < self.min_interval:
time.sleep(self.min_interval - elapsed)
self.last_request_time = time.monotonic()
def fetch(self, endpoint: str, params: dict | None = None) -> dict:
url = f"{self.base_url}/{endpoint}"
headers = {"Authorization": f"Bearer {self.api_key}"}
for attempt in range(self.max_retries):
self._wait_for_rate_limit()
try:
response = httpx.get(url, headers=headers, params=params or {})
if response.status_code == 429:
# Rate limited: back off and retry
wait_time = self.backoff_factor ** attempt
time.sleep(wait_time)
continue
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
if attempt == self.max_retries - 1:
raise
time.sleep(self.backoff_factor ** attempt)
raise RuntimeError(f"Failed to fetch {url} after {self.max_retries} attempts")
Мультимодальнае практыкавання
Этап мультымодальнага прыема работае наяўней, калі яго спрыяваць як вимерную паверхню. Зафіксавайце адны ідеальны прыем дадзеных, адны прыклад неудачы і запіс пра вярнэнне да пачатковага стану, перш чым расширваць сферу дзеяння. Зберагаюце настройкі пазырочна ад коду прыемнай програмы. Файлы сяродавішняе сераўісу, хранілішчы секрэтных дадзеных і пазнакі функцый должны знаходзіцца ў адном месцы, куда аператары можуць адбавляць аудыт без неабяжнага чытання всіх элементаў системы. Раздзеляйце правілы часткавага обробкі дадзеных і правілы ўзяць іх. Змена адных не должна вымагаць перапісву іншых, калі зменяюцыся паказнікі якосці.
Аналіз з урахоўваннем табелей
Этап аналізу з урахоўваннем табелей працюе найэфективней, калі яго розглядаюць як параметры, якія можна вимерыць. Зберагуйце адну ідеальную версію транскрыпцыі, адзін прыклад неудачы і запіс пра відкатанне роботы перш чым расширваць сферу дзеяння. Документавайце як шлях успеху, так і шлях відновлення разам. Перапрыбуткі, людзкі контроль і обработка некоректных поведань ўскладнень є частью продукту, а не етапамі пазнейшай доработкі. Раздзеляйце правілы фрагментавання інфармацыі ад правіл яе запошуку. Змена аднаго з іх не должна прымусіваць перапісванне другага, калі змянююцыся показнікі ясноці. Этап аналізу з урахоўваннем табелей працюе найэфективней, калі яго розглядаюць як параметры, якія можна вимерыць. Зберагуйце адну ідеальную версію транскрыпцыі, адзін прыклад неудачы і запіс пра відкатанне роботы перш чым расширваць сферу дзеяння. Разглядайце этап як кантракт між вхіднымі даннымі і перакананымі выходнымі рэзультатамі. Даўце назвы элементам, задайце критэрыя успеху і не падтрымвайце беззвучнае частковае завершэння роботы.
import fitz
import pdfplumber
def extract_tables_from_pdf(pdf_path: str) -> list[dict]:
"""
Extract tables from a PDF using pdfplumber.
Returns tables as both structured data and serialised text.
Banking example: extracting capital ratio tables from Basel III compliance reports.
"""
tables_output = []
with pdfplumber.open(pdf_path) as pdf:
for page_num, page in enumerate(pdf.pages, start=1):
tables = page.extract_tables()
for table_idx, table in enumerate(tables):
if not table or len(table) < 2:
continue
# First row as headers
headers = [str(h).strip() if h else f"col_{i}"
for i, h in enumerate(table[0])]
rows = table[1:]
# Structured data representation
structured = []
for row in rows:
row_dict = {}
for header, cell in zip(headers, row):
row_dict[header] = str(cell).strip() if cell else ""
structured.append(row_dict)
# Text serialisation (markdown table format for LLM consumption)
header_row = " | ".join(headers)
separator = " | ".join(["---"] * len(headers))
data_rows = [
" | ".join(str(cell or "").strip() for cell in row)
for row in rows
]
serialised = "\n".join([header_row, separator] + data_rows)
tables_output.append({
"page_number": page_num,
"table_index": table_idx,
"headers": headers,
"structured_data": structured,
"serialised_text": serialised,
"source": pdf_path,
})
return tables_output
Працэс уваходжэння зображэнняў і графікав
У стадії абсорбцыяй зображэнняў і графіка неабходна прадзефінаваць вхідныя даны, адпаведальную особу за выкананне крока і критэрыя завершэння пры перамене коду. Аператары должны магчыма было перзапускаць крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Запісваць час выканання і вартасць токенаў або запытак праза функцыйнае рэзультат. Відкрытая інформацыя пра вартасці запобегае неспакоўным рахункам, калі процес пераходзіць з дэмовай среды ў спяльную. Указваць тые часткі тексту, якія фактычна ляглі в основу адпаведнай адказы. Без цых цітатаў аператары не можуць разлічыць галюцинацію ад прасоў у індэксаванні.
import fitz
import base64
import httpx
from io import BytesIO
def extract_and_describe_images(
pdf_path: str,
openai_api_key: str,
min_image_size: int = 5000, # bytes, to skip tiny decorative images
) -> list[dict]:
"""
Extract images from a PDF and generate text descriptions using GPT-4V.
Banking example: describing NPA trend charts and regulatory workflow diagrams.
"""
doc = fitz.open(pdf_path)
image_docs = []
for page_num in range(len(doc)):
page = doc[page_num]
image_list = page.get_images(full=True)
for img_idx, img_info in enumerate(image_list):
xref = img_info[0]
base_image = doc.extract_image(xref)
image_bytes = base_image["image"]
# Skip tiny decorative images
if len(image_bytes) < min_image_size:
continue
# Encode to base64 for the vision API
image_b64 = base64.b64encode(image_bytes).decode("utf-8")
image_ext = base_image["ext"]
# Generate description with GPT-4V
response = httpx.post(
"https://api.openai.com/v1/chat/completions",
headers={"Authorization": f"Bearer {openai_api_key}"},
json={
"model": "gpt-4o",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": (
"This image is from a banking regulatory document. "
"Describe its content precisely, including any numerical "
"values, trends, labels, axes, or process steps shown. "
"If it is a chart, describe what metric is shown, the "
"time period, and the key data points. If it is a "
"diagram, describe the process or relationship shown."
),
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/{image_ext};base64,{image_b64}"
},
},
],
}
],
"max_tokens": 400,
},
)
description = response.json()["choices"][0]["message"]["content"]
image_docs.append({
"type": "image_description",
"page_number": page_num + 1,
"image_index": img_idx,
"description": description,
"source": pdf_path,
})
doc.close()
return image_docs
Абсорбцыя аудыю і відэам
Для стадіі прыемкі аудыя і відэа неабходна пазначыць вхідныя даны, адпаведальнага за гэты крок і критэрыі завершэння пры змяне коду. Аперацыйныя працавнікі должны магчымаць перзапуск кроку з вядомай точкі контролю, не падозрываючы прыхованы стан. Конфігурацыю трэба залічыць пазначальна ўнутры коду прыемкі. Файлы сераўіса, хранільнікі секрэтных данных і флагі функцыйяў должны знаходзіцца ў аднам месцы, якое працавнікі можуць пераглядаць, не чытаючы весь ланцуг прыемкі. Неабходна цітаваць тыя часткі тексту, якія фактычна лежалі в основе адпаведнай адказы. Без цітатаў працавнікі не зможуць адразліць галюцинацыю ад прасоўкі ў індэксаванні.
import httpx
import time
def transcribe_audio(
audio_path: str,
openai_api_key: str,
language: str = "en",
) -> dict:
"""
Transcribe an audio file using OpenAI Whisper.
Banking example: transcribing an RBI monetary policy press conference recording.
"""
with open(audio_path, "rb") as audio_file:
response = httpx.post(
"https://api.openai.com/v1/audio/transcriptions",
headers={"Authorization": f"Bearer {openai_api_key}"},
data={
"model": "whisper-1",
"language": language,
"response_format": "verbose_json", # includes timestamps per segment
"timestamp_granularities[]": "segment",
},
files={"file": (audio_path, audio_file, "audio/mpeg")},
)
result = response.json()
return {
"transcript": result.get("text", ""),
"segments": [
{
"start": seg["start"],
"end": seg["end"],
"text": seg["text"],
}
for seg in result.get("segments", [])
],
"language": result.get("language", language),
"source": audio_path,
}
Аднаўленне ўсьго разам: канвейер прыемкі для практычнага выкарыстоўвання
У стадії «Парадокс А» неабяжна праймеры інпутаў, адпаведальнага за крок і крэтарыяў завершэння пры перадзмене коду. Аперацыйныя працавнікі павінны магчымае перзапускаць крок з вядомай точкі контролю, не спрабоўваючы здагадвацца пра схованы стан. Неабяжна задокументаваць як шлях успеху, так і шлях вярнення да нормы. Перапрыбуткі, людзкія контралі і обработка некоректных паведамленняў є часткай продукту, а не дадатковым доўнесенням пазнейша. Указваць часткі тексту, якія фактычна лежаць у падставе адпаведнай адказу. Без ціх цытатаў аперацыйныя працавнікі не можуць разлічыць галюцинацію ад прасоў у індэксаванні. У стадії «Парадокс А» неабяжна праймеры інпутаў, адпаведальнага за крок і крэтарыяў завершэння пры перадзмене коду. Аперацыйныя працавнікі павінны магчымае перзапускаць крок з вядомай точкі контролю, не спрабоўваючы здагадвацца пра схованы стан. Спрыяць гэтай стадіі як кантракту межа інпутамі і перакананымы выходнымі даннымі. Называць артыфакты, праймеры пераканання ў успеху і адмовляцца ад беззвучнага частковага завершэння.
from pathlib import Path
from enum import Enum
import logging
logger = logging.getLogger(__name__)
class DocumentType(Enum):
PDF_TEXT = "pdf_text"
PDF_SCANNED = "pdf_scanned"
DOCX = "docx"
HTML = "html"
JSON = "json"
AUDIO = "audio"
UNKNOWN = "unknown"
def detect_document_type(file_path: str) -> DocumentType:
path = Path(file_path)
suffix = path.suffix.lower()
if suffix == ".pdf":
classification = classify_pdf(file_path)
return DocumentType.PDF_SCANNED if classification == "scanned" else DocumentType.PDF_TEXT
elif suffix == ".docx":
return DocumentType.DOCX
elif suffix in (".html", ".htm"):
return DocumentType.HTML
elif suffix == ".json":
return DocumentType.JSON
elif suffix in (".mp3", ".mp4", ".wav", ".m4a"):
return DocumentType.AUDIO
return DocumentType.UNKNOWN
def ingest_document(
file_path: str,
azure_endpoint: str | None = None,
azure_api_key: str | None = None,
openai_api_key: str | None = None,
) -> list[dict]:
"""
Route a document to the correct parser and return a list of content blocks
with a consistent schema regardless of input format.
Every output block contains:
- text: the extracted text content
- source: origin path or URL
- doc_type: the detected document type
- page_number: where applicable
- needs_review: flag for human review queue
"""
doc_type = detect_document_type(file_path)
results = []
try:
if doc_type == DocumentType.PDF_TEXT:
pages = load_text_pdf(file_path)
for page in pages:
page["doc_type"] = "pdf_text"
page["needs_review"] = False
results.append(page)
elif doc_type == DocumentType.PDF_SCANNED:
if azure_endpoint and azure_api_key:
pages = load_scanned_pdf_azure(file_path, azure_endpoint, azure_api_key)
for page in pages:
page["doc_type"] = "pdf_scanned"
results.append(page)
else:
logger.warning(
f"Scanned PDF detected but no OCR credentials provided: {file_path}"
)
results.append({
"text": "",
"source": file_path,
"doc_type": "pdf_scanned",
"needs_review": True,
"error": "No OCR credentials available",
})
elif doc_type == DocumentType.DOCX:
blocks = load_docx(file_path)
for block in blocks:
block["doc_type"] = "docx"
block["needs_review"] = False
results.append(block)
elif doc_type == DocumentType.AUDIO:
if openai_api_key:
transcript = transcribe_audio(file_path, openai_api_key)
for segment in transcript["segments"]:
results.append({
"text": segment["text"],
"source": file_path,
"doc_type": "audio_transcript",
"start_seconds": segment["start"],
"end_seconds": segment["end"],
"needs_review": False,
})
else:
logger.warning(f"Unknown document type, skipping: {file_path}")
except Exception as e:
logger.error(f"Ingestion failed for {file_path}: {e}")
results.append({
"text": "",
"source": file_path,
"doc_type": str(doc_type.value),
"needs_review": True,
"error": str(e),
})
return results
Што парадоксуецца ў прыметнай експлуатацыі
Калі працюеце над тым, што парадоксуецца на стадіўцы, спачатку запішыце умовы: неабходныя данні, сігнал успеху і тое, што выходзіць на частым неудачах. Такі список контроля дапамагае заставаць пазнейшыя змены коду чыстымі. Запісвайце час выконання і кост токенаў або запытаў па боку рэзультатаў функцыяналізацыі. Відразувая візуабільнасць костаў запобегае неспакойным рахункам, калі працэс пераходзіць з дэмаверсіі ў спяльныя среды. Перад налаштаваннем запрошэнняў пераканайцеся ў рівень вярнага адналёгчэння на фіксаваны набор запытаў. Частае змінэнне запрошэнняў рэдка калі-небудзь вылечвае слабую систему адналёгчэння.
Компрасы ў прыметнай експлуатацыі
Калі працюеце над стадзіяй «Выборы ў працэкты вырабоцтва», спачатку запісайце контракт: неабходныя вхідныя даны, сигнал успеху і тое, што выканаецца у разы частковага нявучнасці. Такі список контроля дапамагае залічыць змяны ў кодзе пазнейша. Зберагайце настройкі за межамі коду прыемліка. Файлы сераўнавання, хранільнікі секрэтных дадзеных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куды аператары можаюць адбавіць аудыт без неабходнасці чытання всей структуры. Перад налаштаваннем запитоў пераканайцеся ў рэверсі на фіксаваным наборе запитанняў. Частае змена запитоў рэдка калі вярнуе слабую эфектыўнасць пошуку.
Што будзе далей
Калі працюеце над стадзіяй «Што будзе далей», спачатку запісайце угоду: неабяжлівыя данні, сигнал успеху і тое, што выходзіць у разе частковага неудачы. Такі список пераканальвае правдавасць пазнейшых змян у кодзе. Дакументавайце як шлях успеху, так і шлях вярнення. Перапрыбуткі, людзкіе контралі і обработка некоректных паведамленняў ёсць часткай продукту, а не пазнейшым дапрацоўкам. Змяржвайце рэткасць адпаведзенняў на фіксаваны набор запытанняў прычымо да налаштавання запытанняў. Частае змена запытанняў рэдка калі вярнайць слабкую эфектыўнасць адзысквання інформаціі. Калі працюеце над стадзіяй «Што будзе далей», спачатку запісайце угоду: неабяжлівыя данні, сигнал успеху і тое, што выходзіць у разе частковага неудачы. Такі список пераканальвае правдавасць пазнейшых змян у кодзе. Спрыятлівайце гэтай стадзіяй як угоды межа даннімі і перакананымі выходамі. Дайце назвы элементам, задаце критэрыя успеху і адмовіцеся ад тыхоўскага частковага завершэння.
Чысткі ліст для эксплуатацыі
Этап перагляду канцэлекту працюе найкраща, калі яго спрыявае можласць вимеры. Зберагчыце адна ідеальная версія, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану, перш чым расширваць масштабы.
Валідзіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выйшае, прычына неудачы павінна вказываць на адну конкрэтную адпаведальнасць, а не на заплутаны ланцюг задач.
Раздзеліце правілы часткавання інфармацыі ад правіл яе выявлення. Змена адных не павінна прымусваць перапісванне іншых, калі змянююцца паказателі якосці.
Калі дозволяе бюджет, дадзіце тэст, які пераглядае критычны шлях у процесе CI з викорыстаннем фіксатываючых элементоў, а не рэальных платных API.
Спрыяйце цэму этапу як кантракту межа вхіднымі дадзеннямі і перакананымі выходнымі рэзультатамі. Даўце назвы элементам, задаце критэрыя успеху і адмовіцеся ад тыхнага частковага завершэння без паведамлення.
Раздзеліце правілы часткавання інфармацыі ад правіл яе выявлення. Змена адных не павінна прымусваць перапісванне іншых, калі змянююцца паказателі якосці.
Перш чым запускать стак, заморозьце версіі, зафіксавайце «золаты» транскрыпты для критичных етапаў і паказвайце спосабы адвярнення змян. У спільных сераўнах неабходны ліміты частоты запытоў, перакананні ў правах на выкарыстоўванне ресурсаў і чыста вялікі адпаведальны за змяну секрэтных даных. Валіце простую надзейнасць працы над крэатывнымі, але разовымі дамах.
Прыметкі для 771b701b6010: не кладзіце ключы прадаўцаў у репазітарый, задаце верхнюю межу токеноў на кожную сесію і зберагачыце транскрыпты празаўседы ў фіксатыях для ацэнкі, каб пазнейшыя замены моделяў заставаліся пораўнанымі.