notebooks: Useful snippets (markdown / python / R)

notebooks-python-quarto-useful-snippets.md

Some useful snippets for notebooks, Python/R/HTML/CSS code in Quarto/Jupyter notebooks

dotenv: Load environment variables from a .env file in a notebook cell:

import os
from openai import OpenAI
from pprint import pprint
from IPython.display import display, Markdown

#import tiktoken
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv()) # read local .env file

assert os.environ.get('OPENAI_API_KEY'), "API key not found. Make sure to set OPENAI_API_KEY in your .env file."
openai = OpenAI(api_key  = os.environ['OPENAI_API_KEY'])

GUI: Highlight a note or tip in a notebook cell:

# Highlight a note or tip in a notebook cell:
<p style="padding:12px; border:1px solid #ffc107; border-radius:6px">
>⚡ Note: ...
</p>

Tool Calling: Disk-Based Memoization (Essential for Notebooks)

Prevents re-running expensive calls when iterating. Use diskcache—it's pure-Python, zero-config, and persists across notebook restarts.

from diskcache import Cache
import hashlib

cache = Cache("/tmp/tool_cache")  # Simple file-based store

def cached_tool_call(func, **kwargs):
    key = hashlib.md5(f"{func.__name__}:{sorted(kwargs.items())}".encode()).hexdigest()
    if key in cache:
        return cache[key]
    result = func(**kwargs)
    cache[key] = result
    return result

# Wrap your actual tool calls
def fetch_weather(city: str) -> dict:
    return cached_tool_call(_real_fetch_weather, city=city)

Tool Calling: Self-Correction Loop with Structured Errors

Instead of just max_turns, give the LLM clear error signals. This fixes ~80% of schema hallucinations without human intervention.

def run_with_retry(client, messages, tools, max_turns=5):
    for turn in range(max_turns):
        response = client.chat.completions.create(
            messages=messages, tools=tools, tool_choice="auto"
        )
        
        if not response.choices[0].message.tool_calls:
            return response  # Done
        
        for call in response.choices[0].message.tool_calls:
            try:
                # Validate & execute
                args = json.loads(call.function.arguments)
                result = execute_tool(call.function.name, args)  # Your tool router
                messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
            except Exception as e:
                # Feed error back to LLM
                messages.append({
                    "role": "tool", 
                    "tool_call_id": call.id, 
                    "content": f"ERROR: {type(e).__name__}: {e}. Try fixing arguments."
                })
                break  # Re-submit to LLM immediately
    return response

tool-calling-kimi-k2-thinking-turbo.md

Tool Calling with Kimi-K2-Thinking-Turbo: A Minimalist Guide for Solo Devs

For your solo-dev, notebook-to-cron workflow, you don't need heavy orchestration. The classic max_turns loop is still your skeleton—just add a few lightweight guardrails to make it robust and idempotent. Here are the patterns that actually matter for small scripts:

---

1. Disk-Based Memoization (Essential for Notebooks)

Prevents re-running expensive calls when iterating. Use diskcache—it's pure-Python, zero-config, and persists across notebook restarts.

from diskcache import Cache
import hashlib

cache = Cache("/tmp/tool_cache")  # Simple file-based store

def cached_tool_call(func, **kwargs):
    key = hashlib.md5(f"{func.__name__}:{sorted(kwargs.items())}".encode()).hexdigest()
    if key in cache:
        return cache[key]
    result = func(**kwargs)
    cache[key] = result
    return result

# Wrap your actual tool calls
def fetch_weather(city: str) -> dict:
    return cached_tool_call(_real_fetch_weather, city=city)

---

2. Self-Correction Loop with Structured Errors

Instead of just max_turns, give the LLM clear error signals. This fixes ~80% of schema hallucinations without human intervention.

def run_with_retry(client, messages, tools, max_turns=5):
    for turn in range(max_turns):
        response = client.chat.completions.create(
            messages=messages, tools=tools, tool_choice="auto"
        )

        if not response.choices[0].message.tool_calls:
            return response  # Done

        for call in response.choices[0].message.tool_calls:
            try:
                # Validate & execute
                args = json.loads(call.function.arguments)
                result = execute_tool(call.function.name, args)  # Your tool router
                messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
            except Exception as e:
                # Feed error back to LLM
                messages.append({
                    "role": "tool",
                    "tool_call_id": call.id,
                    "content": f"ERROR: {type(e).__name__}: {e}. Try fixing arguments."
                })
                break  # Re-submit to LLM immediately
    return response

---

3. Idempotency for Cron Safety

Cron jobs re-run on failure. Use a deterministic run ID (e.g., date-based) to make tool calls idempotent.

import os
from datetime import date

RUN_ID = os.getenv("RUN_ID", date.today().isoformat())  # e.g., "2025-11-06"

def write_to_file_tool(filename: str, content: str):
    # Make it idempotent: include RUN_ID in filename or content hash
    safe_name = f"{RUN_ID}_{filename}"
    if os.path.exists(safe_name):
        return f"Skipped: {safe_name} already exists"
    with open(safe_name, "w") as f:
        f.write(content)
    return f"Wrote {len(content)} bytes to {safe_name}"

---

4. Lightweight State Persistence for Resume

For long-running cron jobs, checkpoint state to disk. If interrupted, resume without re-doing everything.

import json

STATE_FILE = "/tmp/agent_state.json"

def load_state():
    try:
        with open(STATE_FILE) as f:
            return json.load(f)
    except FileNotFoundError:
        return {"messages": [], "completed_steps": []}

def save_state(messages, completed_steps):
    with open(STATE_FILE, "w") as f:
        json.dump({"messages": messages, "completed_steps": completed_steps}, f)

# In your loop:
state = load_state()
# ... run one step ...
state["completed_steps"].append(step_name)
save_state(messages, state["completed_steps"])

---

5. Convert Notebook to Script (One-Liner)

Strip markdown cells, keep code cells. No need for complex exporters.

# save_as_script.py
import nbformat, sys

nb = nbformat.read(sys.argv[1], 4)
code_cells = [c.source for c in nb.cells if c.cell_type == "code"]
print("\n\n".join(code_cells))

Usage: python save_as_script.py my_notebook.ipynb > my_script.py

---

🎯 Complete Minimal Cron-Ready Template

Putting it together: a self-contained script that logs to file and respects a conservative turn limit.

import openai, os, json, logging
from datetime import date

logging.basicConfig(filename="/tmp/agent.log", level=logging.INFO)
RUN_ID = os.getenv("RUN_ID", date.today().isoformat())
MAX_TURNS = 3  # Be strict for cron

def main():
    client = openai.Client()
    messages = [{"role": "user", "content": "Your task here"}]
    tools = [...]  # Your tool definitions

    for turn in range(MAX_TURNS):
        logging.info(f"Turn {turn+1}")
        response = client.chat.completions.create(messages=messages, tools=tools)

        if not response.choices[0].message.tool_calls:
            logging.info("Done: " + response.choices[0].message.content)
            break

        for call in response.choices[0].message.tool_calls:
            try:
                args = json.loads(call.function.arguments)
                result = execute_tool(call.function.name, args)  # Your router
                messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
            except Exception as e:
                logging.error(f"Tool error: {e}")
                messages.append({
                    "role": "tool",
                    "tool_call_id": call.id,
                    "content": f"ERROR: {e}"
                })
    else:
        logging.warning("Max turns reached")

if __name__ == "__main__":
    main()

---

📌 Cron Setup Tips

• Lock file: Prevent overlapping runs: flock -n /tmp/myagent.lock python my_script.py
• Env vars: Store API keys in crontab: 0 9 * * * OPENAI_API_KEY=... python /path/to/my_script.py
• Failure alerts: Cron only emails on stderr; use logging for diagnostics, print() for final output.

These patterns add ~30 lines of boilerplate but make your tiny scripts reliable enough to run unsupervised.

tool-calling-qwen36-27b.md

qwen3.6-27b tool calling best practices for cron jobs and notebooks

Given your workflow (notebook → tiny projects → cron), you don't need complex multi-agent or streaming architectures. You need deterministic termination, minimal boilerplate, and clean batch execution. The best fit is a Guardrailed Tool Loop with explicit exit conditions, not just n turns.

Here’s a practical, production-ready approach tailored to your context:

---

🔑 Core Pattern: Guardrailed Tool Loop

Instead of a naive for _ in range(max_turns):, use a loop that exits on any of these conditions:

Condition	Why it matters for cron/notebooks
max_turns reached	Hard cap prevents infinite loops
max_runtime exceeded	Cron jobs shouldn't hang forever
LLM returns no tool calls	Normal completion
tool fails N times consecutively	Prevents cascading API/network failures
idempotency check passes	Skip if job already ran successfully today
dry_run=True	Test logic without executing tools

---

🛠️ Key Techniques for Your Workflow

1. Stateless, Export-Ready Tool Functions

Avoid notebook globals or mutable state. Tools should be pure functions that take explicit args and return structured dicts:

def get_weather(city: str) -> dict:
    # Simulate API call
    return {"city": city, "temp_c": 22, "status": "ok"}

def calculate_bill(items: list[dict]) -> dict:
    total = sum(item["price"] * item["qty"] for item in items)
    return {"total": total, "tax": total * 0.1}

2. Structured Conversation History

Keep a simple list of messages. Easy to dump for debugging or convert to script later:

history = [
    {"role": "system", "content": "You are a helpful assistant with access to tools."},
    {"role": "user", "content": "What's the weather in Berlin and how much is 3x coffee at $2.50?"}
]

3. Graceful Tool Failure Handling

Never let a tool crash the loop. Wrap execution, return clear error strings to the LLM:

def execute_tool(name: str, args: dict) -> str:
    try:
        tool = TOOLS[name]
        result = tool(**args)
        return str(result)
    except Exception as e:
        return f"Tool '{name}' failed: {str(e)}"

Why? The LLM can adapt ("retry with different args" or "skip this step") instead of your script dying.

4. Cron-Ready Exit Codes & Logging

Replace print() with logging. Return explicit exit codes:

import logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")

def run_agent_loop(...):
    # ... loop logic ...
    logging.info("Agent loop completed successfully")
    return 0  # Cron treats 0 as success

5. Environment-Driven Config

Never hardcode API keys or limits. Use python-dotenv or os.environ:

import os
from dotenv import load_dotenv
load_dotenv()

MAX_TURNS = int(os.getenv("MAX_TURNS", "5"))
MAX_RUNTIME_S = int(os.getenv("MAX_RUNTIME_S", "60"))
API_KEY = os.getenv("OPENAI_API_KEY")

---

📦 Minimal Template (Notebook → Cron Ready)

# config.py or top of notebook
import os, time, json, logging
from openai import OpenAI

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

TOOLS = {
    "get_weather": lambda city: {"city": city, "temp_c": 22, "status": "ok"},
    "calculate_bill": lambda items: {"total": sum(i["price"]*i["qty"] for i in items)},
}

def run_loop(user_prompt: str, max_turns: int = 5, max_runtime_s: int = 60) -> str:
    history = [{"role": "system", "content": "Use tools when needed. Return plain text when done."},
               {"role": "user", "content": user_prompt}]
    start = time.time()

    for turn in range(max_turns):
        if time.time() - start > max_runtime_s:
            logging.warning("Timeout reached")
            break

        resp = client.chat.completions.create(
            model="gpt-4o", messages=history, tools=[...], tool_choice="auto"
        )
        msg = resp.choices[0].message
        history.append(msg)

        if not msg.tool_calls:
            return msg.content

        for tc in msg.tool_calls:
            result = execute_tool(tc.function.name, json.loads(tc.function.arguments))
            history.append({"role": "tool", "tool_call_id": tc.id, "content": result})

    return "Loop terminated (max turns/runtime). Last LLM message: " + msg.content

Note: Replace [...] with proper OpenAI tool schema. This is intentionally minimal.

---

🔄 Notebook → Script Transition Checklist

Notebook Habit	Cron-Ready Fix
%pip install, !wget	Move to requirements.txt + pip install in cron or venv
print() for debugging	Use logging with file handler
Mutable globals across cells	Wrap logic in functions, pass explicit args
Interactive input() prompts	Add dry_run or --batch flag
Hidden state in _ip or %store	Explicit JSON/file state or keep stateless

Export command:

jupyter nbconvert --to script my_project.ipynb
# Fix any magic commands, then:
python my_project.py

---

⏱️ Cron-Specific Best Practices

1. Prevent overlapping runs: Use flock or a PID/temp file:

   0 2 * * * /usr/bin/flock -n /tmp/my_agent.lock /path/to/venv/bin/python /path/to/script.py >> /var/log/my_agent.log 2>&1

2. Set explicit timeouts: Cron doesn't kill long jobs automatically. Use timeout 300 python script.py in crontab.
3. Idempotency: Check a marker file or DB flag before running:

   if os.path.exists("last_run_success.json"):
       logging.info("Already ran today. Skipping.")
       exit(0)

4. Network resilience: Add retry logic for external APIs (tools, LLM calls). Use tenacity or simple try/except with backoff.

---

🚫 Common Pitfalls & Fixes

Pitfall	Fix
LLM loops calling same tool with same args	Add max_consecutive_same_tool check or require arg mutation
Tool output too long → context overflow	Truncate tool results, return only essential fields
Cron fails silently	Always redirect stdout/stderr to log files, check exit codes
Notebook state leaks into script	Run python script.py early in dev; don't wait until cron
LLM ignores tool results	Add system prompt: "You must acknowledge tool output before proceeding."

---

✅ Recommendation Summary

1. Start with a guardrailed tool loop (max turns + timeout + error backoff).
2. Keep tools stateless & pure functions.
3. Use logging + exit codes, not print().
4. Design for cron from day 1: env vars, idempotency, flock, timeouts.
5. Convert notebooks early using nbconvert, fix magic cells, test as scripts.
6. Add dry_run=True flag for safe iteration.

This pattern scales from interactive notebook exploration to unattended cron jobs with minimal refactoring. If you want, I can provide a complete, runnable template with proper OpenAI tool schemas, logging, and cron-safe structure. Just say the word.