Step Guide by Step in Ai Agent Development using Microsoft Agent-Lightning
In this lesson, we are moving with enhanced AI agent using Microsoft's Agent-lightning Outline. We use all the item directly within Google Colab, which means we can try both servers and customers in one place. By explaining a small pipe agent, connecting to the local Agent-Lightning Server, and train the Multiple System Complets, we can see how the frame supports the app updates, in the default assessment. Look Full codes here.
!pip -q install 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.trainer import Trainer
from agentlightning.server import AgentLightningServer
from agentlightning.types import PromptTemplate
import openai
if not os.getenv("OPENAI_API_KEY"):
try:
os.environ["OPENAI_API_KEY"] = getpass("š Enter OPENAI_API_KEY (leave blank if using a local/proxy base): ") or ""
except Exception:
pass
MODEL = os.getenv("MODEL", "gpt-4o-mini")We start by installing the required libraries and invites all the main modules we need to find the agent's lightning. We also put our Openai API key safely and explain the model to use for the study. Look Full codes here.
class QAAgent(LitAgent):
def training_rollout(self, task, rollout_id, resources):
"""Given a task {'prompt':..., 'answer':...}, ask LLM using the server-provided system prompt and return a reward in [0,1]."""
sys_prompt = resources["system_prompt"].template
user = task["prompt"]; gold = task.get("answer","").strip().lower()
try:
r = openai.chat.completions.create(
model=MODEL,
messages=[{"role":"system","content":sys_prompt},
{"role":"user","content":user}],
temperature=0.2,
)
pred = r.choices[0].message.content.strip()
except Exception as e:
pred = f"[error]{e}"
def score(pred, gold):
P = pred.lower()
base = 1.0 if gold and gold in P else 0.0
gt = set(gold.split()); pr = set(P.split());
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.split())<=8 else 0.0
return max(0.0, min(1.0, 0.7*base + 0.25*overlap + brevity))
return float(score(pred, gold))We explain the simple qaaagent by extinguish the latagent, where we handle each of the issues immediately by sending user to the llm, collecting the answer, and a golden response. We design a reward work to ensure accuracy, token to clean, and brevity, making an agent read and produce short and accurate results. Look Full codes here.
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", 9997We describe a small sign with three QA works and a multi-curate in the preferred system it moves to get better. We then put the nest_asyncio and set our local server and local port, which we have to use the agent's server within a single colab runte. Look Full codes here.
async def run_server_and_search():
server = AgentLightningServer(host=HOST, port=PORT)
await server.start()
print("ā
Server started")
await asyncio.sleep(1.5)
results = []
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(sample=t, mode="train")
rollout = await server.poll_completed_rollout(tid, timeout=40) # waits for a worker
if rollout is None:
print("ā³ Timeout waiting for rollout; continuing...")
continue
scores.append(float(getattr(rollout, "final_reward", 0.0)))
avg = sum(scores)/len(scores) if scores else 0.0
print(f"š Prompt avg: {avg:.3f} | {sp}")
results.append((sp, avg))
best = max(results, key=lambda x: x[1]) if results else ("",0)
print("nš BEST PROMPT:", best[0], " | score:", f"{best[1]:.3f}")
await server.stop() We begin the agents server and apply through our financial system, revitalize the stolen system .Prompt before downloading each training work. We have been thrown to vote completed, between the Middle Ages immediately, report the best best, and make the server kind. Look Full codes here.
def run_client_in_thread():
agent = QAAgent()
trainer = Trainer(n_workers=2)
trainer.fit(agent, backend=f"
client_thr = threading.Thread(target=run_client_in_thread, daemon=True)
client_thr.start()
asyncio.run(run_server_and_search())We introduce the client to a separate cable with two similar workers, allowing the services to be sent by server. At the same time, we use a server loop, checking the opposite increase, collecting the output results, and reports immediately the excellent program based on the middle.
In conclusion, we will see that the Agent lightning – how it helps us to build a variable agent with a few lines of the code. We can begin the server, run the same customers, check the dynamic systems, and automatically measure the performance, everything in one column area. This shows how the framework guides the process of construction, assessment and performing the AGENTS in an orderly manner.
Look Full codes here. Feel free to look our GITHUB page for tutorials, codes and letters of writing. Also, feel free to follow it Sane and don't forget to join ours 100K + ml subreddit Then sign up for Our newspaper.
Asphazzaq is a Markteach Media Inc. According to a View Business and Developer, Asifi is committed to integrating a good social intelligence. His latest attempt is launched by the launch of the chemistrylife plan for an intelligence, MarktechPost, a devastating intimate practice of a machine learning and deep learning issues that are clearly and easily understood. The platform is adhering to more than two million moon visits, indicating its popularity between the audience.
