LangChain: agents in plain Python

def main():
    llm = ChatGroq(model=os.getenv("LLM_MODEL"),temperature=1)
    query = input(" Enter the question ")
    response = llm.invoke(query)
    print(response.content)

def main():
    agent = create_agent(model=os.environ["GEMINI_LLM_MODEL"])
    query = input("Ask your question: ")
    result = agent.invoke({"messages": [("user", query)]})
    print(result["messages"][-1].text)

import { test, expect } from '@playwright/test';

const URL = 'https://app.thetestingacademy.com/ai/blueprint/learn/langchain.html';

test('the agent calls the calculator tool for a pass rate', async ({ page }) => {
  await page.goto(URL);
  await page.getByTestId('lc-query').selectOption('pass-rate');
  await page.getByTestId('lc-tool-run').click();

  const messages = page.getByTestId('lc-messages').locator('li');
  await expect(messages).toHaveCount(4);
  await expect(messages.nth(1)).toContainText('qa_metric_calculator');
  await expect(page.getByTestId('lc-messages').locator('li[data-kind="ok"]')).toContainText('87.6');
});

test('a question without maths gets no tool call', async ({ page }) => {
  await page.goto(URL);
  await page.getByTestId('lc-query').selectOption('no-maths');
  await page.getByTestId('lc-tool-run').click();
  await expect(page.getByTestId('lc-messages').locator('li')).toHaveCount(2);
  await expect(page.getByTestId('lc-messages')).not.toContainText('tool_calls');
});

test('the tool refuses code instead of running it', async ({ page }) => {
  await page.goto(URL);
  await page.getByTestId('lc-tool-attack').click();
  await expect(page.getByTestId('lc-messages').locator('li[data-kind="bad"]'))
    .toContainText("Error: give me a plain arithmetic expression, e.g. '(438/500)*100'");
});

test('retrieval ranks the closest existing test case first', async ({ page }) => {
  await page.goto(URL);
  await page.getByTestId('lc-mode-rag').click();
  await page.getByTestId('lc-ticket').selectOption('VWO-114');
  await expect(page.getByTestId('lc-hits').locator('li').first()).toHaveAttribute('data-case', 'TTA-002');
  await expect(page.getByTestId('lc-rag-status')).toContainText('TTA-002 0.211');

  await page.getByTestId('lc-ticket').selectOption('VWO-49');
  await expect(page.getByTestId('lc-hits').locator('li').first()).toHaveAttribute('data-case', 'TTA-020');
});

test('the same tools with two models end differently', async ({ page }) => {
  await page.goto(URL);
  await page.getByTestId('lc-mode-models').click();
  await page.getByTestId('lc-model').selectOption('groq');
  await page.getByTestId('lc-replay-run').click();
  await expect(page.getByTestId('lc-verdict')).toContainText('Gave up after 3 steps');

  await page.getByTestId('lc-model').selectOption('deepseek');
  await page.getByTestId('lc-replay-run').click();
  await expect(page.getByTestId('lc-verdict')).toContainText('TEST PASSED');
  await expect(page.getByTestId('lc-trace').locator('li[data-tool="close_browser"]')).toHaveCount(1);
});

import ast
import operator

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.tools import tool          # create custom tools

load_dotenv()

# eval() on text a model produced is arbitrary code execution. This walks the
# parsed expression instead, so only arithmetic can ever run.
_OPS = {
    ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
    ast.Div: operator.truediv, ast.Pow: operator.pow, ast.USub: operator.neg,
}


def _safe_eval(node):
    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
        return node.value
    if isinstance(node, ast.BinOp) and type(node.op) in _OPS:
        return _OPS[type(node.op)](_safe_eval(node.left), _safe_eval(node.right))
    if isinstance(node, ast.UnaryOp) and type(node.op) in _OPS:
        return _OPS[type(node.op)](_safe_eval(node.operand))
    raise ValueError("only arithmetic is allowed")


@tool
def qa_metric_calculator(expression: str) -> str:
    """Calculate a QA metric from an arithmetic expression.

    Use this for pass rate, defect density, automation coverage, defect leakage
    or execution time. Input must be plain arithmetic with no words, for example
    '(438/500)*100' for a pass rate or '18/12' for defects per KLOC.
    """
    try:
        result = _safe_eval(ast.parse(expression, mode="eval").body)
        return f"{round(result, 2)}"
    except Exception:
        return "Error: give me a plain arithmetic expression, e.g. '(438/500)*100'"


agent = create_agent(
    model="google_genai:gemini-flash-lite-latest",
    tools=[qa_metric_calculator],          # list of available tools
    system_prompt=(
        "You are a QA metrics assistant for a test automation team. "
        "Use the qa_metric_calculator tool whenever a number must be computed. "
        "State the formula you used, then the result with its unit."
    ),
)

queries = [
    # maths -> the agent must call the tool
    "We ran 500 regression tests and 438 passed. What is the pass rate?",
    "A module of 12 KLOC has 18 defects. What is the defect density per KLOC?",
    # no maths -> the agent answers directly, no tool call
    "What is the difference between smoke testing and sanity testing?",
]

for q in queries:
    print(f"\nQuestion: {q}")
    result = agent.invoke({"messages": [{"role": "user", "content": q}]})
    print("Agent:", result["messages"][-1].text)

from typing import Literal
from dotenv import load_dotenv
from pydantic import BaseModel
from langchain.agents import create_agent

load_dotenv()

# 1) Define the response schema with Pydantic
class TestCase(BaseModel):
    title: str
    description: str
    steps: list[str]
    expected_result: str
    priority: Literal["High", "Medium", "Low"]
    tags: list[str]

class TestCaseList(BaseModel):
    # Tip: always wrap arrays inside an object (better compatibility, especially with Anthropic)
    test_cases: list[TestCase]

# 2) Create the agent with response_format
agent = create_agent(
    model="google_genai:gemini-flash-lite-latest",
    # model="anthropic:claude-haiku-4-5",     # optional
    system_prompt=(
        "You are a senior automation testing engineer. Always return test cases "
        "in the exact structured JSON format requested. Be detailed and professional."
    ),
    response_format=TestCaseList,             # enforce the schema on the final output
)

# 3) Provide a user story and invoke the agent
user_story = ("As a logged-in user, I want to add items to my shopping cart "
              "so that I can purchase them later.")

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": f"Generate 2 detailed test cases for this user story: {user_story}",
    }]
})

# 4) Work with the structured result: no manual parsing
print("=== Structured Output ===")
test_cases = result["structured_response"].test_cases
for tc in test_cases:
    print(tc.model_dump_json(indent=2))

print(f"\nSuccessfully generated {len(test_cases)} test cases!")

"""A very small local RAG over the TTACart test-case library.

No server, no database, no API key. 20 test cases in a JSON file, embedded once
with fastembed (ONNX, ~130MB model, downloads on first run) and cached to disk as
a .npy matrix. Retrieval is a cosine dot product over 20 rows, which is instant.

Why this is enough: RAG is retrieval + generation. At 20 documents a vector
"database" is a numpy array, and anything heavier is ceremony. Swap the search()
body for a Qdrant call when the corpus outgrows memory, the interface is the same.

If fastembed is missing the module degrades to keyword overlap scoring, so a
class demo still works offline with no install.
"""

import hashlib
import json
import re
from pathlib import Path

CORPUS_PATH = Path(__file__).parent / "rag_corpus" / "tta_testcases.json"
INDEX_PATH = Path(__file__).parent / "rag_corpus" / ".index.npz"
MODEL_NAME = "BAAI/bge-small-en-v1.5"      # 384 dims, ~130MB


def _document(case: dict) -> str:
    """What actually gets embedded. Title and tags carry most of the signal."""
    return (f"{case['title']}. Area: {case['area']}. Type: {case['type']} test case. "
            f"Tags: {', '.join(case['tags'])}. "
            f"Steps: {' '.join(case['steps'])} Expected: {case['expected']}")


class TestCaseRAG:
    def __init__(self, corpus_path: Path = CORPUS_PATH):
        self.cases: list[dict] = json.loads(corpus_path.read_text())
        self.docs = [_document(c) for c in self.cases]
        self._vectors = None
        self.backend = "keyword"
        self._try_load_vectors()

    # ---------------------------------------------------------------- index
    def _corpus_fingerprint(self) -> str:
        return hashlib.sha256("\n".join(self.docs).encode()).hexdigest()[:16]

    def _try_load_vectors(self) -> None:
        try:
            import numpy as np
            from fastembed import TextEmbedding
        except ImportError:
            return                                    # stay on keyword scoring

        fp = self._corpus_fingerprint()
        if INDEX_PATH.exists():
            cached = np.load(INDEX_PATH, allow_pickle=False)
            if str(cached["fingerprint"]) == fp:      # corpus unchanged -> reuse
                self._vectors, self.backend = cached["vectors"], f"embeddings ({MODEL_NAME}, cached)"
                return

        model = TextEmbedding(model_name=MODEL_NAME)
        vectors = np.array(list(model.embed(self.docs)), dtype="float32")
        vectors /= np.linalg.norm(vectors, axis=1, keepdims=True)   # unit -> dot == cosine
        np.savez(INDEX_PATH, vectors=vectors, fingerprint=fp)
        self._vectors, self.backend = vectors, f"embeddings ({MODEL_NAME}, built)"

    # --------------------------------------------------------------- search
    def search(self, query: str, k: int = 5) -> list[tuple[float, dict]]:
        """Return the k most similar test cases as (score, case), best first."""
        if self._vectors is None:
            return self._keyword_search(query, k)
        import numpy as np
        from fastembed import TextEmbedding
        q = np.array(next(iter(TextEmbedding(model_name=MODEL_NAME).embed([query]))), dtype="float32")
        q /= np.linalg.norm(q)
        scores = self._vectors @ q
        order = np.argsort(-scores)[:k]
        return [(float(scores[i]), self.cases[i]) for i in order]

    def _keyword_search(self, query: str, k: int) -> list[tuple[float, dict]]:
        """Jaccard overlap fallback. Crude, but it never fails to install."""
        def toks(s: str) -> set[str]:
            return {w for w in re.findall(r"[a-z0-9]+", s.lower()) if len(w) > 2}
        q = toks(query)
        scored = [(len(q & toks(d)) / max(len(q | toks(d)), 1), c)
                  for d, c in zip(self.docs, self.cases)]
        return sorted(scored, key=lambda x: -x[0])[:k]


def format_for_prompt(hits: list[tuple[float, dict]]) -> str:
    """Render retrieved cases as context the planner can cite by id."""
    if not hits:
        return "No existing test cases matched."
    out = []
    for score, c in hits:
        out.append(
            f"[{c['id']}] ({c['type'].upper()}, {c['priority']}, {c['area']}, "
            f"{c['status']}, last run: {c['last_result']}, similarity {score:.3f})\n"
            f"  {c['title']}\n"
            f"  Expected: {c['expected']}\n"
            f"  Tags: {', '.join(c['tags'])}")
    return "\n".join(out)


if __name__ == "__main__":
    rag = TestCaseRAG()
    print(f"corpus: {len(rag.cases)} cases | backend: {rag.backend}\n")
    for q in ["SSO and passkey login buttons",
              "wrong password shows an error",
              "tax and total calculation at checkout"]:
        print(f"QUERY: {q}")
        for s, c in rag.search(q, 3):
            print(f"   {s:.3f}  {c['id']}  {c['title']}")
        print()

# ---------------------------------------------------------------- stage 2
class TestCase(BaseModel):
    id: str = Field(description="e.g. TC-01")
    title: str
    priority: Literal["P0", "P1", "P2"]
    steps: list[str] = Field(description="Concrete UI steps against the target app")
    expected_result: str
    automatable: bool = Field(
        description="False when the ticket describes something absent from the target app")
    reason_if_not: str = Field(default="", description="Why it cannot be automated here")


class TestPlan(BaseModel):
    ticket_key: str
    scope: str = Field(description="2-3 sentences: what is tested and what is not")
    risks: list[str]
    test_cases: list[TestCase]


PLANNER_PROMPT = f"""You are a senior QA engineer writing a test plan from a Jira ticket.

The test cases will be executed by a browser agent against THIS app, which may be
a different product from the one the ticket describes:

    Target app: {TARGET_URL}
    Login:      {TARGET_USER} / {TARGET_PASS}

Rules:
1. Write 5-8 test cases covering the ticket's acceptance criteria.
2. Set automatable=true ONLY when every step can be performed on the target app
   above. If the ticket describes a feature the target app does not have (for
   example SSO or passkey buttons that are not on this login page), set
   automatable=false and say so in reason_if_not. Do NOT invent UI.
3. Steps must be concrete and clickable: name the field, button or text to check.
4. Never invent credentials beyond the ones given."""


# ---------------------------------------------------------------- stage 3
EXECUTOR_PROMPT = """You are an end-to-end automation testing agent driving a real
browser with Playwright.

Rules:
1. Call launch_browser FIRST.
2. Call snapshot_page on every new page instead of guessing selectors.
3. Prefer click_by_role and type_by_label over raw CSS.
4. Verify with assert_visible / assert_text_contains. Never assume a step worked.
5. If a case's element genuinely does not exist, report it FAILED with what you
   saw. Do not pretend it passed and do not substitute a different element.
6. Before closing, call get_console_errors and take_screenshot.
7. Report each test case id with PASS or FAIL, then a final summary line."""