标签：#python

GitHub热门项目 | Open-Source Frontier Voice AI | Stars: 48,317 | 219 stars today | 语言: Python

GitHub热门项目 | Robust Speech Recognition via Large-Scale Weak Supervision | Stars: 101,694 | 116 stars today | 语言: Python

Quark’s Outlines: Python User-Defined Methods Overview, Historical Timeline, Problems & Solutions An Overview of Python User-Defined Methods What is a Python user-defined method? When you define a function inside a class, Python does not treat it as just a function. When you call it from an instance, Python changes it into a method. This method knows which object it was called from. It adds that object as the first argument when the function runs. A Python user-defined method joins a function, a class, and a class instance (or None ). It is created when you get a function from a class or an instance. Python binds the instance to the function and forms a method. Python lets you bind a class function to an instance as a method. class Box : def show ( self , word ): print ( " Box says: " , word ) x = Box () x . show ( " hi " ) # prints: # Box says: hi The method x.show is bound to the instance x . Python passes x as the first argument. What does "bound method" mean in Python? When a method is bound, it remembers the instance that called it. A bound method is created when you get a method from an object. It holds a reference to both the function and the instance. Python will pass the instance automatically when you call the method. If you get the same method from the class, Python gives you an unbound method. That means the function is not tied to any one object. Python uses bound methods to remember which object to call with. class Lamp : def turn_on ( self ): print ( " The lamp is now on. " ) l = Lamp () m = Lamp () a = l . turn_on b = m . turn_on a () b () # prints: # The lamp is now on. # The lamp is now on. Each bound method remembers which Lamp it came from. A Historical Timeline of Python User-Defined Methods Where do Python user-defined methods come from? Python user-defined methods grew from early ideas in object-oriented design. In many languages, methods are just functions that get special treatment when called from an object. Python made this clear by lettin

Published on : 2026-06-06 Reading time : 6 min Tags : #python #async #performance #optimization The Problem: Fake Async The supabase-async library claimed to be async but actually wrapped synchronous calls with ThreadPoolExecutor: # ❌ Fake async (old code) class SupabaseAsync : def __init__ ( self ): self . _executor = ThreadPoolExecutor ( max_workers = 3 ) async def select ( self , table : str ): loop = asyncio . get_event_loop () r = await loop . run_in_executor ( self . _executor , lambda : requests . get ( url ) # Sync call wrapped as async ) return r . json () Problems : Max 3 concurrent requests (not scalable) Thread overhead per request High memory usage No connection pooling Solution: httpx AsyncClient Use true async HTTP with httpx: # ✅ Real async (new code) import httpx class SupabaseAsync : def __init__ ( self ): self . _client : Optional [ httpx . AsyncClient ] = None async def _get_client ( self ) -> httpx . AsyncClient : if self . _client is None : self . _client = httpx . AsyncClient ( headers = self . _headers , timeout = 30 , limits = httpx . Limits ( max_connections = 10 ) ) return self . _client async def select ( self , table : str ): client = await self . _get_client () r = await client . get ( f " { self . _base } / { table } " ) r . raise_for_status () return r . json () Performance Gains Metric ThreadPoolExecutor(3) httpx(10) Max concurrent 3 requests 10 requests Avg response 450ms 150ms Memory usage 250MB 180MB Throughput 6.7 req/s 20 req/s Real benchmark : 100 concurrent requests ThreadPoolExecutor: 15 seconds httpx AsyncClient: 5 seconds 3x faster ⚡ Migration Steps 1. Client Initialization with Lazy Loading async def _get_client ( self ) -> httpx . AsyncClient : if self . _client is None : self . _client = httpx . AsyncClient ( headers = self . _headers , timeout = 30 , limits = httpx . Limits ( max_connections = 10 , max_keepalive_connections = 5 ) ) return self . _client 2. HTTP Methods (GET, POST, etc.) async def _request ( self , metho

Published on : 2026-06-06 Reading time : 6 min Tags : #python #async #performance #optimization 문제: 거짓 비동기 supabase-async 라이브러리는 이름은 async이지만, 실제로는 ThreadPoolExecutor로 동기 호출을 래핑하고 있었습니다. # ❌ 거짓 비동기 (기존 코드) class SupabaseAsync : def __init__ ( self ): self . _executor = ThreadPoolExecutor ( max_workers = 3 ) async def select ( self , table : str ): loop = asyncio . get_event_loop () r = await loop . run_in_executor ( self . _executor , lambda : requests . get ( url ) # 동기 호출을 async로 포장 ) return r . json () 문제점 : 최대 3개 동시 요청만 가능 (동시성 부족) 스레드 오버헤드 (각 요청마다 스레드 생성) 높은 메모리 사용량 해결책: httpx AsyncClient 진정한 비동기 HTTP 클라이언트인 httpx를 사용합니다. # ✅ 진정한 비동기 (수정된 코드) import httpx class SupabaseAsync : def __init__ ( self ): self . _client : Optional [ httpx . AsyncClient ] = None async def _get_client ( self ) -> httpx . AsyncClient : if self . _client is None : self . _client = httpx . AsyncClient ( headers = self . _headers , timeout = 30 , limits = httpx . Limits ( max_connections = 10 ) ) return self . _client async def select ( self , table : str ): client = await self . _get_client () r = await client . get ( f " { self . _base } / { table } " ) r . raise_for_status () return r . json () 성능 개선 동시성 비교 지표 ThreadPoolExecutor(3) httpx(10) 최대 동시 요청 3개 10개 평균 응답 시간 450ms 150ms 메모리 사용량 250MB 180MB 초당 처리량 6.7 req/s 20 req/s 벤치마크 # 100개 동시 요청 처리 시간 ThreadPoolExecutor : 15 초 httpx AsyncClient : 5 초 → 3 배 빠름 마이그레이션 단계 1. 클라이언트 초기화 async def _get_client ( self ) -> httpx . AsyncClient : if self . _client is None : self . _client = httpx . AsyncClient ( headers = self . _headers , timeout = 30 , limits = httpx . Limits ( max_connections = 10 , max_keepalive_connections = 5 ) ) return self . _client 2. 요청 메서드 async def _request ( self , method : str , url : str , ** kwargs ): client = await self . _get_client () if method == " GET " : return await client . get ( url , ** kwargs ) elif method == " POST " : return await client . post ( url , ** kwargs ) # ... 3. Context Manager 지원 async def clos

Published on : 2026-06-06 Reading time : 8 min Tags : #security #python #audit #devops 개요 3개월에 걸쳐 개발한 6개 Python 프로젝트(3개 봇 + 3개 라이브러리)를 종합 감사했습니다. 25개 보안/코드 이슈를 발견했고, 23개를 즉시 수정했습니다. 감사 대상 : FastAPI + Telegram Bot + LLM 통합 시스템 총 파일 : 91개 Python 파일 발견 이슈 : 25개 (심각 5개, 중간 18개, 경미 2개) 수정율 : 92% (23/25) 심각도 높음 - 5개 이슈 1. API 키가 Git 히스토리에 노출됨 🔴 문제 : Anthropic, Supabase, Telegram API 키가 .env 파일로 커밋됨 # ❌ 노출된 상태 (git log에서 확인 가능) ANTHROPIC_API_KEY = sk - ant - api03 - xxxxxxxxxx SUPABASE_KEY = sb_publishable_xxxxxxxxxx 위험도 : 누구든 이전 커밋으로 API 키 접근 가능 → 리소스 도용, 데이터 침해 해결책 : # 1. BFG로 히스토리 정리 bfg --delete-files ".env" --no-blob-protection . # 2. Git에서 제거 git rm --cached .env echo ".env" >> .gitignore # 3. API 키 로테이션 (필수) # - Anthropic: console.anthropic.com/account/keys # - Supabase: app.supabase.com → Settings → API # - Telegram: @BotFather → /token 2. SSL 검증 비활성화 (MITM 공격 위험) 🔴 문제 : requests 호출에 verify=False 사용 (10곳) # ❌ 위험한 코드 response = requests . get ( url , verify = False ) # ✅ 안전한 코드 response = requests . get ( url , verify = True ) # 기본값 영향 : HTTPS 중간자 공격(MITM) 가능 → 민감한 데이터 도청 수정 : contest-agent, supabase-async 전체 10곳 제거 3. 광범위한 예외 처리 🔴 문제 : except Exception 으로 모든 오류를 무시 (114곳) # ❌ 버그 추적 불가 try : result = await db_select ( " contests " ) except Exception : print ( " failed " ) # 어떤 오류인지 알 수 없음 # ✅ 구체적인 처리 try : result = await db_select ( " contests " ) except requests . HTTPError as e : logger . error ( f " DB error: { e } " , exc_info = True ) raise 영향 : 버그 원인 파악 불가 → 프로덕션 문제 대응 시간 증가 4. 라이브러리 __init__.py 부실 문제 : llm-router, supabase-async, telegram-agent의 __init__.py 비어있음 # ❌ 기존 (빈 파일) # __init__.py # (아무것도 없음) # ✅ 수정 후 from llm_router import LLMRouter __version__ = " 0.1.0 " __all__ = [ " LLMRouter " ] 영향 : PyPI 설치 후 import 실패 from llm_router import LLMRouter # ❌ ImportError 5. 문법 오류 (try-except 들여쓰기) ai-insight-curator의 processor.py에서 DB 작업이 try 블록 밖에 있었음 → 예외 처리 안 됨 심각도 중간 - 18개 이슈 의존성 버전 불일치 Anthropic: 0.25.0 / 0.34.0 혼재 → 0.34.0으로 통일 Supabase: 2.0.0 / 2.4.0 혼재 → 2

Six months ago, I sat down with a YouTube playlist, a blank notebook, and one goal: learn Python. What I did not expect was how hard it would be, not the Python itself, but figuring out how to actually learn it. I started with a YouTube playlist. Simple enough. Except nobody tells you what to do after you watch a video. Do you rewatch it? Take notes? Jump straight to code? I had no system. I'd watch a concept, feel like I understood it, open VS Code, and stare at a blank file. That's when I realized I had fallen into passive learning. And passive learning in the age of AI is a particularly dangerous trap, because it's so easy to confuse activity with progress. I could watch a video, feel good. I could ask Claude to explain a concept, feel good. I could even ask AI to write code, read it, nod along, and feel like I'd learned something. I hadn't. I'd just consumed. There's a difference. The real moment of honesty came when I was stuck on a coding problem. My instinct, everyone's instinct now is to open ChatGPT or Claude immediately. And I knew, sitting there with the cursor blinking, that if I did that every single time I got stuck, I was building nothing. My brain would never develop the muscle of working through problems. I would be someone who can prompt AI to code, not someone who can think in code. And in a world where AI can already write decent code, the person who can't think independently isn't valuable. They're replaceable. So I had to build a system that forced me to actually learn. After a lot of trial and failure, I landed on a 5-phase checklist that I wrote out by hand and kept next to my laptop. Phase 1: is what I call First Contact — watch one focused video, then write a summary purely from memory, then discuss it with an LLM not to get answers but to pressure-test what I thought I understood. Phase 2: is Deep Understanding — read a written source, write proper notes, map the concept visually, and list every edge case and exception I can find. Phase 3:

In the previous article, we explored how to build our first CRUD API using FastAPI. While our API worked correctly, there was one major problem. We were storing data inside Python lists, which exist only in memory. If you've ever wondered how applications like Instagram, LinkedIn, or ChatGPT remember information even after a server restart, the answer is simple: databases. In this article, we'll solve the problem of in-memory storage by connecting our FastAPI application to SQLite using SQLAlchemy. If you haven't read the previous post, check it out: FastAPI for AI Engineers - Part 2: Building Your First CRUD API Ananya S Ananya S Ananya S Follow Jun 1 FastAPI for AI Engineers - Part 2: Building Your First CRUD API # ai # backend # fastapi # python 7 reactions Comments Add Comment 4 min read By the end of this article, you'll understand: Why in-memory storage is a problem What SQLite is What SQLAlchemy is How ORM works How to create database tables using Python classes How to perform CRUD operations using a real database The Problem with In-Memory Storage Previously, our application stored students inside a Python list. students = [ { " id " : 1 , " name " : " Ananya " , " department " : " CSE " , " cgpa " : 8.9 } ] This worked for learning CRUD operations. However, consider what happens when the server restarts: FastAPI Server Stops ↓ Python Memory Cleared ↓ All Student Data Lost This is unacceptable in real-world applications. We need a place where data can survive application restarts. This is where databases come in. What is SQLite? SQLite is a lightweight relational database. Unlike MySQL or PostgreSQL, SQLite doesn't require a separate database server. Instead, everything is stored inside a single file. students.db Advantages of SQLite: No installation required Lightweight Easy to learn Perfect for local development Great for small projects For this article, we'll use SQLite. What is SQLAlchemy? Before SQLAlchemy, developers often wrote raw SQL queries. Exampl

Imagine your smartwatch detects an irregular heart rhythm at 3 AM. Instead of just waking you up with a frantic "beep," an AI agent immediately analyzes your historical health data, searches for the best cardiologist nearby, and prepares a calendar invite for a consultation. This isn't science fiction—it's the power of Healthcare Automation driven by AI Agents . In this tutorial, we are diving deep into LangGraph , the cutting-edge framework for building stateful, multi-agent applications. We’ll explore how to use State Machines to orchestrate a complex medical workflow, moving from an "Abnormal Heart Rate Alert" to a "Specialist Appointment" using the Tavily API for research and Twilio for urgent notifications. By the end of this guide, you’ll understand how to manage non-linear LLM workflows that require reliability and precision. The Architecture: Why LangGraph? Traditional LLM chains are linear. But medical emergencies are not. They require loops, conditional branching (e.g., "Is this an emergency or a routine check-up?"), and state persistence. LangGraph allows us to define a graph where each node is a function and edges define the transition logic. Data Flow Overview The following diagram illustrates how our agent processes a heart rate alert: graph TD A[Start: Heart Rate Alert] --> B{Severity Triage} B -- Emergency --> C[Twilio: Alert Emergency Services] B -- High Risk --> D[Tavily API: Find Best Specialist] B -- Normal/Review --> E[Log to Health Records] D --> F[Google Calendar: Draft Appointment] F --> G[Twilio: SMS Patient Confirmation] C --> H[End] G --> H E --> H Prerequisites 🛠️ To follow along with this advanced tutorial, you'll need: Python 3.10+ LangGraph & LangChain : The orchestration engine. Tavily API Key : For searching local medical specialists. Twilio Account : For SMS/Voice alerting. An OpenAI API Key (GPT-4o is recommended for medical reasoning). Step 1: Defining the Agent State In LangGraph, the State is a shared schema that evolves as it m

We live in an era where scientific breakthroughs are published faster than we can read them. For the biohacking community, the gap between a new PubMed study on NAD+ precursors and actually knowing what dose to take is a chasm of manual research. What if you could build an LLM Agent that monitors research papers, processes them through a RAG (Retrieval-Augmented Generation) pipeline, and maps findings to your specific health profile? In this tutorial, we are building Paper-to-Action , a state-of-the-art agentic workflow using LangGraph , ChromaDB , and Mistral-7B . This isn't just a simple bot; it's a multi-stage reasoning engine designed to turn raw academic data into actionable health interventions. If you've been looking to master AI agents and personalized medicine automation, you’re in the right place. 🚀 The Architecture: From Raw Paper to Personalized Habit Traditional RAG pipelines are linear. To handle the nuance of medical research, we need a "looping" logic. We use LangGraph to manage the state of our agent, allowing it to decide if a paper is relevant before attempting to extract a protocol. System Flow graph TD A[Start: Keyword Trigger] --> B[Search PubMed/Arxiv API] B --> C{Relevance Filter} C -- No --> B C -- Yes --> D[Store in ChromaDB] D --> E[RAG: Extract Intervention Protocol] E --> F[Cross-Reference with User Profile] F --> G[Generate Personalized Action Plan] G --> H[End: Push to Health Checklist] Prerequisites To follow this advanced guide, you'll need: LangGraph : For the agentic state machine. ChromaDB : As our high-performance vector store. Mistral-7B : Running via Ollama or vLLM for local, private inference. Python 3.10+ Step 1: Defining the Agent State In LangGraph, everything revolves around the State . We need to track the fetched papers, the extracted data, and the final recommendation. from typing import Annotated , List , TypedDict from langgraph.graph import StateGraph , END class AgentState ( TypedDict ): keywords : List [ str ] user

This is part 3 of the Adaptive Model Routing series. Part 1 built an LLM categorizer with Groq — 8 categories, 3 tiers. Part 2 added k-NN embedding lookup in shadow mode, discovered 83% tier accuracy, and found 61% cost savings on paper. This post covers what happened next. When Phase 2 ended, I had a working embedding pool in shadow mode inside crab-bot. The category accuracy was sitting at 78.6%. Not bad — but the breakdown hid something worth looking at. Phase 3: When Validation Tells You a Category Doesn't Need to Exist The leave-one-out accuracy by category told the real story: Category Accuracy Tier casual 94% cheap simple_lookup 91% cheap creative 88% medium coding 92% strong reasoning 89% strong analysis 59% medium research_lookup 61% medium Two categories were basically a coin flip. And they were confusing each other — almost all of analysis's misses landed on research_lookup and vice versa. The obvious move would be to try fixing the categorizer prompt, tuning the LLM, or gathering more labeled data. I was about to go down that road when I noticed the column next to the accuracy: both categories mapped to the same tier . Medium. That changed everything. The question stopped being "why can't the model tell these apart?" and became: "what routing decision are we actually getting wrong?" The answer was zero. A misclassification between analysis and research_lookup produces no routing error. The routing outcome is identical either way. The confusion wasn't a model failure — it was a signal from the embedding space that the boundary between these two categories was artificial. If k-NN can't draw a line between them in 384 dimensions with 1,300 examples, maybe the line doesn't belong there. Decision: merge research_lookup into analysis. -- Re-label 243 rows where category was 'research_lookup' UPDATE routing_log SET category = 'analysis' WHERE category = 'research_lookup' ; The embeddings didn't change. The vectors were already correct — only the label stored al

GitHub热门项目 | LLM inference server with continuous batching & SSD caching for Apple Silicon — managed from the macOS menu bar | Stars: 15,979 | 108 stars today | 语言: Python

The Problem With Choosing a Local Model Everyone has an opinion on which local LLM is best. "Use Llama — it's the most popular." "Mistral 7B has the best quality." "Phi-3 Mini is small and efficient." None of these claims come with numbers. Specifically: your numbers, on your hardware, for your workload. I built a benchmarking system to change that. Three models, 30 prompts, full latency distribution, memory profiling per inference call, and a JSON validation layer to measure structured output reliability. Here's what I found — and why the results matter for anyone deploying local models in production. The Setup Three models tested: llama3.2:3b — 3B parameters, Q4_K_M quantization, 2 GB download phi3:mini — 3.8B parameters, Q4_K_M, 2.3 GB download mistral:7b — 7B parameters, Q4_K_M, 4.1 GB download Hardware: CPU only, no GPU acceleration. This is the worst-case baseline — the scenario that exposes real latency and memory numbers. 30 test prompts across 5 categories: Short factual (10): "What is the capital of France?" Reasoning (8): "Explain why the sky appears blue." Code generation (5): "Write a Python function to reverse a string." Structured output (5): "List 3 frameworks in JSON format with name and use_case." Multi-step (2): Complex chained reasoning tasks. Architecture POST /query → Pydantic validation → Ollama HTTP API → JSON Validator → QueryResponse POST /benchmark → Load test_prompts.json → For each prompt: psutil memory before → Ollama → psutil memory after → NumPy: P50/P95/P99 latency, avg TPS, peak/avg memory → BenchmarkResult JSON The benchmark runs prompts sequentially, not in parallel. Parallel would contaminate the per-prompt memory measurements. Results Llama 3.2 3B (Q4_K_M) avg_tokens_per_second : 42.3 p50_latency_ms : 1203 p95_latency_ms : 3847 p99_latency_ms : 5120 peak_memory_mb : 6953 avg_memory_mb : 6842 total_test_duration_s : 87.4 Interpretation: P50 at 1.2 seconds is excellent. P95 at 3.8 seconds misses a 3-second SLA — the outliers are m

GitHub热门项目 | A framework for building, orchestrating and deploying AI agents and multi-agent workflows with support for Python and .NET. | Stars: 11,059 | 29 stars today | 语言: Python

GitHub热门项目 | LLM驱动的 A/H/美股智能分析：多数据源行情 + 实时新闻 + LLM决策仪表盘 + 多渠道推送，零成本定时运行，纯白嫖. LLM-powered stock analysis system for A/H/US markets. | Stars: 40,920 | 339 stars today | 语言: Python

GitHub热门项目 | The best-benchmarked open-source AI memory system. And it's free. | Stars: 53,647 | 228 stars today | 语言: Python

GitHub热门项目 | A Simple and Universal Swarm Intelligence Engine, Predicting Anything. 简洁通用的群体智能引擎，预测万物 | Stars: 64,478 | 324 stars today | 语言: Python

GitHub热门项目 | Give your AI agent eyes to see the entire internet. Read & search Twitter, Reddit, YouTube, GitHub, Bilibili, XiaoHongShu — one CLI, zero API fees. | Stars: 21,280 | 127 stars today | 语言: Python

I Built a Python Package to Fix SSE Resumability in the MCP SDK Your MCP server crashed. Your client reconnected. Every event from that session? Gone. The Gap The Model Context Protocol Python SDK ships with a built-in EventStore that powers SSE stream resumability — when a client reconnects with a Last-Event-ID header, the server replays the events it missed. This works great in development. The catch: that store lives entirely in memory. Restart the process, roll a new deployment, or — in a multi-worker setup — have the reconnecting client land on a different pod, and the session is gone. The store was local to the process that died. Resumability silently returns nothing. This isn't a bug in the SDK. It's a scope decision — the in-memory store is a correct, useful default for single-process development. But the moment you deploy to production, you need something durable. That's the gap mcp-persist fills. What It Does mcp-persist adds three drop-in EventStore backends — SQLite , Redis , and PostgreSQL — that survive process restarts and work across multi-worker deployments. Pick the one that fits your infrastructure; the API is identical across all three. pip install "mcp-persist[sqlite]" # no external service needed pip install "mcp-persist[redis]" # for multi-worker deployments pip install "mcp-persist[postgres]" # for teams already running Postgres The Two-Line Setup Wiring resumability by hand is tedious — you need a store, a StreamableHTTPSessionManager , a Starlette lifespan to open and close both, and a Mount . The with_persistence() helper collapses all of that. Pass your FastMCP instance, get back a runnable ASGI app: import uvicorn from mcp.server.fastmcp import FastMCP from mcp_persist import with_persistence mcp = FastMCP ( name = " MyServer " ) app = with_persistence ( mcp , backend = " sqlite " , url = " events.db " , ttl = 3600 ) uvicorn . run ( app , host = " 127.0.0.1 " , port = 8000 ) # MCP endpoint at /mcp Switching to Redis is a one-word change:

Building a Real-Time Chat Feature with Django Channels and React Real-time features have become table stakes for modern web applications. Whether it is a customer support widget, a collaborative tool, or a social platform, users expect instant communication without page refreshes. In this article, I will walk through how we built a production-ready real-time chat feature using Django Channels and React at UCDREAMS. Why Django Channels? Django is traditionally synchronous. It handles one request at a time per worker. This works fine for standard HTTP requests, but WebSocket connections require persistent, bidirectional communication. Django Channels extends Django to handle WebSockets, background tasks, and asynchronous protocols alongside traditional HTTP. The beauty of Channels is that it does not replace Django. It layers on top, letting you keep your existing models, ORM, authentication, and admin panel while adding real-time capabilities. For a team already invested in Django, this is a massive advantage over introducing an entirely separate real-time server. Setting Up the Backend Start by installing Django Channels and a channel layer. Redis is the recommended backend for production use: channels == 4.0 . 0 channels - redis == 4.2 . 0 daphne == 4.0 . 0 Configure your Django settings: INSTALLED_APPS = [ ... " channels " , ] ASGI_APPLICATION = " your_project.asgi.application " CHANNEL_LAYERS = { " default " : { " BACKEND " : " channels_redis.core.RedisChannelLayer " , " CONFIG " : { " hosts " : [( " 127.0.0.1 " , 6379 )], }, }, } Building the WebSocket Consumer The consumer handles WebSocket connections: import json from channels.generic.websocket import AsyncWebsocketConsumer class ChatConsumer ( AsyncWebsocketConsumer ): async def connect ( self ): self . room_name = self . scope [ " url_route " ][ " kwargs " ][ " room_name " ] self . room_group_name = f " chat_ { self . room_name } " await self . channel_layer . group_add ( self . room_group_name , self . cha