В этом руководстве мы реализуем конвейер AI-агента с использованием Parsl, используя его возможности параллельного выполнения для запуска нескольких вычислительных задач в виде независимых приложений Python. Мы настраиваем локальный ThreadPoolExecutor для параллельного выполнения, определяем специализированные инструменты, такие как вычисление чисел Фибоначчи, подсчёт простых чисел, извлечение ключевых слов и имитация вызовов API, и координируем их через упрощённый планировщик, который сопоставляет цель пользователя с вызовами задач. Результаты всех задач объединяются и передаются через модель генерации текста Hugging Face для создания связного, удобочитаемого резюме.
Установка и импорт библиотек
Сначала мы устанавливаем необходимые библиотеки и импортируем все необходимые модули для нашего рабочего процесса. Затем настраиваем Parsl с помощью локального ThreadPoolExecutor для параллельного выполнения задач и загружаем эту конфигурацию, чтобы мы могли выполнять наши приложения Python параллельно.
“`python
import math, json, time, random
from typing import List, Dict, Any
import parsl
from parsl.config import Config
from parsl.executors import ThreadPoolExecutor
from parsl import python_app
parsl.load(Config(executors=[ThreadPoolExecutor(label=”local”, max_threads=8)]))
“`
Определение функций
Мы определяем четыре функции Parsl @python_app, которые выполняются асинхронно как часть рабочего процесса нашего агента. Мы создаём калькулятор чисел Фибоначчи, подпрограмму подсчёта простых чисел, средство извлечения ключевых слов для обработки текста и имитируемый инструмент, который имитирует внешние вызовы API со случайными задержками. Эти модульные приложения позволяют нам выполнять разнообразные вычисления параллельно, формируя строительные блоки для нашего мультиинструментария AI-агента.
“`python
@python_app
def calc_fibonacci(n: int) -> Dict[str, Any]:
def fib(k):
a, b = 0, 1
for _ in range(k): a, b = b, a + b
return a
t0 = time.time(); val = fib(n); dt = time.time() – t0
return {“task”: “fibonacci”, “n”: n, “value”: val, “secs”: round(dt, 4)}
“`
Генерация сводки
Мы реализуем функцию tinyllmsummary, которая использует конвейер Hugging Face с лёгкой моделью sshleifer/tiny-gpt2 для создания кратких сводок результатов нашего агента. Она форматирует собранные выходные данные задач в виде маркированных пунктов, добавляет подсказку «Заключение:» и извлекает только окончательное сгенерированное заключение для получения чистого, удобочитаемого резюме.
“`python
def tinyllmsummary(bullets: List[str]) -> str:
from transformers import pipeline
gen = pipeline(“text-generation”, model=”sshleifer/tiny-gpt2″)
prompt = “Summarize these agent results clearly:\n- ” + “\n- “.join(bullets) + “\nConclusion:”
out = gen(prompt, maxlength=160, dosample=False)[0][“generated_text”]
return out.split(“Conclusion:”, 1)[-1].strip()
“`
Планирование задач
Мы определяем функцию plan для сопоставления цели пользователя со структурированным списком вызовов инструментов. Она проверяет текст цели на наличие ключевых слов, таких как «fibonacci» или «primes», чтобы запустить определённые вычислительные задачи, а затем добавляет стандартные действия, такие как имитация запросов API, получение метрик и извлечение ключевых слов, формируя план выполнения для нашего AI-агента.
“`python
def plan(user_goal: str) -> List[Dict[str, Any]]:
intents = []
if “fibonacci” in user_goal.lower():
intents.append({“tool”: “calc_fibonacci”, “args”: {“n”: 35}})
if “primes” in user_goal.lower():
intents.append({“tool”: “countprimes”, “args”: {“limit”: 100000}})
intents += [
{“tool”: “simulatetool”, “args”: {“name”: “vectordbsearch”, “payload”: {“q”: usergoal}}},
{“tool”: “simulatetool”, “args”: {“name”: “metricsfetch”, “payload”: {“kpi”: “latency_ms”}}},
{“tool”: “extractkeywords”, “args”: {“text”: usergoal}}
]
return intents
“`
Выполнение агента
В функции runagent мы выполняем полный рабочий процесс агента, сначала создавая план задач на основе цели пользователя, а затем отправляя каждый инструмент в виде приложения Parsl для параллельного выполнения. Как только все фьючерсы будут завершены, мы преобразуем их результаты в чёткие маркированные пункты и передаём их в нашу функцию tinyllm_summary для создания краткого повествования. Функция возвращает структурированный словарь, содержащий исходную цель, подробные маркированные пункты, сгенерированное LLM резюме и необработанные выходные данные инструмента.
“`python
def runagent(usergoal: str) -> Dict[str, Any]:
tasks = plan(user_goal)
futures = []
for t in tasks:
if t[“tool”] == “calcfibonacci”: futures.append(calcfibonacci(t[“args”]))
elif t[“tool”] == “countprimes”: futures.append(countprimes(t[“args”]))
elif t[“tool”] == “extractkeywords”: futures.append(extractkeywords(t[“args”]))
elif t[“tool”] == “simulatetool”: futures.append(simulatetool(t[“args”]))
raw = [f.result() for f in futures]
bullets = []
for r in raw:
if r[“task”] == “fibonacci”:
bullets.append(f”Fibonacci({r[‘n’]}) = {r[‘value’]} computed in {r[‘secs’]}s.”)
elif r[“task”] == “count_primes”:
bullets.append(f”{r[‘count’]} primes found ≤ {r[‘limit’]}.”)
elif r[“task”] == “keywords”:
bullets.append(“Top keywords: ” + “, “.join(r[“keywords”]))
else:
bullets.append(f”Tool {r[‘task’]} responded with status={r[‘status’]}.”)
narrative = tinyllmsummary(bullets)
return {“goal”: user_goal, “bullets”: bullets, “summary”: narrative, “raw”: raw}
“`
Основная часть
В основном блоке выполнения мы определяем пример цели, которая объединяет числовой анализ, подсчёт простых чисел и генерацию сводки. Мы запускаем агента по этой цели, печатаем сгенерированные маркированные пункты, отображаем сгенерированное LLM резюме и предварительно просматриваем необработанный вывод JSON, чтобы проверить как удобочитаемые, так и структурированные результаты.
“`python
if name == “main“:
goal = (“Analyze fibonacci(35) performance, count primes under 100k, “
“and prepare a concise executive summary highlighting insights for planning.”)
result = run_agent(goal)
print(“\n=== Agent Bullets ===”)
for b in result[“bullets”]: print(“•”, b)
print(“\n=== LLM Summary ===\n”, result[“summary”])
print(“\n=== Raw JSON ===\n”, json.dumps(result[“raw”], indent=2)[:800], “…”)
“`
В заключение, эта реализация демонстрирует, как асинхронная модель приложения Parsl может эффективно координировать разнообразные рабочие нагрузки параллельно, позволяя AI-агенту комбинировать числовой анализ, обработку текста и имитацию внешних служб в едином конвейере. Интегрируя небольшую LLM на заключительном этапе, мы преобразуем структурированные результаты в естественный язык, иллюстрируя, как параллельные вычисления и модели ИИ могут быть объединены для создания отзывчивых, расширяемых агентов, подходящих для задач в реальном времени или крупномасштабных задач.
1. Какие инструменты и библиотеки используются в статье для реализации интеллектуальных параллельных рабочих процессов в Parsl?
В статье используются библиотеки Parsl для параллельного выполнения задач, а также Python для написания функций и приложений. Также применяется Hugging Face для генерации текста.
2. Как в статье реализована функция планирования задач для AI-агента?
Функция plan сопоставляет цель пользователя со структурированным списком вызовов инструментов. Она проверяет текст цели на наличие ключевых слов, таких как «fibonacci» или «primes», чтобы запустить определённые вычислительные задачи, а затем добавляет стандартные действия, такие как имитация запросов API, получение метрик и извлечение ключевых слов.
3. Какие функции определены в статье для выполнения асинхронных задач в рамках рабочего процесса AI-агента?
В статье определены четыре функции Parsl @python_app, которые выполняются асинхронно как часть рабочего процесса агента: калькулятор чисел Фибоначчи, подсчёт простых чисел, средство извлечения ключевых слов и имитируемый инструмент, который имитирует внешние вызовы API со случайными задержками.
4. Как в статье реализована генерация сводки результатов работы AI-агента?
Для генерации сводки используется функция tinyllmsummary, которая использует конвейер Hugging Face с лёгкой моделью sshleifer/tiny-gpt2 для создания кратких сводок результатов агента. Она форматирует собранные выходные данные задач в виде маркированных пунктов, добавляет подсказку «Заключение:» и извлекает только окончательное сгенерированное заключение для получения чистого, удобочитаемого резюме.
5. Какие преимущества даёт использование асинхронной модели приложения Parsl для координации разнообразных рабочих нагрузок?
Асинхронная модель приложения Parsl позволяет эффективно координировать разнообразные рабочие нагрузки параллельно, что ускоряет выполнение задач и повышает эффективность работы AI-агента. Это особенно полезно для задач, требующих одновременного выполнения нескольких вычислений или обработки больших объёмов данных.