Shared AI Code Patterns & Blocks

import json
from typing import Dict, Any, Callable

def run_agent_loop(
    client: Any,
    model: str,
    messages: list,
    tools: list,
    tool_map: Dict[str, Callable],
    max_turns: int = 6
) -> str:
    """
    Executes an autonomous tool-calling reasoning loop until the model
    produces a terminal response or reaches max_turns.
    """
    for turn in range(max_turns):
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        msg = response.choices[0].message
        messages.append(msg)

        # Terminal state reached if no tool calls requested
        if not msg.tool_calls:
            return msg.content or ""

        # Dispatch each requested tool execution
        for tool_call in msg.tool_calls:
            fn_name = tool_call.function.name
            try:
                fn_args = json.loads(tool_call.function.arguments)
                if fn_name in tool_map:
                    result = tool_map[fn_name](**fn_args)
                else:
                    result = {"error": f"Tool '{fn_name}' not registered."}
            except Exception as e:
                result = {"error": f"Tool execution failed: {str(e)}"}

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result)
            })

    return "Agent execution stopped: Maximum conversation turns reached."

from typing import List, Dict, Any

def reciprocal_rank_fusion(
    dense_results: List[Dict[str, Any]],
    sparse_results: List[Dict[str, Any]],
    k: int = 60
) -> List[Dict[str, Any]]:
    """
    Fuses dense vector results and sparse BM25 keyword rankings using RRF.
    Score = sum(1.0 / (k + rank)) for each document appearing in candidate lists.
    """
    rrf_scores: Dict[str, float] = {}
    doc_lookup: Dict[str, Dict[str, Any]] = {}

    # Score vector rank positions (1-indexed)
    for rank, doc in enumerate(dense_results, start=1):
        doc_id = str(doc["id"])
        doc_lookup[doc_id] = doc
        rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0) + (1.0 / (k + rank))

    # Score BM25 keyword rank positions
    for rank, doc in enumerate(sparse_results, start=1):
        doc_id = str(doc["id"])
        doc_lookup[doc_id] = doc
        rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0) + (1.0 / (k + rank))

    # Sort documents by combined RRF score descending
    sorted_ranked = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
    return [{"doc": doc_lookup[doc_id], "rrf_score": round(score, 5)} for doc_id, score in sorted_ranked]

from mcp.server.fastmcp import FastMCP
from typing import Dict, Any

# Initialize FastMCP application instance
mcp = FastMCP("UtilityHub-Agent-Tools")

@mcp.tool()
def search_knowledge_base(query: str, limit: int = 5) -> Dict[str, Any]:
    """Search internal documentation chunks for semantic query matches."""
    return {
        "query": query,
        "results_count": limit,
        "status": "ready"
    }

@mcp.tool()
def fetch_system_metrics() -> Dict[str, Any]:
    """Retrieve operational telemetry and memory state."""
    return {"status": "healthy", "uptime_pct": 99.98}

if __name__ == "__main__":
    # Start stdio transport server
    mcp.run(transport="stdio")

import asyncio
import websockets
from typing import AsyncGenerator

async def audio_stream_handler(websocket, path):
    """
    Receives binary audio PCM frames from client microphone,
    routes to streaming STT, and yields synthesized voice chunks.
    """
    print("Voice client connected.")
    try:
        async for message in websocket:
            if isinstance(message, bytes):
                # Process audio chunk (e.g. VAD chunking + Deepgram streaming)
                pass
            elif isinstance(message, str):
                await websocket.send('{"status": "listening"}')
    except websockets.exceptions.ConnectionClosed:
        print("Voice client disconnected gracefully.")