An Implementation Data To Design Intelligent Parallel Workflows In Parsl For Multi-Software program AI Agent Execution

On this tutorial, we implement an AI agent pipeline using Parsl, leveraging its parallel execution capabilities to run quite a lot of computational duties as neutral Python apps. We configure an space ThreadPoolExecutor for concurrency, define specialised devices akin to Fibonacci computation, prime counting, key phrase extraction, and simulated API calls, and coordinate them by the use of a lightweight planner that maps an individual goal to course of invocations. The outputs from all duties are aggregated and handed by the use of a Hugging Face text-generation model to produce a coherent, human-readable summary. Check out the FULL CODES proper right here.

!pip arrange -q parsl transformers pace up


import math, json, time, random
from typing import File, 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)]))

We begin by placing within the required libraries & importing all very important modules for our workflow. We then configure Parsl with an space ThreadPoolExecutor to run duties concurrently and cargo this configuration so we are going to execute our Python apps in parallel. Check out the FULL CODES proper right here.

@python_app
def calc_fibonacci(n: int) -> Dict[str, Any]:
   def fib(okay):
       a, b = 0, 1
       for _ in range(okay): a, b = b, a + b
       return a
   t0 = time.time(); val = fib(n); dt = time.time() - t0
   return {"course of": "fibonacci", "n": n, "value": val, "secs": spherical(dt, 4)}


@python_app
def extract_keywords(textual content material: str, okay: int = 8) -> Dict[str, Any]:
   import re, collections
   phrases = [w.lower() for w in re.findall(r"[a-zA-Z][a-zA-Z0-9-]+", textual content material)]
   stop = set("the a an and or to of is are was had been be been in on for with as by from at this that it its if then else not no".break up())
   cand = [w for w in words if w not in stop and len(w) > 3]
   freq = collections.Counter(cand)
   scored = sorted(freq.objects(), key=lambda x: (x[1], len(x[0])), reverse=True)[:k]
   return {"course of":"key phrases","key phrases":[w for w,_ in scored]}


@python_app
def simulate_tool(determine: str, payload: Dict[str, Any]) -> Dict[str, Any]:
   time.sleep(0.3 + random.random()*0.5)
   return {"course of": determine, "payload": payload, "standing": "okay", "timestamp": time.time()}

We define 4 Parsl @python_app options that run asynchronously as part of our agent’s workflow. We create a Fibonacci calculator, a prime-counting routine, a key phrase extractor for textual content material processing, and a simulated gadget that mimics exterior API calls with randomized delays. These modular apps enable us to hold out quite a few computations in parallel, forming the developing blocks for our multi-tool AI agent. Check out the FULL CODES proper right here.

def tiny_llm_summary(bullets: File[str]) -> str:
   from transformers import pipeline
   gen = pipeline("text-generation", model="sshleifer/tiny-gpt2")
   quick = "Summarize these agent outcomes clearly:n- " + "n- ".be part of(bullets) + "nConclusion:"
   out = gen(quick, max_length=160, do_sample=False)[0]["generated_text"]
   return out.break up("Conclusion:", 1)[-1].strip()

We implement a tiny_llm_summary function that makes use of Hugging Face’s pipeline with the lightweight sshleifer/tiny-gpt2 model to generate concise summaries of our agent’s outcomes. It codecs the collected course of outputs as bullet components, appends a “Conclusion:” cue, and extracts solely the final word generated conclusion for a transparent, human-readable summary. Check out the FULL CODES proper right here.

def plan(user_goal: str) -> File[Dict[str, Any]]:
   intents = []
   if "fibonacci" in user_goal.lower():
       intents.append({"gadget":"calc_fibonacci", "args":{"n":35}})
   if "primes" in user_goal.lower():
       intents.append({"gadget":"count_primes", "args":{"prohibit":100_000}})
   intents += [
       {"tool":"simulate_tool", "args":{"name":"vector_db_search","payload":{"q":user_goal}}},
       {"tool":"simulate_tool", "args":{"name":"metrics_fetch","payload":{"kpi":"latency_ms"}}},
       {"tool":"extract_keywords", "args":{"text":user_goal}}
   ]
   return intents

We define the plan function to map an individual’s goal proper right into a structured guidelines of gadget invocations. It checks the goal textual content material for key phrases like “fibonacci” or “primes” to set off specific computational duties, then offers default actions akin to simulated API queries, metrics retrieval, and key phrase extraction, forming the execution blueprint for our AI agent. Check out the FULL CODES proper right here.

def run_agent(user_goal: str) -> Dict[str, Any]:
   duties = plan(user_goal)
   futures = []
   for t in duties:
       if t["tool"]=="calc_fibonacci": futures.append(calc_fibonacci(**t["args"]))
       elif t["tool"]=="count_primes": futures.append(count_primes(**t["args"]))
       elif t["tool"]=="extract_keywords": futures.append(extract_keywords(**t["args"]))
       elif t["tool"]=="simulate_tool": futures.append(simulate_tool(**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"]=="key phrases":
           bullets.append("Excessive key phrases: " + ", ".be part of(r["keywords"]))
       else:
           bullets.append(f"Software program {r['task']} responded with standing={r['status']}.")


   narrative = tiny_llm_summary(bullets)
   return {"goal": user_goal, "bullets": bullets, "summary": narrative, "raw": raw}

Throughout the run_agent function, we execute the entire agent workflow by first producing a course of plan from the individual’s goal, then dispatching each gadget as a Parsl app to run in parallel. As quickly as all futures are full, we convert their outcomes into clear bullet components and feed them to our tiny_llm_summary function to create a concise narrative. The function returns a structured dictionary containing the distinctive goal, detailed bullet components, the LLM-generated summary, and the raw gadget outputs. Check out the FULL CODES proper right here.

if __name__ == "__main__":
   goal = ("Analyze fibonacci(35) effectivity, rely primes under 100k, "
           "and put collectively a concise govt summary highlighting insights for planning.")
   finish end result = run_agent(goal)
   print("n=== Agent Bullets ===")
   for b in finish end result["bullets"]: print("•", b)
   print("n=== LLM Summary ===n", finish end result["summary"])
   print("n=== Raw JSON ===n", json.dumps(finish end result["raw"], indent=2)[:800], "...")

Within the precept execution block, we define a sample goal that mixes numeric computation, prime counting, and summary period. We run the agent on this goal, print the generated bullet components, present the LLM-crafted summary, and preview the raw JSON output to verify every the human-readable and structured outcomes.

In conclusion, this implementation demonstrates how Parsl’s asynchronous app model can successfully orchestrate quite a few workloads in parallel, enabling an AI agent to combine numerical analysis, textual content material processing, and simulated exterior corporations in a unified pipeline. By integrating a small LLM on the final word stage, we rework structured outcomes into pure language, illustrating how parallel computation and AI fashions could also be blended to create responsive, extensible brokers acceptable for real-time or large-scale duties.

Check out the FULL CODES proper right here. Be pleased to try our GitHub Net web page for Tutorials, Codes and Notebooks. Moreover, be at liberty to adjust to us on Twitter and don’t neglect to affix our 100k+ ML SubReddit and Subscribe to our Publication.