Step-by-Step Information to AI Agent Growth Utilizing Microsoft Agent-Lightning

!pip -q set up agentlightning openai nest_asyncio python-dotenv > /dev/null
import os, threading, time, asyncio, nest_asyncio, random
from getpass import getpass
from agentlightning.litagent import LitAgent
from agentlightning.coach import Coach
from agentlightning.server import AgentLightningServer
from agentlightning.sorts import PromptTemplate
import openai
if not os.getenv("OPENAI_API_KEY"):
   attempt:
       os.environ["OPENAI_API_KEY"] = getpass("🔑 Enter OPENAI_API_KEY (go away clean if utilizing a neighborhood/proxy base): ") or ""
   besides Exception:
       cross
MODEL = os.getenv("MODEL", "gpt-4o-mini")

class QAAgent(LitAgent):
   def training_rollout(self, job, rollout_id, assets):
       """Given a job {'immediate':..., 'reply':...}, ask LLM utilizing the server-provided system immediate and return a reward in [0,1]."""
       sys_prompt = assets["system_prompt"].template
       consumer = job["prompt"]; gold = job.get("reply","").strip().decrease()
       attempt:
           r = openai.chat.completions.create(
               mannequin=MODEL,
               messages=[{"role":"system","content":sys_prompt},
                         {"role":"user","content":user}],
               temperature=0.2,
           )
           pred = r.decisions[0].message.content material.strip()
       besides Exception as e:
           pred = f"[error]{e}"
       def rating(pred, gold):
           P = pred.decrease()
           base = 1.0 if gold and gold in P else 0.0
           gt = set(gold.break up()); pr = set(P.break up());
           inter = len(gt & pr); denom = (len(gt)+len(pr)) or 1
           overlap = 2*inter/denom
           brevity = 0.2 if base==1.0 and len(P.break up())<=8 else 0.0
           return max(0.0, min(1.0, 0.7*base + 0.25*overlap + brevity))
       return float(rating(pred, gold))

TASKS = [
   {"prompt":"Capital of France?","answer":"Paris"},
   {"prompt":"Who wrote Pride and Prejudice?","answer":"Jane Austen"},
   {"prompt":"2+2 = ?","answer":"4"},
]
PROMPTS = [
   "You are a terse expert. Answer with only the final fact, no sentences.",
   "You are a helpful, knowledgeable AI. Prefer concise, correct answers.",
   "Answer as a rigorous evaluator; return only the canonical fact.",
   "Be a friendly tutor. Give the one-word answer if obvious."
]
nest_asyncio.apply()
HOST, PORT = "127.0.0.1", 9997

async def run_server_and_search():
   server = AgentLightningServer(host=HOST, port=PORT)
   await server.begin()
   print("✅ Server began")
   await asyncio.sleep(1.5)
   outcomes = []
   for sp in PROMPTS:
       await server.update_resources({"system_prompt": PromptTemplate(template=sp, engine="f-string")})
       scores = []
       for t in TASKS:
           tid = await server.queue_task(pattern=t, mode="practice")
           rollout = await server.poll_completed_rollout(tid, timeout=40)  # waits for a employee
           if rollout is None:
               print("⏳ Timeout ready for rollout; persevering with...")
               proceed
           scores.append(float(getattr(rollout, "final_reward", 0.0)))
       avg = sum(scores)/len(scores) if scores else 0.0
       print(f"🔎 Immediate avg: {avg:.3f}  |  {sp}")
       outcomes.append((sp, avg))
   greatest = max(outcomes, key=lambda x: x[1]) if outcomes else ("",0)
   print("n🏁 BEST PROMPT:", greatest[0], " | rating:", f"{greatest[1]:.3f}")
   await server.cease()

def run_client_in_thread():
   agent = QAAgent()
   coach = Coach(n_workers=2)    
   coach.match(agent, backend=f"http://{HOST}:{PORT}")
client_thr = threading.Thread(goal=run_client_in_thread, daemon=True)
client_thr.begin()
asyncio.run(run_server_and_search())