В этом руководстве мы покажем, как Haystack позволяет создавать продвинутые системы искусственного интеллекта (ИИ), которые выходят далеко за рамки простых примеров, оставаясь при этом полностью работоспособными. Мы сосредоточимся на комплексной, сквозной настройке, которая подчёркивает организацию, принятие решений с учётом состояния, выполнение инструментов и структурированный поток управления. Это демонстрирует, как сложное поведение агентов может быть чётко выражено.
Установка и импорт библиотек
Мы устанавливаем и импортируем все необходимые библиотеки, обеспечивая доступность Haystack, OpenAI и инструментов обработки данных, а также безопасно загружаем ключ API OpenAI во время выполнения. Мы настраиваем среду так, чтобы она могла корректно обрабатывать отсутствующие зависимости и запрашивать учётные данные без жёсткого кодирования конфиденциальной информации.
«`python
import os, json, math, random, textwrap
from datetime import datetime, timedelta
try:
import pandas as pd
except Exception:
os.system(«pip -q install pandas»)
import pandas as pd
try:
import numpy as np
except Exception:
os.system(«pip -q install numpy»)
import numpy as np
try:
import duckdb
except Exception:
os.system(«pip -q install duckdb»)
import duckdb
os.system(«pip -q install haystack-ai openai»)
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.tools import tool
from haystack.components.agents.state import State
from haystack.components.agents.state.stateutils import mergelists
from haystack.tools import ComponentTool
from getpass import getpass
if not os.getenv(«OPENAIAPIKEY»):
key = getpass(«Enter OPENAIAPIKEY (input hidden): «).strip()
if key:
os.environ[«OPENAIAPIKEY»] = key
if not os.getenv(«OPENAIAPIKEY»):
raise RuntimeError(«OPENAIAPIKEY missing. Set it in the environment or paste when prompted.»)
«`
Генерация синтетических данных
Мы генерируем реалистичный 24-часовой поток синтетических показателей работы служб с периодическим поведением и шумом. Мы намеренно создаём окно инцидента, во время которого задержка и частота ошибок резко возрастают, а пропускная способность запросов снижается.
«`python
def synth_metrics(n=1440):
ts = [start + timedelta(minutes=i) for i in range(n)]
base_rps = 220 + 40np.sin(np.linspace(0, 8math.pi, n)) + np.random.normal(0, 10, n)
base_p95 = 180 + 30np.sin(np.linspace(0, 6math.pi, n) + 0.5) + np.random.normal(0, 8, n)
base_err = np.clip(np.random.normal(0.006, 0.002, n), 0.0, 0.05)
incident_t0 = int(n*0.62)
incidentt1 = incidentt0 + int(n*0.10)
basep95[incidentt0:incidentt1] += np.linspace(120, 520, incidentt1-incident_t0)
baseerr[incidentt0:incidentt1] += np.linspace(0.01, 0.07, incidentt1-incident_t0)
baserps[incidentt0:incidentt1] -= np.linspace(5, 80, incidentt1-incident_t0)
df = pd.DataFrame({
«ts»: ts,
«rps»: np.clip(base_rps, 5, None),
«p95ms»: np.clip(basep95, 10, None),
«errorrate»: np.clip(baseerr, 0.0, 0.2),
})
return df, (ts[incidentt0], ts[incidentt1])
metricsdf, (incidentbegin, incidentend) = synthmetrics()
«`
Инструменты для расследования
Мы определяем набор инструментов для расследования, которые позволяют агентам запрашивать данные, извлекать закономерности и предлагать конкретные меры по устранению неполадок. Мы организуем работу специалистов-профилировщиков и писателей под руководством координатора, который управляет сквозным, не-RAG рабочим процессом инцидента.
«`python
@tool
def sql_investigate(query: str) -> dict:
try:
df = con.execute(query).df()
head = df.head(30)
return {
«rows»: int(len(df)),
«columns»: list(df.columns),
«preview»: head.to_dict(orient=»records»)
}
except Exception as e:
return {«error»: str(e)}
@tool
def logpatternscan(windowstartiso: str, windowendiso: str, top_k: int = 8) -> dict:
ws = pd.todatetime(windowstart_iso)
we = pd.todatetime(windowend_iso)
df = logsdf[(logsdf[«ts»] >= ws) & (logs_df[«ts»] <= we)].copy()
if df.empty:
return {«rows»: 0, «toperrorkinds»: [], «topservices»: [], «topendpoints»: []}
df[«errorkindnorm»] = df[«error_kind»].fillna(«»).replace(«», «NONE»)
err = df[df[«level»].isin([«WARN»,»ERROR»])].copy()
toperr = err[«errorkindnorm»].valuecounts().head(int(topk)).todict()
topsvc = err[«service»].valuecounts().head(int(topk)).todict()
topep = err[«endpoint»].valuecounts().head(int(topk)).todict()
byregion = err.groupby(«region»).size().sortvalues(ascending=False).head(int(topk)).todict()
p95latency = float(np.percentile(df[«latencyms»].values, 95))
return {
«rows»: int(len(df)),
«warnerrorrows»: int(len(err)),
«p95latencyms»: p95_latency,
«toperrorkinds»: top_err,
«topservices»: topsvc,
«topendpoints»: topep,
«errorbyregion»: by_region
}
«`
Координация агентов
Мы настраиваем запросы, схемы состояний и мостики между инструментами, чтобы система создавала фальсифицируемые гипотезы, действенные планы и структурированный постмортем.
«`python
profiler_agent = Agent(
chat_generator=llm,
tools=[loadinputs, detectincidentwindow, sqlinvestigate, logpatternscan],
systemprompt=profilerprompt,
exit_conditions=[«text»],
stateschema=stateschema
)
writer_agent = Agent(
chat_generator=llm,
tools=[draft_postmortem],
systemprompt=writerprompt,
exit_conditions=[«text»],
stateschema=stateschema
)
coordinator_agent = Agent(
chat_generator=llm,
tools=[
load_inputs,
detectincidentwindow,
sql_investigate,
logpatternscan,
propose_mitigations,
profiler_tool,
writer_tool,
draft_postmortem
],
systemprompt=coordinatorprompt,
exit_conditions=[«text»],
stateschema=stateschema
)
«`
Мы показали, как Haystack поддерживает сложные агентские шаблоны, которые масштабируются по сложности, не становясь хрупкими или сложными для понимания. Мы продемонстрировали, что даже в блокноте можно выразить сложную логику агентов, поддерживать явное состояние и координировать несколько компонентов контролируемым и расширяемым способом.
1. Какие инструменты и библиотеки используются для создания мультиагентной системы на базе Haystack?
В статье используются следующие библиотеки и инструменты: Haystack, OpenAI, pandas, numpy, duckdb, а также инструменты для обработки данных и генерации синтетических данных.
2. Как в системе Haystack генерируются синтетические данные для тестирования системы обнаружения инцидентов?
В системе Haystack генерируются синтетические данные с помощью функции synth_metrics, которая создаёт реалистичный 24-часовой поток показателей работы служб с периодическим поведением и шумом. Для создания окна инцидента, во время которого задержка и частота ошибок резко возрастают, а пропускная способность запросов снижается, используются массивы numpy и pandas.
3. Какие инструменты для расследования определены в системе Haystack и как они помогают агентам в обнаружении инцидентов?
В системе Haystack определены следующие инструменты для расследования: sqlinvestigate (позволяет агентам запрашивать данные и извлекать закономерности) и logpattern_scan (позволяет извлекать закономерности из журналов и предлагать конкретные меры по устранению неполадок). Эти инструменты помогают агентам в обнаружении инцидентов, предоставляя им необходимые данные и инструменты для анализа.
4. Как в системе Haystack организована координация агентов для обнаружения и устранения инцидентов?
В системе Haystack координация агентов организована с помощью настройки запросов, схем состояний и мостиков между инструментами. Это позволяет системе создавать фальсифицируемые гипотезы, действенные планы и структурированный постмортем.
5. Какие преимущества предоставляет использование Haystack для создания продвинутых систем искусственного интеллекта?
Использование Haystack предоставляет следующие преимущества:
* поддержка сложных агентских шаблонов, которые масштабируются по сложности;
* возможность выразить сложную логику агентов в блокноте;
* поддержание явного состояния и координация нескольких компонентов контролируемым и расширяемым способом.