diff --git a/agent.py b/agent.py index 35c8acc..7433ce5 100644 --- a/agent.py +++ b/agent.py @@ -1,23 +1,34 @@ """ -Onboarding Agent -Meridian Financial Services workshop repo +Onboarding Agent — Meridian Financial Services workshop repo +============================================================ -ReAct loop: Reason · Act · Observe · Repeat +A small, readable ReAct agent (Reason → Act → Observe → repeat) that drives +the four-step onboarding workflow defined in system_prompt.md: + + 1. Retrieve the employee profile from HR + 2. Check compliance acknowledgement status + 3. Provision system access — only after compliance is CLEARED + 4. Send a welcome notification + +The behaviour of the agent lives in system_prompt.md (structured with the +RICE principle: Role, Intent, Context, Enforcement). This file only supplies +the *machinery*: the tools, the reasoning loop, and the plumbing that lets a +language model use those tools safely. Run: - python3 agent.py --dry-run # verify environment, no API calls - python3 agent.py --employee EMP-2026-0847 # full agent run - python3 agent.py --employee EMP-2026-0847 --no-enforcement # demo without compliance gate + python agent.py --dry-run # verify setup, no API calls + python agent.py --employee EMP-2026-0847 # full agent run + python agent.py --employee EMP-2026-0847 --no-enforcement # demo: compliance gate removed + python agent.py --employee EMP-2026-0847 --reset # clear cached session first """ import os -import sys import json import argparse import importlib.util from datetime import datetime -# ── PATHS ───────────────────────────────────────────────────────── +# ── PATHS ────────────────────────────────────────────────────────── ROOT = os.path.dirname(os.path.abspath(__file__)) TOOLS_DIR = os.path.join(ROOT, 'tools') DATA_DIR = os.path.join(ROOT, 'data') @@ -26,8 +37,38 @@ COMPLIANCE_DB = os.path.join(DATA_DIR, 'compliance_state.json') CACHE_PATH = os.path.join(DATA_DIR, 'session_cache.json') +# Enforcement rule 2: at most 3 attempts per tool call before escalation. +MAX_TOOL_RETRIES = 3 + +ENV_PATH = os.path.join(ROOT, '.env') + + +# ── ENV LOADING ──────────────────────────────────────────────────── +# Load KEY=value pairs from a local .env (git-ignored) so secrets like +# GROQ_API_KEY never live in code and never get committed. No dependency. + +def load_dotenv(path: str = ENV_PATH) -> None: + if not os.path.exists(path): + return + with open(path) as f: + for line in f: + line = line.strip() + if not line or line.startswith('#') or '=' not in line: + continue + key, _, val = line.partition('=') + os.environ.setdefault(key.strip(), val.strip().strip('"').strip("'")) + + +# ── AUDIT-SAFE LOGGING (Enforcement rule 4) ──────────────────────── +# Never write name/email/personal details to logs — only the employee_id. + +def log(msg: str) -> None: + print(f" [log] {msg}") -# ── SESSION CACHE ──────────────────────────────────────────────── + +# ── SESSION CACHE ────────────────────────────────────────────────── +# Persists the step history per employee so a run can be resumed. The agent +# is expensive (each step is an LLM call), so we don't want to redo work. def load_cache() -> dict: if os.path.exists(CACHE_PATH): @@ -41,9 +82,14 @@ def save_cache(cache: dict) -> None: json.dump(cache, f, indent=2) -# ── TOOL: get_employee_profile ──────────────────────────────────── +# ── TOOLS ────────────────────────────────────────────────────────── +# Four tools back the four workflow steps. Two read local JSON "databases"; +# two are pre-built stubs in tools/ that simulate side effects and print what +# they would do. The agent never calls these directly — it emits a tool name +# and JSON arguments, and the ReAct loop dispatches through the TOOLS registry. def get_employee_profile(employee_id: str) -> dict: + """Retrieve the employee profile from HR. Raises EmployeeNotFound.""" with open(EMPLOYEES_DB) as f: db = json.load(f) if employee_id not in db: @@ -51,38 +97,43 @@ def get_employee_profile(employee_id: str) -> dict: return db[employee_id] -# ── TOOL: check_compliance_status ───────────────────────────────── - def check_compliance_status(employee_id: str) -> dict: + """Return compliance acknowledgements and overall_status (CLEARED|PENDING). + + An employee with no record is treated as fully PENDING — fail closed, so a + missing record can never accidentally pass the compliance gate. + """ with open(COMPLIANCE_DB) as f: db = json.load(f) if employee_id not in db: return { "employee_id": employee_id, - "overall_status": "PENDING", "code_of_conduct": "PENDING", "data_handling_policy": "PENDING", "security_guidelines": "PENDING", - "posh_training": "PENDING" + "posh_training": "PENDING", + "overall_status": "PENDING", } return db[employee_id] -# ── PRE-BUILT STUBS (loaded from tools/) ───────────────────────── - def _load_stub(filename: str, fn_name: str): - """Dynamically load a function from tools/ directory.""" + """Dynamically import a single function from the tools/ directory.""" path = os.path.join(TOOLS_DIR, filename) spec = importlib.util.spec_from_file_location(fn_name, path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return getattr(mod, fn_name) + provision_access = _load_stub('provision_access.py', 'provision_access') send_notification = _load_stub('send_notification.py', 'send_notification') -# ── TOOL REGISTRY ───────────────────────────────────────────────── +# ── TOOL REGISTRY & SCHEMAS ──────────────────────────────────────── +# TOOLS maps name -> callable (used to dispatch). TOOL_SCHEMAS is the +# human/LLM-readable description injected into the prompt so the model knows +# what each tool does and what arguments it takes. TOOLS = { "get_employee_profile": get_employee_profile, @@ -94,313 +145,423 @@ def _load_stub(filename: str, fn_name: str): TOOL_SCHEMAS = [ { "name": "get_employee_profile", - "description": "Retrieve employee profile (role, dept, manager, access_profile) from HR database", - "params": ["employee_id: str"] + "params": ["employee_id: str"], + "description": "Retrieve employee profile (name, role, level, department, " + "manager_id, employment_type, access_profile) from HR. " + "Raises EmployeeNotFound if no record exists.", }, { "name": "check_compliance_status", - "description": "Check whether the employee has acknowledged code_of_conduct, data_handling_policy, security_guidelines, posh_training. Returns overall_status: CLEARED or PENDING.", - "params": ["employee_id: str"] + "params": ["employee_id: str"], + "description": "Check acknowledgement of code_of_conduct, data_handling_policy, " + "security_guidelines, posh_training. Returns overall_status: " + "CLEARED or PENDING.", }, { "name": "provision_access", - "description": "Provision GitHub, JIRA, Confluence, Slack access based on access_profile. Only call after compliance_status is CLEARED.", - "params": ["employee_id: str", "access_profile: dict"] + "params": ["employee_id: str", "access_profile: dict"], + "description": "Provision GitHub, JIRA, Confluence, Slack per the access_profile. " + "Only call after compliance overall_status is CLEARED. Never " + "provision anything not listed in access_profile.", }, { "name": "send_notification", - "description": "Send a notification. For email channel, recipient MUST be the employee's email address (from their profile). For slack, use Slack user ID. For whatsapp, use phone number.", - "params": ["recipient: str", "channel: str (email|slack|whatsapp)", "subject: str", "body: str"] - } + "params": ["recipient: str", "channel: str (email|slack|whatsapp)", "subject: str", "body: str"], + "description": "Send a notification. For channel=email, recipient MUST be the " + "employee's real email from their profile (not the employee_id).", + }, ] -# ── LLM CALL (Gemini default) ──────────────────────────────────── +# ── LLM CALL ─────────────────────────────────────────────────────── +# The 'brain'. Swap the provider/model here without touching the loop. + +def get_api_key() -> str: + """Return the Groq API key from the environment, or prompt for it once. + + Resolution order: GROQ_API_KEY env var (incl. values loaded from .env) -> + secure interactive prompt (getpass, so the key is never echoed or logged). + The prompted value is cached in the environment for the rest of the run. + """ + key = (os.environ.get('GROQ_API_KEY') or '').strip() + if key: + return key + import getpass + try: + key = getpass.getpass(" >>> Enter your GROQ_API_KEY (input hidden): ").strip() + except (EOFError, OSError): + key = '' + if not key: + raise RuntimeError( + "No GROQ_API_KEY provided. Add it to the .env file or set the " + "environment variable, or run with --offline for no key." + ) + os.environ['GROQ_API_KEY'] = key # cache for the rest of this run + return key + def call_llm(prompt: str) -> str: - """Call the configured LLM provider. Default: Groq Llama 3.3 70B.""" + """Call the configured LLM. Default: Groq Llama 3.3 70B.""" from groq import Groq - api_key = os.environ.get('GROQ_API_KEY') - if not api_key: - raise RuntimeError("GROQ_API_KEY not set. Run: export GROQ_API_KEY=your-key") - client = Groq(api_key=api_key) + client = Groq(api_key=get_api_key()) response = client.chat.completions.create( model='llama-3.3-70b-versatile', - messages=[{"role": "user", "content": prompt}] + messages=[{"role": "user", "content": prompt}], ) return response.choices[0].message.content -# ── THOUGHT PARSER ─────────────────────────────────────────────── +# ── BRAINS ───────────────────────────────────────────────────────── +# A "brain" decides the next action. It takes (system_prompt, history, goal) +# and returns raw THOUGHT/ACTION text — exactly what parse_action reads. This +# lets us swap a real LLM for a deterministic mock without touching the loop. -def parse_thought(text: str) -> dict: - """ - Parse LLM output into a structured action. +def live_brain(system_prompt: str, history: list, goal: str) -> str: + """Real reasoning: build the prompt and ask the LLM. Needs GROQ_API_KEY.""" + return call_llm(build_prompt(system_prompt, history, goal)) - Expected output formats: - ACTION: call_tool - TOOL: - INPUT: - ACTION: ask_human - QUESTION: +def make_offline_brain(employee_id: str, enforce: bool = True): + """Return a deterministic mock brain that needs no API key. - ACTION: complete - SUMMARY: + It reads the run history and emits the next scripted step, following the + same workflow and Enforcement rules a well-behaved LLM would. Great for + students without a key, and for showing the loop mechanics in isolation. """ - lines = [l.strip() for l in text.strip().split('\n') if l.strip()] - fields = {} - cur_key = None - cur_val = [] - - for line in lines: - if ':' in line: - key, _, val = line.partition(':') - key = key.strip().upper() - if key in ('ACTION', 'TOOL', 'INPUT', 'QUESTION', 'SUMMARY', 'THOUGHT'): - if cur_key: - fields[cur_key] = ' '.join(cur_val).strip() - cur_key = key - cur_val = [val.strip()] if val.strip() else [] - continue - if cur_key: - cur_val.append(line) - if cur_key: - fields[cur_key] = ' '.join(cur_val).strip() + def _block(thought: str, action: str, **kw) -> str: + lines = [f"THOUGHT: {thought}", f"ACTION: {action}"] + if action == 'call_tool': + lines += [f"TOOL: {kw['tool']}", f"INPUT: {json.dumps(kw['input'])}"] + elif action == 'ask_human': + lines += [f"QUESTION: {kw['question']}"] + elif action == 'complete': + lines += [f"SUMMARY: {kw['summary']}"] + return '\n'.join(lines) + + def brain(system_prompt: str, history: list, goal: str) -> str: + profile = compliance = None + reminder_sent = welcome_sent = provisioned = False + human_reply = None + + for step in history: + if step.get('action_type') == 'call_tool': + tool, obs = step['tool'], step.get('observation', {}) + inp = step.get('input', {}) or {} + if tool == 'get_employee_profile' and isinstance(obs, dict) and 'access_profile' in obs: + profile = obs + elif tool == 'check_compliance_status' and isinstance(obs, dict) and 'overall_status' in obs: + compliance = obs + elif tool == 'send_notification' and isinstance(obs, dict) and obs.get('status') == 'SUCCESS': + if 'welcome' in str(inp.get('subject', '')).lower(): + welcome_sent = True + else: + reminder_sent = True + elif tool == 'provision_access' and isinstance(obs, dict) and obs.get('status') == 'SUCCESS': + provisioned = True + elif step.get('action_type') == 'ask_human': + human_reply = (step.get('human_input') or '').strip().lower() + + # 1) Profile + if profile is None: + return _block("I need the employee profile before anything else.", + 'call_tool', tool='get_employee_profile', + input={"employee_id": employee_id}) + + email = profile.get('email', employee_id) + access_profile = profile.get('access_profile', {}) + + # 2) Compliance + if compliance is None: + return _block("I have the profile. Check compliance before provisioning.", + 'call_tool', tool='check_compliance_status', + input={"employee_id": employee_id}) + + cleared = compliance.get('overall_status') == 'CLEARED' + + # 3) Compliance gate (Enforcement rule 1) — only when enforcing. + if enforce and not cleared: + if not reminder_sent: + return _block("Compliance is PENDING. Send a reminder before asking to proceed.", + 'call_tool', tool='send_notification', + input={"recipient": email, "channel": "email", + "subject": "Action required: complete onboarding policies", + "body": "Please complete your pending compliance acknowledgements to finish onboarding."}) + if human_reply is None: + return _block("Reminder sent. Compliance is still PENDING, so I must ask a human before provisioning.", + 'ask_human', + question="Compliance is PENDING. Confirm whether I should proceed to provision access anyway. (yes/no)") + negative = any(w in human_reply for w in ('no', 'stop', 'deny', 'decline', 'cancel', 'hold')) + if negative: + return _block("Human declined. Halt without provisioning.", + 'complete', + summary=f"Onboarding for {employee_id} halted: compliance PENDING and human declined to proceed. No access provisioned.") + # positive → fall through to provisioning + + # 4) Provision (only systems in access_profile — Enforcement rule 3) + if not provisioned: + reason = ("Compliance CLEARED — safe to provision." + if cleared else "Human confirmed — provisioning per access_profile only.") + return _block(reason, 'call_tool', tool='provision_access', + input={"employee_id": employee_id, "access_profile": access_profile}) + + # 5) Welcome + if not welcome_sent: + return _block("Access provisioned. Send the welcome notification.", + 'call_tool', tool='send_notification', + input={"recipient": email, "channel": "email", + "subject": "Welcome to Meridian!", + "body": "Welcome aboard! Your accounts are ready. Reach out to your manager with any questions."}) + + # 6) Done + gate = "CLEARED" if cleared else "PENDING (proceeded after human confirmation)" + return _block("Workflow finished.", 'complete', + summary=(f"Onboarding complete for {employee_id}. Compliance {gate}; " + f"access provisioned per profile; welcome notification sent.")) + + return brain + + +# ── PROMPT ASSEMBLY ──────────────────────────────────────────────── +# Each reasoning step, we hand the model: the system prompt (Role/Intent/ +# Context/Enforcement), the tool list, the goal, and the full history so far. +# The model replies in a strict THOUGHT/ACTION format that parse_action reads. + +RESPONSE_FORMAT = """Decide your next action. Respond in EXACTLY this format and nothing else: - action_type = fields.get('ACTION', '').lower() +THOUGHT: +ACTION: - if 'call_tool' in action_type or 'call tool' in action_type: - tool_name = fields.get('TOOL', '').strip() - input_str = fields.get('INPUT', '{}').strip() - try: - tool_input = json.loads(input_str) - except json.JSONDecodeError: - tool_input = {"raw": input_str} - return { - "type": "call_tool", - "tool": tool_name, - "input": tool_input, - "thought": fields.get('THOUGHT', '') - } +If ACTION is call_tool: +TOOL: +INPUT: - if 'ask_human' in action_type or 'ask human' in action_type: - return { - "type": "ask_human", - "question": fields.get('QUESTION', 'Please confirm to proceed'), - "thought": fields.get('THOUGHT', '') - } +If ACTION is ask_human: +QUESTION: - if 'complete' in action_type: - return { - "type": "complete", - "summary": fields.get('SUMMARY', 'Done'), - "thought": fields.get('THOUGHT', '') - } - - return { - "type": "unknown", - "raw": text, - "thought": fields.get('THOUGHT', '') - } +If ACTION is complete: +SUMMARY: +Respond with ONLY the THOUGHT and ACTION block. No prose before or after.""" -# ── REACT LOOP ─────────────────────────────────────────────────── -def build_prompt(system_prompt: str, history: list, tool_schemas: list, goal: str) -> str: - """Assemble the prompt for the next reasoning step.""" - tools_text = '\n'.join([ +def _render_history(history: list) -> str: + out = [] + for i, step in enumerate(history, start=1): + block = [f"\nSTEP {i}:", f"THOUGHT: {step.get('thought', '')}"] + if step.get('action_type') == 'call_tool': + block.append(f"ACTION: call_tool") + block.append(f"TOOL: {step['tool']}") + block.append(f"INPUT: {json.dumps(step['input'])}") + # Do NOT truncate: the model must see the full observation (e.g. the + # complete access_profile) or it will guess and provision wrong data. + block.append(f"OBSERVATION: {json.dumps(step['observation'])}") + elif step.get('action_type') == 'ask_human': + block.append(f"ACTION: ask_human") + block.append(f"QUESTION: {step['question']}") + block.append(f"HUMAN_INPUT: {step['human_input']}") + out.append('\n'.join(block)) + return ''.join(out) if out else ' (no steps yet)' + + +def build_prompt(system_prompt: str, history: list, goal: str) -> str: + tools_text = '\n'.join( f" - {t['name']}({', '.join(t['params'])}): {t['description']}" - for t in tool_schemas - ]) + for t in TOOL_SCHEMAS + ) + return ( + f"{system_prompt}\n\n" + f"AVAILABLE TOOLS:\n{tools_text}\n\n" + f"GOAL: {goal}\n\n" + f"HISTORY SO FAR:{_render_history(history)}\n\n" + f"{RESPONSE_FORMAT}" + ) - history_text = '' - for i, step in enumerate(history): - if 'thought' in step: - history_text += f"\nSTEP {i+1}:\nTHOUGHT: {step.get('thought','')}\n" - if step.get('action_type') == 'call_tool': - history_text += f"ACTION: call_tool\nTOOL: {step['tool']}\nINPUT: {json.dumps(step['input'])}\nOBSERVATION: {json.dumps(step['observation'])[:500]}\n" - elif step.get('action_type') == 'ask_human': - history_text += f"ACTION: ask_human\nQUESTION: {step['question']}\nHUMAN_INPUT: {step['human_input']}\n" - return f"""{system_prompt} +# ── ACTION PARSER ────────────────────────────────────────────────── +# Turn the model's free text into a structured action dict. + +def parse_action(text: str) -> dict: + fields, cur_key, cur_val = {}, None, [] + known = ('THOUGHT', 'ACTION', 'TOOL', 'INPUT', 'QUESTION', 'SUMMARY') + + for line in (l.rstrip() for l in text.strip().splitlines()): + if not line.strip(): + continue + head = line.split(':', 1)[0].strip().upper() + if ':' in line and head in known: + if cur_key: + fields[cur_key] = ' '.join(cur_val).strip() + cur_key = head + first = line.split(':', 1)[1].strip() + cur_val = [first] if first else [] + elif cur_key: + cur_val.append(line.strip()) + if cur_key: + fields[cur_key] = ' '.join(cur_val).strip() -AVAILABLE TOOLS: -{tools_text} + thought = fields.get('THOUGHT', '') + action = fields.get('ACTION', '').lower() -GOAL: {goal} + if 'call_tool' in action or 'call tool' in action: + raw_input = fields.get('INPUT', '{}').strip() + try: + tool_input = json.loads(raw_input) + except json.JSONDecodeError: + tool_input = {"raw": raw_input} + return {"type": "call_tool", "tool": fields.get('TOOL', '').strip(), + "input": tool_input, "thought": thought} -HISTORY SO FAR:{history_text if history_text else ' (no steps yet)'} + if 'ask_human' in action or 'ask human' in action: + return {"type": "ask_human", + "question": fields.get('QUESTION', 'Please confirm to proceed.'), + "thought": thought} -Decide your next action. Respond in this exact format: + if 'complete' in action: + return {"type": "complete", "summary": fields.get('SUMMARY', 'Done.'), + "thought": thought} -THOUGHT: -ACTION: + return {"type": "unknown", "raw": text, "thought": thought} -If call_tool: -TOOL: -INPUT: -If ask_human: -QUESTION: +# ── TOOL DISPATCH with retry/escalation (Enforcement rule 2) ─────── -If complete: -SUMMARY: +def dispatch_tool(tool_name: str, tool_input: dict) -> dict: + """Call a tool, retrying up to MAX_TOOL_RETRIES times on exception. -Respond with ONLY the THOUGHT and ACTION block. No prose before or after.""" + Returns the tool result, or an {"error", "escalate": True} dict after the + retry budget is exhausted so the model can escalate to the manager. + """ + if tool_name not in TOOLS: + return {"error": f"Unknown tool: {tool_name}"} + + fn = TOOLS[tool_name] + last_error = None + for attempt in range(1, MAX_TOOL_RETRIES + 1): + try: + return fn(**tool_input) if isinstance(tool_input, dict) else fn(tool_input) + except Exception as e: # noqa: BLE001 — surface any tool failure to the agent + last_error = str(e) + log(f"tool '{tool_name}' attempt {attempt}/{MAX_TOOL_RETRIES} failed") + return { + "error": last_error, + "attempts": MAX_TOOL_RETRIES, + "escalate": True, + "note": "Retry budget exhausted. Escalate to the manager via send_notification and stop.", + } + + +# ── REACT LOOP ───────────────────────────────────────────────────── +def run_agent(employee_id: str, system_prompt: str, brain=live_brain, max_steps: int = 15) -> dict: + """Run Reason → Act → Observe until complete, parse error, or max_steps. -def run_agent(employee_id: str, system_prompt: str, max_steps: int = 15) -> dict: - """Run the ReAct loop until goal complete or max_steps reached.""" - goal = f"Onboard employee {employee_id}: retrieve profile, check compliance, provision access (only if cleared), send welcome notification." + `brain(system_prompt, history, goal) -> raw text` produces each decision. + Defaults to the live LLM; pass an offline brain to run without an API key. + """ + goal = (f"Onboard employee {employee_id}: retrieve profile, check compliance, " + f"provision access (only if CLEARED), send welcome notification.") - # ── Resume from cache ────────────────────────────────────── - cache = load_cache() + cache = load_cache() history = cache.get(employee_id, []) if history: - print(f" [cache] Resuming session for {employee_id} ({len(history)} previous step(s) loaded)") + log(f"resuming session for {employee_id} ({len(history)} step(s) cached)") - # ReAct loop: each pass = Reason -> Act -> Observe. The loop ends when the - # agent returns 'complete', hits a parse error, or reaches max_steps. - for step_num in range(len(history) + 1, max_steps + 1): - print(f"\n+--- STEP {step_num} {'-'*44}") + def persist(): + cache[employee_id] = history + save_cache(cache) - # REASON: rebuild the prompt with the full history so the model can see - # everything that has happened, ask the model what to do next, then - # parse its free-text answer into a structured action. - prompt = build_prompt(system_prompt, history, TOOL_SCHEMAS, goal) - raw = call_llm(prompt) - action = parse_thought(raw) + for step_num in range(len(history) + 1, max_steps + 1): + print(f"\n+--- STEP {step_num} {'-' * 44}") - thought = action.get('thought', '').strip() or '(no explicit thought)' + # REASON -------------------------------------------------- + raw = brain(system_prompt, history, goal) + action = parse_action(raw) + thought = (action.get('thought') or '(no explicit thought)').strip() print(f"| THOUGHT: {thought[:200]}") - # ACT + OBSERVE: carry out the action the model chose. After a tool - # call we append the result (the OBSERVATION) to history so the next - # REASON pass can see it. + # ACT + OBSERVE ------------------------------------------- if action['type'] == 'call_tool': - tool_name = action['tool'] - tool_in = action['input'] - print(f"| ACTION: call_tool -> {tool_name}") - print(f"| INPUT: {json.dumps(tool_in)[:200]}") - - if tool_name not in TOOLS: - obs = {"error": f"Unknown tool: {tool_name}"} - else: - try: - fn = TOOLS[tool_name] - # Call with kwargs from the JSON - obs = fn(**tool_in) if isinstance(tool_in, dict) else fn(tool_in) - except Exception as e: - obs = {"error": str(e)} - + print(f"| ACTION: call_tool -> {action['tool']}") + print(f"| INPUT: {json.dumps(action['input'])[:200]}") + obs = dispatch_tool(action['tool'], action['input']) print(f"| OBSERVE: {json.dumps(obs)[:300]}") history.append({ - 'thought': thought, - 'action_type': 'call_tool', - 'tool': tool_name, - 'input': tool_in, - 'observation': obs + 'thought': thought, 'action_type': 'call_tool', + 'tool': action['tool'], 'input': action['input'], 'observation': obs, }) - cache[employee_id] = history - save_cache(cache) + persist() elif action['type'] == 'ask_human': - question = action['question'] print(f"| ACTION: ask_human") - print(f"| QUESTION: {question}") - print(f"+{'-'*57}\n") + print(f"| QUESTION: {action['question']}") + print(f"+{'-' * 57}\n") try: human_input = input(" >>> HUMAN INPUT REQUIRED: ").strip() except (EOFError, OSError): human_input = "proceed" print(f" (non-interactive; defaulting to '{human_input}')") history.append({ - 'thought': thought, - 'action_type': 'ask_human', - 'question': question, - 'human_input': human_input + 'thought': thought, 'action_type': 'ask_human', + 'question': action['question'], 'human_input': human_input, }) - cache[employee_id] = history - save_cache(cache) + persist() elif action['type'] == 'complete': - summary = action['summary'] print(f"| ACTION: complete") - print(f"| SUMMARY: {summary}") - print(f"+{'-'*57}\n") - cache[employee_id] = history - save_cache(cache) - return { - 'status': 'COMPLETE', - 'summary': summary, - 'steps': step_num, - 'history': history - } - - else: - print(f"| ACTION: unknown - raw: {action.get('raw','')[:200]}") - cache[employee_id] = history - save_cache(cache) - return { - 'status': 'PARSE_ERROR', - 'raw': action.get('raw', ''), - 'steps': step_num, - 'history': history - } + print(f"| SUMMARY: {action['summary']}") + print(f"+{'-' * 57}\n") + persist() + return {'status': 'COMPLETE', 'summary': action['summary'], + 'steps': step_num, 'history': history} - print(f"+{'-'*57}") + else: # unknown / unparseable + print(f"| ACTION: unknown — raw: {action.get('raw', '')[:200]}") + persist() + return {'status': 'PARSE_ERROR', 'raw': action.get('raw', ''), + 'steps': step_num, 'history': history} - cache[employee_id] = history - save_cache(cache) - return { - 'status': 'MAX_STEPS_REACHED', - 'steps': max_steps, - 'history': history - } + print(f"+{'-' * 57}") + + persist() + return {'status': 'MAX_STEPS_REACHED', 'steps': max_steps, 'history': history} -# ── DRY RUN ────────────────────────────────────────────────────── +# ── DRY RUN ──────────────────────────────────────────────────────── def dry_run() -> None: - """Verify environment without making any LLM calls.""" - print("Onboarding Agent - dry run") + """Verify the environment without making any LLM calls.""" + print("Onboarding Agent — dry run") print("=" * 60) - # 1. Files for label, path in [ - ('system_prompt.md', PROMPT_PATH), - ('employees.json', EMPLOYEES_DB), - ('compliance_state.json', COMPLIANCE_DB), - ('tools/provision_access.py', os.path.join(TOOLS_DIR, 'provision_access.py')), + ('system_prompt.md', PROMPT_PATH), + ('data/employees.json', EMPLOYEES_DB), + ('data/compliance_state.json', COMPLIANCE_DB), + ('tools/provision_access.py', os.path.join(TOOLS_DIR, 'provision_access.py')), ('tools/send_notification.py', os.path.join(TOOLS_DIR, 'send_notification.py')), ]: - ok = os.path.exists(path) - print(f" [{'PASS' if ok else 'FAIL'}] {label}") + print(f" [{'PASS' if os.path.exists(path) else 'FAIL'}] {label}") - # 2. Tool registry print(f"\nTool registry: {len(TOOLS)} tools loaded") for name in TOOLS: print(f" - {name}") - # 3. Mock data smoke test print("\nSmoke test:") try: - profile = get_employee_profile("EMP-2026-0847") + get_employee_profile("EMP-2026-0847") print(" [OK] get_employee_profile('EMP-2026-0847') -> profile loaded") except Exception as e: - print(f" [FAIL] get_employee_profile failed: {e}") - + print(f" [FAIL] get_employee_profile: {e}") try: - compliance = check_compliance_status("EMP-2026-0847") - print(f" [OK] check_compliance_status('EMP-2026-0847') -> {compliance['overall_status']}") + c = check_compliance_status("EMP-2026-0847") + print(f" [OK] check_compliance_status('EMP-2026-0847') -> {c['overall_status']}") except Exception as e: - print(f" [FAIL] check_compliance_status failed: {e}") + print(f" [FAIL] check_compliance_status: {e}") - # 4. LLM key check key = os.environ.get('GROQ_API_KEY') print(f"\nLLM provider: Groq / llama-3.3-70b-versatile") - print(f" [{'SET' if key else 'UNSET'}] GROQ_API_KEY {'set' if key else 'NOT SET - agent will fail when run'}") + print(f" [{'SET' if key else 'UNSET'}] GROQ_API_KEY " + f"{'set' if key else 'NOT SET — agent will fail when run'}") print("\n" + "=" * 60) print("Dry run complete. To run the agent:") @@ -408,16 +569,33 @@ def dry_run() -> None: print(" python agent.py --employee EMP-2026-0847") -# ── CLI ────────────────────────────────────────────────────────── +# ── CLI ──────────────────────────────────────────────────────────── + +# NOTE: This exact string must match Enforcement rule 1 in system_prompt.md +# verbatim so --no-enforcement can strip it. If you edit the rule, edit here too. +COMPLIANCE_GATE_RULE = ( + "1. **Compliance gate** - Never call provision_access before\n" + " check_compliance_status returns overall_status: CLEARED. If status is\n" + " PENDING, send a reminder notification then ask_human for confirmation\n" + " before proceeding." +) + def main(): parser = argparse.ArgumentParser(description='Onboarding Agent') - parser.add_argument('--dry-run', action='store_true', help='Verify environment without calling the LLM') + parser.add_argument('--dry-run', action='store_true', + help='Verify the environment without calling the LLM') parser.add_argument('--employee', type=str, help='Employee ID to onboard') parser.add_argument('--no-enforcement', action='store_true', - help='Demo Step 6 - remove the compliance gate Enforcement rule') + help='Demo: remove the compliance gate Enforcement rule') + parser.add_argument('--offline', action='store_true', + help='Run with a deterministic mock brain — no API key required') + parser.add_argument('--reset', action='store_true', + help='Clear the cached session for this employee before running') args = parser.parse_args() + load_dotenv() # pick up GROQ_API_KEY from .env if present + if args.dry_run: dry_run() return @@ -426,25 +604,35 @@ def main(): parser.print_help() return - # Load system prompt + if args.reset: + cache = load_cache() + if cache.pop(args.employee, None) is not None: + save_cache(cache) + log(f"cleared cached session for {args.employee}") + with open(PROMPT_PATH) as f: system_prompt = f.read() - # Step 6 deliberate failure: strip the compliance gate rule if args.no_enforcement: - print("\n [!] Running with compliance gate REMOVED - demo mode") + print("\n [!] Running with compliance gate REMOVED — demo mode") system_prompt = system_prompt.replace( - "1. **Compliance gate** - Never call provision_access before\n check_compliance_status returns overall_status: CLEARED. If status is\n PENDING, send a reminder notification then ask_human for confirmation\n before proceeding.", - "1. ~~Compliance gate rule removed for this run~~" + COMPLIANCE_GATE_RULE, + "1. ~~Compliance gate rule removed for this run~~", ) - result = run_agent(args.employee, system_prompt) - print(f"\n+--- FINAL RESULT {'-'*41}") + if args.offline: + print("\n [i] Offline mode — deterministic mock brain, no API key used") + brain = make_offline_brain(args.employee, enforce=not args.no_enforcement) + else: + brain = live_brain + + result = run_agent(args.employee, system_prompt, brain=brain) + print(f"\n+--- FINAL RESULT {'-' * 41}") print(f"| Status: {result['status']}") print(f"| Steps: {result['steps']}") if result.get('summary'): print(f"| Summary: {result['summary']}") - print(f"+{'-'*57}") + print(f"+{'-' * 57}") if __name__ == '__main__': diff --git a/system_prompt.md b/system_prompt.md index 7f1f815..75d6471 100644 --- a/system_prompt.md +++ b/system_prompt.md @@ -1,35 +1,62 @@ # Meridian Onboarding Agent - System Prompt -> This file defines the behavior of the sample onboarding agent. Students -> can refine it during the workshop, but the core rules should remain clear -> and testable. +> This file defines the behavior of the sample onboarding agent. It is +> structured using the **RICE** principle — **R**ole, **I**ntent, +> **C**ontext, **E**nforcement. Students can refine it during the workshop, +> but the core rules must stay clear, explicit, and testable. ## Role -You are the onboarding coordinator for the Meridian Financial Services -engineering team in Hyderabad. You autonomously complete onboarding for new -hires while strictly enforcing compliance and security rules. +You are the **Onboarding Coordinator** for the Meridian Financial Services +engineering team in Hyderabad. You are an autonomous agent that reasons +step by step, decides when to call tools, and pauses for human confirmation +when policy requires it. You act on behalf of the company, so you treat every +compliance and security rule as non-negotiable — a shortcut here becomes an +audit finding. ## Intent -Given an employee_id, complete the four-step onboarding workflow: -1. Retrieve the employee profile -2. Check compliance acknowledgement status -3. Provision system access -4. Send a welcome notification +Given a single `employee_id`, complete the four-step onboarding workflow: -Output a summary when onboarding is complete. Escalate to the manager if -you cannot proceed. +1. **Retrieve** the employee profile from HR. +2. **Check** compliance acknowledgement status. +3. **Provision** system access — only after compliance is `CLEARED`. +4. **Notify** the employee with a welcome message. + +End the run with a short, factual summary of what was accomplished. If you +cannot proceed safely, escalate to the manager and stop rather than guess. ## Context -You have four tools available: -- get_employee_profile(employee_id) -- check_compliance_status(employee_id) -- provision_access(employee_id, access_profile) -- send_notification(recipient, channel, subject, body) +You operate a Reason → Act → Observe loop. On each step you produce one +thought and one action, observe the result, then reason again. + +You have four tools plus a human-in-the-loop escape hatch: + +- `get_employee_profile(employee_id)` — returns name, role, level, + department, manager_id, employment_type, access_profile. Raises + `EmployeeNotFound` if no record exists. +- `check_compliance_status(employee_id)` — returns the status of + code_of_conduct, data_handling_policy, security_guidelines, + posh_training, and an `overall_status` of `CLEARED` or `PENDING`. +- `provision_access(employee_id, access_profile)` — provisions GitHub, + JIRA, Confluence, and Slack based on the access profile. Returns + `SUCCESS` with the items provisioned. +- `send_notification(recipient, channel, subject, body)` — sends a message + via `email`, `slack`, or `whatsapp`. Returns `SUCCESS`. + +You may also `ask_human(question)` whenever human judgment or confirmation +is required by an enforcement rule. + +Reference workflow for a well-behaved run: -You may also ask_human(question) when human judgment is required. +1. Retrieve the employee profile. +2. Check compliance status. +3. If compliance is `PENDING`, send a reminder notification. +4. Ask the human for confirmation. +5. After confirmation, provision only the systems in `access_profile`. +6. Send a welcome notification. +7. Complete with a short summary. ## Enforcement @@ -46,7 +73,7 @@ between a deployable agent and an audit finding. send_notification and stop. Do not retry indefinitely. 3. **Access scope** — Never provision access to systems not listed in the - employee's access_profile. Do not infer additional access. + employee's access_profile. Do not infer or add extra access. 4. **PII in logs** - Never log the employee's email, name, or personal details. Use only the employee_id in any system log entry.