Zhang XiJun

a = [1, 2, 3]      # ob_refcnt = 1
b = a              # ob_refcnt = 2
c = [a, "hello"]   # ob_refcnt = 3（因为 c[0] 也引用了它）
del b              # ob_refcnt = 2

操作系统
 └── 进程（Process） ← 资源容器（内存、文件描述符等）
      └── 线程（Thread） ← CPU 调度单元（至少一个）
           └── 协程（Coroutine） ← 用户态逻辑任务（可多个，协作式切换）

from time import sleep, perf_counter

def fetch_url(url):
    print('Fetching the URL')
    sleep(1)#模拟阻塞
    print('Finished fetching')
    return 'url_content'

def read_file(filepath):
    print('Reading the file')
    sleep(1)#模拟阻塞
    print('Finished reading')
    return 'file_content'

def main():
    url = 'example.com'
    filepath = 'example.txt'
    fetch_result = fetch_url(url)
    read_result = read_file(filepath)

if __name__ == '__main__':
    start_time = perf_counter()
    main()
    end_time = perf_counter()
    print(f'Time taken: {end_time - start_time:.2f} seconds')

Fetching the URL
Finished fetching
Reading the file
Finished reading
Time taken: 2.00 seconds

from time import sleep, perf_counter
import asyncio

#定义协程函数
async def fetch_url(url):
    print(f'Fetching the url')
    #把阻塞操作包装成协程
    await asyncio.sleep(1)
    print('Finished fetching')
    return 'url_content'

async def read_file(filepath):
    print('Reading the file')
    await asyncio.sleep(1)
    print('Finished reading')
    return 'file_content'

#若想使用await，需要把main函数定义成协程函数
async def main():
    url = 'example.com'
    filepath = 'example.txt'
    #创建任务
    tasks = [asyncio.create_task(coro) for coro in [
        fetch_url(url),
        read_file(filepath)]]
    fetch_result = await tasks[0]
    read_result = await tasks[1]

if __name__ == '__main__':
    start_time = perf_counter()
    #创建事件循环
    asyncio.run(main())
    end_time = perf_counter()
    print(f'Time taken: {end_time - start_time:.2f} seconds')

Fetching the url
Reading the file
Finished fetching
Finished reading
Time taken: 1.01 seconds

import asyncio

#协程函数
async def coroutine_func():
    return 'coroutine_result'

def main():
    print(coroutine_func())
    result = asyncio.run(coroutine_func())
    print("---")
    print(result)
    

if __name__ == '__main__':
    main()

<coroutine object coroutine_func at 0x000001958D88A4B0>
d:\code\python\learn_pythontips\learn_async\coroutine_func.py:8: RuntimeWarning: coroutine 'coroutine_func' was never awaited
  print(coroutine_func())
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
---
coroutine_result

import asyncio

async def work(name, delay):
    print(f"开始 {name}")
    await asyncio.sleep(delay)
    print(f"完成 {name}")
    return f"result-{name}"

async def main():
    # 创建任务 → 立即开始执行！
    task1 = asyncio.create_task(work("A", 2))
    task2 = asyncio.create_task(work("B", 1))

    # 等待任务完成（可获取结果）
    r1 = await task1
    r2 = await task2
    print(r1, r2)

asyncio.run(main())

开始 A
开始 B
完成 B
完成 A
result-A result-B

async def main():
    url = 'example.com'
    filepath = 'example.txt'
    results = await asyncio.gather(
        fetch_url(url),
        read_file(filepath))
    print(results)

Fetching the url
Reading the file
Finished fetching
Finished reading
['url_content', 'file_content']
Time taken: 1.00 seconds

results = asyncio.as_completed([
    fetch_url(url),
    read_file(filepath)])
for result in results:
    print(await result)

import asyncio
import time

def blocking_io():
    time.sleep(1)
    return "Done!"

async def main():
    result = await asyncio.to_thread(blocking_io)
    print(result)

asyncio.run(main())

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

asyncio.run(fetch("https://example.com"))

import aiofiles
import asyncio

async def main():
    # 写文件
    async with aiofiles.open('data.txt', 'w') as f:
        await f.write('Hello, async files!')

    # 读文件
    async with aiofiles.open('data.txt', 'r') as f:
        content = await f.read()
        print(content)

asyncio.run(main())

class DatabaseConnection:
    def __init__(self, host, port, username, password):
        self.host = host
        self.port = port
        self.username = username
        self.password = password

    def connect(self):
        return f"Connecting to database at {self.host}:{self.port} with username '{self.username}'"


def client():
    main_db = DatabaseConnection('localhost', 3306, 'root', 'password123')
    analytics_db = DatabaseConnection('192.168.1.1', 5432, 'admin', 'securepass')
    cache_db = DatabaseConnection('10.0.0.1', 27017, 'cacheuser', 'cachepass')

    print(main_db.connect())
    print(analytics_db.connect())
    print(cache_db.connect())


client()

def connection_factory(db_type):
    db_configs = {
        'main': {
            'host': 'localhost',
            'port': 3306,
            'username': 'root',
            'password': 'password123'
        },
        'analytics': {
            'host': '192.168.1.1',
            'port': 5432,
            'username': 'admin',
            'password': 'securepass'
        },
        'cache': {
            'host': '10.0.0.1',
            'port': 27017,
            'username': 'cacheuser',
            'password': 'cachepass'
        }
    }
    
    return DatabaseConnection(**db_configs[db_type])


# 测试工厂模式
if __name__ == "__main__":
    # 使用工厂函数创建不同类型的数据库连接
    main_db = connection_factory('main')
    analytics_db = connection_factory('analytics')
    cache_db = connection_factory('cache')
    
    # 测试连接
    print(main_db.connect())
    print(analytics_db.connect())
    print(cache_db.connect())

#factory_pattern.py
def connection_factory(db_type):
    """
    工厂函数：根据数据库类型创建相应的数据库连接
    
    Args:
        db_type (str): 数据库类型 ('main', 'analytics', 'cache')
    
    Returns:
        DatabaseConnection: 数据库连接实例
    
    Raises:
        KeyError: 当提供的数据库类型不存在时
    """
    from config import DATABASE_CONFIGS
    if db_type not in DATABASE_CONFIGS:
        raise ValueError(f"Unknown database type: {db_type}. Available types: {list(DATABASE_CONFIGS.keys())}")
    
    return DatabaseConnection(**DATABASE_CONFIGS[db_type])

#config.py
# 数据库配置字典
DATABASE_CONFIGS = {
    'main': {
        'host': 'localhost',
        'port': 3306,
        'username': 'root',
        'password': 'password123'
    },
    'analytics': {
        'host': '192.168.1.1',
        'port': 5432,
        'username': 'admin',
        'password': 'securepass'
    },
    'cache': {
        'host': '10.0.0.1',
        'port': 27017,
        'username': 'cacheuser',
        'password': 'cachepass'
    }
}

class DatabaseConnectionBuilder:
    """数据库连接建造者类"""
    
    def __init__(self,host,port,username,password):
        self._config = {
            'host': host,
            'port': port,
            'username': username,
            'password': password,
        }
        
    def set_max_connections(self, max_connections):
        """设置最大连接数"""
        if max_connections is not None and max_connections <= 0:
            raise ValueError("Max connections must be positive")
        self._config['max_connections'] = max_connections
        return self
    
    def set_timeout(self, timeout):
        """设置超时时间"""
        if timeout is not None and timeout <= 0:
            raise ValueError("Connect timeout must be positive")
        self._config['timeout'] = timeout
        return self
        
    def build(self):
        """构建最终的数据库连接对象"""  
        return DatabaseConnection(**self._config)
        
        
# 演示建造者模式的使用
if __name__ == "__main__":
    # 创建一个数据库连接建造者
    builder = DatabaseConnectionBuilder(
        host='localhost',
        port=3306,
        username='root',
        password='password123'
    )
    
    # 使用建造者模式构建数据库连接
    connection = builder.set_max_connections(100) \
                        .set_timeout(30) \
                        .enable_ssl() \
                        .set_connection_pool('my_pool') \
                        .set_retry_attempts(3) \
                        .enable_compression() \
                        .set_read_preference('secondaryPreferred') \
                        .build()

class DatabaseConnection:
    """数据库连接单例类"""
    
    _instance = None
    _lock = threading.Lock()  # 线程锁，确保线程安全
    
    def __new__(cls, *args, **kwargs):
        """重写__new__方法实现单例模式"""
        if cls._instance is None:
            with cls._lock:  # 使用双重检查锁定
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance
    
    def __init__(self, host, port, username, password):
        """初始化数据库连接参数"""
        # 防止重复初始化
        if hasattr(self, '_initialized'):
            return
        
        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self._initialized = True
    
    def connect(self):
        """连接数据库"""
        return f"Connecting to database at {self.host}:{self.port} with username '{self.username}'"

def client():
    db1 = DatabaseConnection(host='localhost', port=3306, username='root', password='password')
    db2 = DatabaseConnection(host='localhost', port=3306, username='root', password='password')
    print(db1 is db2)

client()

pip install package_name.whl

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

pip install .
# 或
python -m build

my_project/
├── pyproject.toml
├── src/
│   └── my_utils/               ← 你写的包
│       ├── __init__.py
│       └── math.py
├── scripts/
│   └── run_demo.py             ← 想在这里用 my_utils

# 👇 新增：显式声明包位置
[tool.hatch.build.targets.wheel]
packages = ["src/my_utils"]

uv pip install -e .

my_project/
├── my_package/
│   ├── __init__.py
│   └── module.py
├── tests/
├── setup.py
└── README.md

my_project/
├── src/
│   └── my_package/
│       ├── __init__.py
│       └── module.py
├── tests/
├── pyproject.toml
└── README.md

dist/my_package-0.1.0-py3-none-any.whl
dist/my_package-0.1.0.tar.gz

uv pip install -e .

import sys
print(sys.path)

@field_validator("name")
def _validate_name(cls, v: str) -> str:#cls 表示当前模型类
    v = str(v).strip()
    if not v:
        raise ValueError("name 不能为空")
    return v

─src
│  └─pytest_demo
│          calculator.py
│          __init__.py

"""
pytest_demo package
"""

from .calculator import Calculator

__all__ = ['Calculator']

└─tests
    │  test_calculator.py

from pytest_demo.calculator import Calculator

def test_add() -> None:
    calc = Calculator()
    assert calc.add(2, 1) == 3


def test_div() -> None:
    calc = Calculator()
    assert calc.divide(2, 1) == 2

import pytest

def test_divide_by_zero() -> None:
    calc = Calculator()
    with pytest.raises(ZeroDivisionError):
        calc.divide(2, 0)

#工厂函数
@pytest.fixture
def calc() -> Calculator:
    return Calculator()

def test_add(calc: Calculator) -> None:
    assert calc.add(2, 1) == 3

@pytest.fixture(scope="session")  # 整个测试会话只启动一次
def browser():
    driver = webdriver.Chrome()
    yield driver
    driver.quit()

def test_login(browser):
    browser.get("/login")
def test_profile(browser):
    browser.get("/profile")

pytest.fixture
def temp_file():
    # setup: 创建临时文件
    path = "temp.txt"
    with open(path, "w") as f:
        f.write("hello")
    
    yield path  # 测试函数在此处获得 path

    # teardown: 清理
    os.remove(path)

def test_read_file(temp_file):
    with open(temp_file) as f:
        assert f.read() == "hello"

pip install "lightrag-hku[api]"
cp env.example .env
lightrag-server

pip install lightrag-hku

LLM_BINDING=openai
LLM_MODEL=qwen3-max-preview
LLM_BINDING_HOST=https://dashscope.aliyuncs.com/compatible-mode/v1
LLM_BINDING_API_KEY=sk-

EMBEDDING_BINDING=openai
EMBEDDING_MODEL=text-embedding-v4
EMBEDDING_DIM=1024
EMBEDDING_BINDING_API_KEY=sk-
EMBEDDING_BINDING_HOST=https://dashscope.aliyuncs.com/compatible-mode/v1

############################
### 数据存储选择
############################
### 默认存储（推荐用于小规模部署）
LIGHTRAG_KV_STORAGE=JsonKVStorage
LIGHTRAG_DOC_STATUS_STORAGE=JsonDocStatusStorage
LIGHTRAG_GRAPH_STORAGE=NetworkXStorage
LIGHTRAG_VECTOR_STORAGE=NanoVectorDBStorage

### 图存储（推荐用于生产部署）
# LIGHTRAG_GRAPH_STORAGE=Neo4JStorage
# LIGHTRAG_GRAPH_STORAGE=MemgraphStorage

### PostgreSQL
# LIGHTRAG_KV_STORAGE=PGKVStorage
# LIGHTRAG_DOC_STATUS_STORAGE=PGDocStatusStorage
# LIGHTRAG_GRAPH_STORAGE=PGGraphStorage
# LIGHTRAG_VECTOR_STORAGE=PGVectorStorage

### PostgreSQL 配置
# POSTGRES_HOST=localhost
# POSTGRES_PORT=5432
# POSTGRES_USER=您的用户名
# POSTGRES_PASSWORD='您的密码'
# POSTGRES_DATABASE=您的数据库

### Neo4j 配置
# NEO4J_URI=neo4j+s://xxxxxxxx.databases.neo4j.io
# NEO4J_USERNAME=neo4j
# NEO4J_PASSWORD='您的密码'
# NEO4J_DATABASE=noe4j

### Milvus 配置
# MILVUS_URI=http://localhost:19530
# MILVUS_DB_NAME=lightrag
# MILVUS_USER=root
# MILVUS_PASSWORD=您的密码

lightrag-server

# 启动实例1
lightrag-server --port 9621 --workspace space1

# 启动实例2
lightrag-server --port 9622 --workspace space2

LIGHTRAG_KV_STORAGE=PGKVStorage
LIGHTRAG_VECTOR_STORAGE=PGVectorStorage
LIGHTRAG_GRAPH_STORAGE=PGGraphStorage
LIGHTRAG_DOC_STATUS_STORAGE=PGDocStatusStorage

class LightRAGStorage:
    """LightRAG存储和检索类

    支持多种存储后端：
    - KV存储: PostgreSQL
    - 文档状态存储: PostgreSQL
    - 图存储: Neo4j
    - 向量存储: Milvus

    使用workspace实现数据隔离
    """

    def __init__(self, workspace: str = "default"):
        """初始化LightRAG存储

        Args:
            workspace: 工作空间名称，用于数据隔离
        """
        self.workspace = workspace

        # 加载环境变量
        load_dotenv()

        # 使用文件同级的 lightrag_storage 目录
        self.working_dir = os.path.join(os.path.dirname(__file__), "lightrag_storage")

        self.rag: Optional[LightRAG] = None

        # 确保工作目录存在
        os.makedirs(self.working_dir, exist_ok=True)

    async def _get_llm_model_func(self):
        """LLM模型函数"""
        async def llm_model_func(
            prompt,
            system_prompt=None,
            history_messages=[],
            keyword_extraction=False,
            **kwargs
        ) -> str:
            return await openai_complete_if_cache(
                model=os.getenv("DASHSCOPE_MODEL", "qwen-max"),
                prompt=prompt,
                system_prompt=system_prompt,
                history_messages=history_messages,
                api_key=os.getenv("DASHSCOPE_API_KEY"),
                base_url=os.getenv("DASHSCOPE_API_BASE", "https://dashscope.aliyuncs.com/compatible-mode/v1"),
                **kwargs
            )
        return llm_model_func

    async def _get_embedding_func(self):
        """嵌入模型函数"""
        async def embedding_func(texts: List[str]) -> np.ndarray:
            return await openai_embed(
                texts,
                model=os.getenv("DASHSCOPE_EMBEDDING_MODEL", "text-embedding-v4"),
                api_key=os.getenv("DASHSCOPE_API_KEY"),
                base_url=os.getenv("DASHSCOPE_API_BASE", "https://dashscope.aliyuncs.com/compatible-mode/v1")
            )
        return embedding_func


    async def initialize(self) -> None:
        """初始化LightRAG实例"""
        if self.rag is not None:
            return

        # 获取模型函数
        llm_model_func = await self._get_llm_model_func()
        embedding_func = await self._get_embedding_func()

        # 存储配置 - 统一使用字符串方式，让LightRAG自动处理
        # 图存储配置
        graph_storage = os.getenv("LIGHTRAG_GRAPH_STORAGE", "Neo4JStorage")

        # KV存储配置
        kv_storage = os.getenv("LIGHTRAG_KV_STORAGE", "PGKVStorage")

        # 文档状态存储配置
        doc_status_storage = os.getenv("LIGHTRAG_DOC_STATUS_STORAGE", "PGDocStatusStorage")

        # 向量存储配置
        vector_storage = os.getenv("LIGHTRAG_VECTOR_STORAGE", "MilvusVectorDBStorage")

        # 创建LightRAG实例
        self.rag = LightRAG(
            working_dir=self.working_dir,
            embedding_func=EmbeddingFunc(
                func=embedding_func,
                embedding_dim=int(os.getenv("EMBEDDING_DIM", 1024))
            ),
            llm_model_func=llm_model_func,
            workspace=self.workspace,
            graph_storage=graph_storage,
            kv_storage=kv_storage,
            doc_status_storage=doc_status_storage,
            vector_storage=vector_storage
        )

        # 初始化存储
        await self.rag.initialize_storages()
        await initialize_pipeline_status()

    async def insert_text(self, text: str) -> None:
        """插入文本到LightRAG

        Args:
            text: 要插入的文本内容
        """
        if self.rag is None:
            await self.initialize()

        await self.rag.ainsert(text)

    async def insert_texts(self, texts: List[str]) -> None:
        """批量插入文本

        Args:
            texts: 文本列表
        """
        for text in texts:
            await self.insert_text(text)

    async def query(
        self,
        query: str,
        mode: str = "hybrid",
        **kwargs
    ) -> str:
        """执行查询

        Args:
            query: 查询文本
            mode: 查询模式 ("naive", "local", "global", "hybrid")
            **kwargs: 其他查询参数

        Returns:
            查询结果
        """
        if self.rag is None:
            await self.initialize()

        return await self.rag.aquery(
            query,
            param=QueryParam(mode=mode, **kwargs)
        )

公式: (x - 均值) / 标准差

问题: 均值和标准差都会被异常值严重影响！

例如:
正常值: [100, 150, 200, 180, 220]  → 均值 = 170
加入异常值: [100, 150, 200, 180, 220, 6445]  → 均值 = 1216 ❌

结果: 所有正常值都变成负数，异常值主导了整个标准化过程

公式: (x - 中位数) / IQR
其中 IQR = Q3 - Q1 (四分位距)

优势: 中位数和IQR不受异常值影响！

例如:
正常值: [100, 150, 200, 180, 220]  → 中位数 = 180, IQR = 70
加入异常值: [100, 150, 200, 180, 220, 6445]  → 中位数 = 190, IQR = 70 ✅

结果: 异常值不会扭曲正常数据的标准化结果

null
└── d(4)
    ├── b(3)
    │   ├── a(2)
    │   └── c(1)
    └── a(1)
        └── c(1)

useradd -d /home/bob -m bob
useradd -d /home/john -m john
useradd -d /home/mike -m mike

passwd alice
passwd bob
passwd john
passwd mike

mkdir /home/work

groupadd workgroup

usermod -a -G workgroup alice   # 添加为附加组
usermod -g workgroup alice       # 设置为 alice 的主组（primary group）
usermod -a -G workgroup bob
usermod -a -G workgroup john

getent group workgroup

chown alice:workgroup /home/work

chmod ug+rwx,o-rwx /home/work

su - bob

touch /home/work/bob.txt

ls -l /home/work/bob.txt

su - john
cat /home/work/bob.txt  #查看文件内容
echo "hello" >> /home/work/bob.txt
ls -l /home/work/bob.txt   #查看权限

su - mike
cat /home/work/bob.txt
echo "mike tried" >> /home/work/bob.txt
cd /home/work

#!/bin/bash
while echo "I'm making files!"
do
    mkdir adir
    cd adir
    touch afile
    sleep 10s
done

chmod +x badproc.sh
#验证
ls -l badproc.sh

./badproc.sh &

ps aux | grep badproc

kill 12345

rm -rf adir

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>

int main() {
    pid_t pid;

    /* fork another process */
    pid = fork();
    if (pid < 0) { /* error occurred */
        fprintf(stderr, "Fork Failed\n");
        exit(-1);
    }
    else if (pid == 0) { /* child process */
        printf("This is child process, pid=%d\n", getpid());
        execl("/bin/ls", "ls", NULL);  // 执行 ls 命令
        printf("Child process finished\n"); // 这句不会打印，除非 execl 失败
    }
    else { /* parent process */
        printf("This is parent process, pid=%d\n", getpid());
        wait(NULL);  // 等待子进程结束
        printf("Child Complete\n");
        exit(0);
    }
}

gcc -g -o fork fork.c

./fork

gdb ./fork
#在 fork() 调用前设置
(gdb) set follow-fork-mode child
#设置断点和 catch exec
break main
catch exec

#include <iostream>
using namespace std;

int main(void)
{
    cout << "Hello world" << endl;
    return 0;
}

#编译程序
g++ -o hello helloworld.cpp
#运行程序
./hello

/* fred.c */
#include <stdio.h>

void fred(int arg)
{
    printf("fred: we passed %d\n", arg);
}

/* bill.c */
#include <stdio.h>

void bill(char *arg)
{
    printf("bill: we passed %s\n", arg);
}

gcc -c bill.c fred.c

/* lib.h */
void bill(char *);
void fred(int);

/* program.c */
#include <stdlib.h>
#include "lib.h"   // 引入我们自己写的头文件

int main()
{
    bill("Hello World");
    exit(0);       // 正常退出
}

gcc -c program.c

ar crv libfoo.a bill.o fred.o

gcc -o program program.o libfoo.a

./program