1. 寫在最前面

最近加了很多 Python Coding 的任務，雖然在 AI 加持下能夠順利完成，但是還是覺得心理不踏實，覺得很多代碼 AI 寫完自己不是很懂，不喜歡這種感覺。

剛好周日上午沒事，抽空記錄在 Python 開發中遇到的神奇語法和庫。

2. Python 語法

2.1 yield

修改的代碼中，函數的返回很多處用了 Yield 而不是 return ，這引起了我的好奇。

2.1.1 yield 和 return 的區別

在 Python 中，yield 和 return 都用于從函數中返回值，但它們之間有一些重要的區別：

• 返回類型

  • return:

    • return 語句用于結束函數的執行，并將一個值返回給調用者。函數在執行到 return 時會立即退出。

    • 每次調用函數時，都是從頭開始執行，直到遇到 return。

  • yield:

    • yield 語句用于定義生成器函數。生成器函數在執行時不會立即返回值，而是返回一個生成器對象。

    • 當生成器函數被調用時，它不會立即執行，而是返回一個生成器對象。可以通過迭代這個生成器對象來逐步執行函數的代碼，每次遇到 yield 時返回一個值，并在下一次迭代時從上次停止的地方繼續執行。

• 內存使用

  • return:

    • 返回一個完整的結果（例如一個列表），可能會占用較多內存，特別是當結果很大時。

  • yield:

    • 生成器按需生成值，通常會更節省內存，特別是在處理大型數據集時，因為它不會一次性生成所有結果。

• yield 的使用場景：

  • 內存效率：適合處理大數據集，避免一次性加載。

  • 無限序列：可生成無限序列，按需計算。

  • 狀態保持：在函數中保持狀態，適合需要維護狀態的場景。

  • 協程：在異步編程中用于協作式多任務。

  • 管道處理：構建數據處理管道，方便數據流動。

• 示例

  • 使用 return 的函數:

    def get_squares(n):return [x*x for x in range(n)]squares = get_squares(5)
    print(squares)  # 輸出: [0, 1, 4, 9, 16]
    

  • 使用 yield 的生成器:

    def get_squares(n):for x in range(n):yield x*xsquares_gen = get_squares(5)
    for square in squares_gen:print(square)  # 輸出: 0, 1, 4, 9, 16
    

2.1.2 golang 中實現 yield 語法

之前寫的是 golang ，為了用類比法更好的理解 yield ，可以在 golang 中實現一個類似能力的示例。

package mainimport ("fmt"
)// 生成器函數，返回一個通道
func getSquares(n int) <-chan int {ch := make(chan int) // 創建一個通道go func() {for x := 0; x < n; x++ {ch <- x * x // 將計算結果發送到通道}close(ch) // 關閉通道，表示沒有更多值}()return ch // 返回通道
}func main() {squares := getSquares(5) // 獲取生成器通道for square := range squares { // 迭代通道中的值fmt.Println(square) // 輸出: 0, 1, 4, 9, 16}
}

函數說明：

• 通道：在 getSquares 函數中，我們創建了一個通道 ch，用于發送計算結果。

• goroutine：我們使用 go 關鍵字啟動了一個 goroutine，在這個 goroutine 中計算平方并將結果發送到通道。

• 關閉通道：當所有值都發送完后，我們關閉通道，以便接收方知道沒有更多值可供接收。

• 迭代通道：在 main 函數中，我們使用 range 迭代通道，從中接收值。

3. aiohttp 庫

需求是需要在客戶請求大模型前，提前發送一次請求大模型，確保在客戶請求的時候，就可以節省掉 tls 握手和 tcp 建立連接的時間，簡稱之為預熱。

3.1 原始寫法

async def analyze_backend_consistency(num_requests: int = 5, delay: float = 0.5, concurrent: bool = False):# 順序發送請求results = []for i in range(num_requests):try:connector = aiohttp.TCPConnector(ssl=True)async with aiohttp.ClientSession(connector=connector) as session:result = await make_request(session, i + 1)results.append(result)if i < num_requests - 1:await asyncio.sleep(delay)except Exception as e:print(f"請求 {i + 1} 失敗: {str(e)}")# 打印每個請求的結果for result in results:print(f"\n請求 {result['request_id']} 結果:")print(f"遠程IP:端口 = {result['remote_ip']}:{result['remote_port']}")print(f"請求ID = {result['x_request_id']}")print(f"處理時間 = {result['upstream_time']}ms")print(f"總響應時間 = {result['response_time']}ms")print("-" * 50)# 分析結果print("\n分析結果:")unique_ips = set(r['remote_ip'] for r in results)print(f"使用的不同IP數量: {len(unique_ips)}")print(f"IP列表: {', '.join(str(ip) for ip in unique_ips)}")# 計算平均響應時間avg_response_time = sum(float(r['response_time']) for r in results) / len(results)print(f"平均響應時間: {avg_response_time:.2f}ms")# 檢查是否所有請求都使用了同一個連接print(f"所有請求是否使用同一個連接: {len(unique_ips) == 1}")# 分析處理時間upstream_times = [float(r['upstream_time']) for r in results]print(f"后端處理時間范圍: {min(upstream_times):.2f}ms - {max(upstream_times):.2f}ms")print(f"后端處理時間平均值: {sum(upstream_times)/len(upstream_times):.2f}ms")

測試結果：

3.2 修改寫法

async def analyze_backend_consistency(num_requests: int = 5, delay: float = 0.5, concurrent: bool = False):connector = aiohttp.TCPConnector(ssl=True)async with aiohttp.ClientSession(connector=connector) as session:results = []if concurrent:# 并發發送所有請求tasks = [make_request(session, i + 1) for i in range(num_requests)]results = await asyncio.gather(*tasks)else:# 順序發送請求for i in range(num_requests):try:result = await make_request(session, i + 1)results.append(result)if i < num_requests - 1:await asyncio.sleep(delay)except Exception as e:print(f"請求 {i + 1} 失敗: {str(e)}")# 打印每個請求的結果for result in results:print(f"\n請求 {result['request_id']} 結果:")print(f"遠程IP:端口 = {result['remote_ip']}:{result['remote_port']}")print(f"請求ID = {result['x_request_id']}")print(f"處理時間 = {result['upstream_time']}ms")print(f"總響應時間 = {result['response_time']}ms")print("-" * 50)# 分析結果print("\n分析結果:")unique_ips = set(r['remote_ip'] for r in results)print(f"使用的不同IP數量: {len(unique_ips)}")print(f"IP列表: {', '.join(str(ip) for ip in unique_ips)}")# 計算平均響應時間avg_response_time = sum(float(r['response_time']) for r in results) / len(results)print(f"平均響應時間: {avg_response_time:.2f}ms")# 檢查是否所有請求都使用了同一個連接print(f"所有請求是否使用同一個連接: {len(unique_ips) == 1}")# 分析處理時間upstream_times = [float(r['upstream_time']) for r in results]print(f"后端處理時間范圍: {min(upstream_times):.2f}ms - {max(upstream_times):.2f}ms")print(f"后端處理時間平均值: {sum(upstream_times)/len(upstream_times):.2f}ms")

測試結果：

3.2 耗時對比分析

	原始寫法	修改寫法
特點	每個請求都創建新的 TCPConnector 和 ClientSession 每次請求都要重新建立 TCP 連接和 TLS 握手 不復用 HTTP/2 連接 每個請求的耗時 = TCP建立(~40ms) + TLS握手(~200ms) + 請求處理時間	只創建一次 TCPConnector 和 ClientSession TCP 連接和 TLS 握手只進行一次 復用 HTTP/2 連接 第一個請求耗時 = TCP建立 + TLS握手 + 請求處理時間 后續請求耗時 = 請求處理時間

4. 碎碎念

周日的時候寫了 80%，今晚剛好手里的活搞的差不多了，給總結收個尾。上海最近的暴雨和雷聲有點子嚇人：

• 世間最重要的事莫過于懂得讓自己屬于自己。

• 偶爾覺得媽媽很丟人，媽媽為什么連起碼的臉面的自尊都沒有呢？我都覺得上火。比起她自己，她有更想守護的，那就是我。人真正變強大，不是因為守護著自尊心，而是拋開自尊心的時候。所以媽媽很強大。

這周就要回去看媽媽啦，開心，就寫到這里吧。

5. 參考資料

• 如何理解Python中的yield用法?
• https://docs.aiohttp.org/en/stable/tracing_reference.html#aiohttp-client-tracing-reference