一些零碎问题

最近遇到的问题

计算图

找到一条正确的计算图，其实就是拓扑排序，2年前学过，忘了。。

拓扑排序，每次寻找入度为0的放入队列，然后依次更新子节点的入度，循环，直到退出，贪心的思想。

from collections import deque
from typing import List, Dict, Set

class Node:
    """计算图节点"""
    def __init__(self, node_id: int):
        self.id = node_id
        self.children = []   # 子节点（输出依赖）
        self.parents = []    # 父节点（输入依赖）
    
    def add_child(self, child_node):
        """添加子节点"""
        if child_node not in self.children:
            self.children.append(child_node)
            child_node.parents.append(self)


class Graph:
    """计算图"""
    def __init__(self):
        self.nodes: Dict[int, Node] = {}
    
    def add_node(self, node_id: int) -> Node:
        """添加节点"""
        if node_id not in self.nodes:
            self.nodes[node_id] = Node(node_id)
        return self.nodes[node_id]
    
    def add_edge(self, from_id: int, to_id: int):
        """添加边：from -> to"""
        from_node = self.add_node(from_id)
        to_node = self.add_node(to_id)
        from_node.add_child(to_node)
    
    def topological_sort_kahn(self) -> List[int]:
        """
        Kahn算法拓扑排序
        返回节点ID的执行顺序
        """
        # 计算每个节点的入度（父节点数量）
        in_degree = {node_id: len(node.parents) for node_id, node in self.nodes.items()}
        
        # 找到所有入度为0的节点
        queue = deque([node_id for node_id, degree in in_degree.items() if degree == 0])
        
        result = []
        
        while queue:
            # 取出一个入度为0的节点
            node_id = queue.popleft()
            result.append(node_id)
            
            # 减少所有子节点的入度
            for child in self.nodes[node_id].children:
                in_degree[child.id] -= 1
                if in_degree[child.id] == 0:
                    queue.append(child.id)
        
        # 检查是否有环
        if len(result) != len(self.nodes):
            raise ValueError("图中存在环！")
        
        return result
    
    def topological_sort_dfs(self) -> List[int]:
        """
        DFS算法拓扑排序
        返回节点ID的执行顺序
        """
        visited = set()
        result = []
        
        def dfs(node_id: int):
            if node_id in visited:
                return
            visited.add(node_id)
            
            # 先处理所有子节点
            for child in self.nodes[node_id].children:
                if child.id not in visited:
                    dfs(child.id)
            
            # 后序遍历
            result.append(node_id)
        
        # 对所有节点执行DFS
        for node_id in self.nodes:
            if node_id not in visited:
                dfs(node_id)
        
        # 反转得到拓扑序
        result.reverse()
        
        if len(result) != len(self.nodes):
            raise ValueError("图中存在环！")
        
        return result


# 使用示例
if __name__ == "__main__":
    # 创建计算图
    # 输入: 0, 1 -> 卷积: 2, 3 -> ReLU: 4, 5 -> Add: 6 -> 输出: 7
    graph = Graph()
    
    # 添加边（依赖关系）
    graph.add_edge(0, 2)  # 节点0是节点2的输入
    graph.add_edge(1, 3)  # 节点1是节点3的输入
    graph.add_edge(2, 4)  # 节点2的输出给节点4
    graph.add_edge(3, 5)  # 节点3的输出给节点5
    graph.add_edge(4, 6)  # 节点4的输出给节点6
    graph.add_edge(5, 6)  # 节点5的输出给节点6
    graph.add_edge(6, 7)  # 节点6的输出给节点7
    
    # 拓扑排序
    print("Kahn算法:", graph.topological_sort_kahn())
    print("DFS算法:", graph.topological_sort_dfs())

cnn

cnn计算过程，n c h w 输入，kernel是c_out c_in k k，理解一下内部循环计算过程，直观上很简单，写出来还是有点细节需要注意

import numpy as np

def convolution_2d_batch(input_data, kernel, stride=1, padding=0):
    """
    批量卷积 (N, C, H, W)
    
    参数:
        input_data: (N, C, H, W) 批量的输入数据
        kernel: (C_out, C_in, K, K) 卷积核
        stride: 步长
        padding: 填充
    
    返回:
        output: (N, C_out, H_out, W_out)
    """
    n, c_in, h, w = input_data.shape
    c_out, c_in_k, k, _ = kernel.shape
    
    assert c_in == c_in_k, "输入通道数必须匹配"
    
    # 添加填充
    if padding > 0:
        input_padded = np.pad(input_data, ((0, 0), (0, 0), (padding, padding), (padding, padding)),
                              mode='constant', constant_values=0)
    else:
        input_padded = input_data
    
    out_h = (h + 2 * padding - k) // stride + 1
    out_w = (w + 2 * padding - k) // stride + 1
    
    output = np.zeros((n, c_out, out_h, out_w))
    
    for batch in range(n):
        for out_ch in range(c_out):
            for i in range(out_h):
                for j in range(out_w):
                    i_start = i * stride
                    j_start = j * stride
                    
                    total = 0
                    for in_ch in range(c_in):
                        window = input_padded[batch, in_ch, i_start:i_start+k, j_start:j_start+k]
                        total += np.sum(window * kernel[out_ch, in_ch])
                    
                    output[batch, out_ch, i, j] = total
    
    return output


# 示例3：批量卷积
print("\n" + "=" * 50)
print("示例3：批量卷积")
print("=" * 50)

# 输入 (batch=2, channels=1, height=4, width=4)
input_batch = np.array([
    [  # 第1个样本
        [
            [1, 2, 3, 4],
            [5, 6, 7, 8],
            [9, 10, 11, 12],
            [13, 14, 15, 16]
        ]
    ],
    [  # 第2个样本
        [
            [16, 15, 14, 13],
            [12, 11, 10, 9],
            [8, 7, 6, 5],
            [4, 3, 2, 1]
        ]
    ]
])

# 卷积核 (2个输出通道, 1个输入通道, 2x2)
kernel_batch = np.array([
    [  # 输出通道1
        [
            [1, 0],
            [0, -1]
        ]
    ],
    [  # 输出通道2
        [
            [0, 1],
            [-1, 0]
        ]
    ]
])

print("输入 (batch=2, channels=1):")
print(f"Shape: {input_batch.shape}")
print("第一个样本:")
print(input_batch[0, 0])

print("\n卷积核 (2个输出通道):")
print(f"Shape: {kernel_batch.shape}")
print("输出通道1:")
print(kernel_batch[0, 0])
print("输出通道2:")
print(kernel_batch[1, 0])

output_batch = convolution_2d_batch(input_batch, kernel_batch, stride=1, padding=0)
print(f"\n输出 Shape: {output_batch.shape}")
print("第一个样本的输出:")
print(output_batch[0])

并查集+Kruskal

既然提到了DAG有向无环图，顺带也学习一下最小生成树吧，即找到最短的边，构成一个图，但不能成环

环的问题可以用并查集解决，并查集没什么好说的，直接记模板，union、find两个关键函数，在此基础上，结合贪心，就是Kruskal

from typing import List, Tuple

def kruskal_basic(edges: List[Tuple[int, int, int]], n: int) -> Tuple[List[Tuple[int, int, int]], int]:
    """
    Kruskal算法 - 基础版本（无rank优化）
    
    参数:
        edges: 边列表，每个元素是 (u, v, weight)
        n: 顶点数量
    
    返回:
        mst: 最小生成树的边列表
        total_weight: 最小生成树的总权重
    """
    # 1. 按权重从小到大排序
    edges_sorted = sorted(edges, key=lambda x: x[2])
    
    # 2. 初始化并查集：每个节点的父节点是自己
    parent = list(range(n))
    
    # 3. find函数：查找根节点（带路径压缩）
    def find(u):
        if parent[u] != u:
            parent[u] = find(parent[u])
        return parent[u]
    
    # 4. union函数：合并两个集合
    def union(u, v):
        root_u = find(u)
        root_v = find(v)
        
        if root_u == root_v:
            return False  # 已经在同一集合
        
        # 简单的合并：将root_u的父节点设为root_v
        parent[root_u] = root_v
        return True
    
    # 5. 贪心选择边
    mst = []
    total_weight = 0
    
    for u, v, weight in edges_sorted:
        # 如果加入这条边不会形成环
        if union(u, v):
            mst.append((u, v, weight))
            total_weight += weight
            
            # 找到n-1条边就停止
            if len(mst) == n - 1:
                break
    
    return mst, total_weight


# ============================================
# 测试代码
# ============================================

print("=" * 70)
print("基础Kruskal算法")
print("=" * 70)

# 测试用例1：简单图
print("\n测试1：简单图")
print("-" * 40)

edges1 = [
    (0, 1, 10),
    (0, 2, 6),
    (0, 3, 5),
    (1, 3, 15),
    (2, 3, 4),
]
n1 = 4

mst1, weight1 = kruskal_basic(edges1, n1)
print(f"MST边: {mst1}")
print(f"总权重: {weight1}")

# 测试用例2：稍微复杂图
print("\n测试2：5个节点的图")
print("-" * 40)

edges2 = [
    (0, 1, 2),
    (0, 3, 6),
    (1, 2, 3),
    (1, 3, 8),
    (1, 4, 5),
    (2, 4, 7),
    (3, 4, 9),
]
n2 = 5

mst2, weight2 = kruskal_basic(edges2, n2)
print(f"MST边: {mst2}")
print(f"总权重: {weight2}")

# 测试用例3：需要跳过环的图
print("\n测试3：有环的图")
print("-" * 40)

edges3 = [
    (0, 1, 1),
    (0, 2, 2),
    (1, 2, 3),  # 这条会形成环
    (1, 3, 4),
    (2, 3, 5),
]
n3 = 4

mst3, weight3 = kruskal_basic(edges3, n3)
print(f"MST边: {mst3}")
print(f"总权重: {weight3}")

A*

提到了图，那就顺带再复习一下经典的Dijkstra和A*吧

A*就是在Dijkstra加上贪心，选择合适的启发式函数，是能够保证最短的（有数学证明）。

常见的启发式函数包括：曼哈顿距离、对角线距离、欧氏距离

import time

open_list = []
close_list = []


class Position:
    x = 0
    y = 0

    def __init__(self, X, Y):
        self.x, self.y = X, Y  #必须写成self.x,否则x是一个新的对象，将覆盖self.x

    def __eq__(self, other):  #重载== !=函数，self和other可能是None
        if (other and self):
            return self.x == other.x and self.y == other.y
        else:
            return not (self or other)


#存储地图每点信息Message
class Msg:
    G = 0
    H = 0
    IsUnk = 1  #此处替代close_list功能，判断某点是否被搜索过。可直接访问，不需在close_list中搜索 1代表需要搜，0代表搜过了
    parent = None  #Position

    def __init__(self):
        self.G = 0
        self.H = 0
        self.IsUnk = 1

    def GetF(self):  #F不适合写成对象，因为G对象时常更新，F依赖于G
        return self.G + self.H


#地图棋盘，内含Position和Msg信息mapx[Position]=Col*Row个Msg
class Board:
    mapx = []  #Msg[ROW][COL]

    def __init__(self, mapp, target_position):
        for i in range(len(mapp)):
            self.mapx.append([])
            for j in range(len(mapp[0])):
                self.mapx[i].append(
                    Msg())  #是Msg()而不是Msg,否则是一个新的对象，mapx中所有信息同步变动
                self.mapx[i][j].IsUnk = 1 - mapp[i][j]
                # 有很多万法可以估算H值。这里找们使用Manhattan万法，
                # 计算从当前万格横可或纵回移动到达目标所经过的方格数，忽略对角移动，然后把总数乘以10。
                self.mapx[i][j].H = 10 * (abs(target_position.x - i) +
                                          abs(target_position.y - j))

        # for k in range(len(self.mapx)):
        #     for m in range(len(self.mapx[k])):
        #         print(self.mapx[k][m].H,end=" ")
        #     print('')
        # print('')

    def GetMsg(self, pos):  #根据Position通过mapx获得Msg
        return self.mapx[pos.x][pos.y]


def IsInBoard(i, j):
    if (i >= 0 and i < len(mapp) and j >= 0 and j < len(mapp[i])
            and mapp[i][j] == 0):
        return 1
    else:
        return 0


# 相当于每次取出F最小的结点，同时更新该结点周围八个结点的G值，H值初始化的时候就定了，然后根据F=G+H进行排序，每次取最小
def SearchPath(mapp, start_position, target_position):
    start_time = time.time()  #计时
    board = Board(mapp, target_position)  #地图棋盘对象
    board.GetMsg(start_position).IsUnk = 0
    open_list.append(start_position)
    while (open_list != []):
        # for k in range(len(board.mapx)):
        #     for m in range(len(board.mapx[k])):
        #         if(board.mapx[k][m].parent):
        #             print('p',board.mapx[k][m].parent.__dict__,end=" ")
        #         else: print('p',"                ",end=" ")
        #     print('')
        # print('')
        #取出第一个（F最小，判定最优）位置
        current_position = open_list[0]
        open_list.remove(current_position)
        #close_list.append(current_position)
        #到达
        if (current_position == target_position):
            print("成功找到解")
            tmp = []  #内存储Position
            # 从终点，依次取出父结点，然后reverse才是路线
            while (current_position != None):
                tmp.append(current_position)
                current_position = board.GetMsg(current_position).parent
            tmp.reverse()
            for i in tmp:
                print(str(i.__dict__))
            end_time = time.time()  #计时
            print(end_time - start_time)
            return

        #将下一步可到达的位置加入open_list，并检查记录的最短路径G是否需要更新，记录最短路径经过的上一个点
        #斜（上下左右与此思路相同，只是细节有差） 这里只有两个元素
        for i in [current_position.x - 1, current_position.x + 1]:
            for j in [current_position.y - 1, current_position.y + 1]:
                if (IsInBoard(i, j)):
                    new_G = board.GetMsg(current_position).G + 14
                    #维护当前已知最短G
                    if (board.mapx[i][j].IsUnk):
                        board.mapx[i][j].IsUnk = 0
                        open_list.append(Position(i, j))
                        board.mapx[i][j].parent = current_position
                        board.mapx[i][j].G = new_G

                    if (board.mapx[i][j].G > new_G):  #如果未遍历或需更新
                        board.mapx[i][j].parent = current_position
                        board.mapx[i][j].G = new_G
        #上下
        j = current_position.y
        for i in [current_position.x - 1, current_position.x + 1]:
            if (IsInBoard(i, j)):
                new_G = board.GetMsg(current_position).G + 10
                if (board.mapx[i][j].IsUnk):
                    board.mapx[i][j].IsUnk = 0
                    open_list.append(Position(i, j))
                    board.mapx[i][j].parent = current_position
                    board.mapx[i][j].G = new_G

                if (board.mapx[i][j].G > new_G):  #如果未遍历或需更新
                    board.mapx[i][j].parent = current_position
                    board.mapx[i][j].G = new_G
        #左右
        i = current_position.x
        for j in [current_position.y - 1, current_position.y + 1]:
            if (IsInBoard(i, j)):
                new_G = board.GetMsg(current_position).G + 10
                if (board.mapx[i][j].IsUnk):
                    board.mapx[i][j].IsUnk = 0
                    open_list.append(Position(i, j))
                    board.mapx[i][j].parent = current_position
                    board.mapx[i][j].G = new_G

                if (board.mapx[i][j].G > new_G):  #如果未遍历或需更新
                    board.mapx[i][j].parent = current_position
                    board.mapx[i][j].G = new_G
        #open_list.sort(key=searchKey(board))
        #对open_list里的内容按F的大小排序
        open_list.sort(key=lambda elem: board.GetMsg(elem).GetF())


if __name__ == "__main__":
    #定义初始状态
    mapp = [[0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 1, 0, 0,
                                    0], [0, 0, 0, 1, 0, 0, 0],
            [0, 0, 0, 1, 0, 0, 0], [0, 0, 0, 1, 0, 0, 0],
            [0, 0, 0, 0, 0, 0, 0]]
    #mapp=[[0,0,0],[0,1,0],[0,0,0]]
    start_position = Position(0, 0)
    target_position = Position(5, 6)
    SearchPath(mapp, start_position, target_position)

岛屿问题

既然复习到图了，那就再看看经典的岛屿问题吧，顺带复习下BFS和DFS

200. 岛屿数量 - 力扣（LeetCode）

DFS

from typing import List


class Solution:
    def numIslands(self, grid: List[List[str]]) -> int:
        # 四个方向
        dirs = [[0, 1], [1, 0], [-1, 0], [0, -1]]
        n = len(grid)
        m = len(grid[0])
        visited = [[False] * m for _ in range(n)]
        ans = 0

        def dfs(x: int, y: int) -> None:
            # 越界、已访问、或者是水，直接返回
            if x < 0 or x >= n or y < 0 or y >= m:
                return
            if visited[x][y] or grid[x][y] == '0':
                return

            visited[x][y] = True
            for dx, dy in dirs:
                nextx = x + dx
                nexty = y + dy
                dfs(nextx, nexty)

        for i in range(n):
            for j in range(m):
                if not visited[i][j] and grid[i][j] == '1':
                    ans += 1
                    dfs(i, j)

        return ans

BFS

from typing import List
from collections import deque

class Solution:
    def numIslands(self, grid: List[List[str]]) -> int:
        # 四个方向
        dirs = [[0, 1], [1, 0], [-1, 0], [0, -1]]
        n = len(grid)
        m = len(grid[0])
        visited = [[False] * m for _ in range(n)]
        ans = 0
        
        def bfs(x: int, y: int) -> None:
            # 使用队列，只要加入队列就立刻标记
            queue = deque()
            queue.append((x, y))
            visited[x][y] = True  # 只要加入队列，就立刻标记
            
            while queue:
                curx, cury = queue.popleft()  # BFS 使用 popleft
                for dx, dy in dirs:
                    nextx = curx + dx
                    nexty = cury + dy
                    # 边界检查
                    if nextx < 0 or nextx >= n or nexty < 0 or nexty >= m:
                        continue
                    # 如果未访问且是陆地
                    if not visited[nextx][nexty] and grid[nextx][nexty] == '1':
                        queue.append((nextx, nexty))
                        visited[nextx][nexty] = True  # 只要加入队列，就立刻标记
        
        for i in range(n):
            for j in range(m):
                if not visited[i][j] and grid[i][j] == '1':
                    ans += 1
                    bfs(i, j)
        
        return ans

695. 岛屿的最大面积 - 力扣（LeetCode）

from typing import List


class Solution:
    count = 0
    def maxAreaOfIsland(self, grid: List[List[int]]) -> int:
        m, n = len(grid), len(grid[0])
        dir = [[0, 1], [0, -1], [1, 0], [-1, 0]]
        ans = 0
        visited = [[False] * n for _ in range(m)]
        def dfs(grid:List[List[int]], visited:List[List[bool]], x:int, y:int) -> int:
            for i in range(4):
                nextx, nexty = x + dir[i][0], y + dir[i][1]
                if nextx < 0 or nextx >= len(grid) or nexty < 0 or nexty >= len(grid[0]):
                    continue
                if not visited[nextx][nexty] and grid[nextx][nexty] == 1:
                    visited[nextx][nexty] = True
                    self.count += 1
                    dfs(grid, visited, nextx, nexty)

        def bfs(grid:List[List[int]], visited:List[List[bool]], x:int, y:int) -> int:
            queue = [(x, y)]
            while queue:
                x, y = queue.pop(0)
                for i in range(4):
                    nextx, nexty = x + dir[i][0], y + dir[i][1]
                    if nextx < 0 or nextx >= len(grid) or nexty < 0 or nexty >= len(grid[0]):
                        continue
                    if not visited[nextx][nexty] and grid[nextx][nexty] == 1:
                        visited[nextx][nexty] = True
                        self.count += 1
                        queue.append((nextx, nexty))

        for i in range(m):
            for j in range(n):
                if grid[i][j] == 1 and not visited[i][j]:
                    visited[i][j] = True
                    self.count = 1
                    bfs(grid, visited, i, j)
                    ans = max(ans, self.count)
        return ans

堆的问题

最小堆和最大堆问题，底层红黑树，细节就先管了，topk问题是最小堆问题

一些c++问题

move和forward

move

// 位于 <utility> 头文件中
template<typename T>
typename std::remove_reference<T>::type&& move(T&& t) noexcept {
    return static_cast<typename std::remove_reference<T>::type&&>(t);
}

• template<typename T> 模板参数，类型T
• T&& t 万能引用，接受任意类型
• std::remove_reference<T>::type&& 表示移除掉T的引用，得到T类型本身，也就是type，然后加一个&&，即表示右引用本身
• 本质上就是把任意类型的T，不管这个T是左值还是右值，都转换成右值，那么左值和右值到底有什么区别？编译器看到的有区别，左值有地址，右值一般放在寄存器上，作为程序员，我们知道右值是临时变量，主要是拿来移动，可以进行移动构造就够了。

forward

// 位于 <utility> 头文件中
template<typename T>
T&& forward(typename std::remove_reference<T>::type& t) noexcept {
    return static_cast<T&&>(t);
}

• template<typename T> 模板参数，这里必须显示指定，因为需要根据T的类型，去决定返回值

• typename std::remove_reference<T>::type& t 接受一个左值引用的参数

• static_cast<T&&>(t) 把传入的类型t视为T&& 类型

• 本质上，就是根据传入的类型T，永远接受一个左值引用，然后根据T的类型，决定返回左值还是右值

• 应用场景，同时有拷贝构造和移动构造的的时候，用forward可以自动根据参数，决定走哪一个构造函数，而不是统统走拷贝构造，节省性能

• template<typename F, typename... Args>
  auto wrapper(F&& f, Args&&... args) {
      return std::forward<F>(f)(std::forward<Args>(args)...);
  }
  
  std::forward<F>(f)           // 第1步：获取一个可调用的对象
  (                            // 第2步：调用它
      std::forward<Args>(args)...  // 传入参数
  )
  

  • 通过这样的wrapper类，它接收任何可调用对象和参数，原封不动地传递给底层函数，同时保持所有类型信息和值类别，实现零开销的抽象

关于c和c++相互调用

这块直接问AI吧，extern c+额外写一层胶水代码，利用void*万能指针

cuda

GitHub - WingEdge777/vitamin-cuda: 🍎 One kernel a day keeps high latency away. A hands-on CUDA learning path featuring a rich collection of kernels, from the basics to peak performance, seamlessly integrated as PyTorch C++ extensions. · GitHub

感觉可以拿来作为入门

infra的参考资料

• https://mlc.ai/modern-gpu-programming-for-mlsys/index.html
• https://blog.echen.io/
• https://ashraf-bhuiyan.com/blog/01-naive-inference/