深度Qlearning在机器人导航中的应用

    import numpy as np
    import tensorflow as tf
    from collections import deque
    import random
    
    # 定义环境
    class NavigationEnv:
    def __init__(self, map_size, start, goal):
        self.map_size = map_size
        self.start = start
        self.goal = goal
        self.state = start
        self.obstacles = [] # 障碍物坐标列表
    
    def reset(self):
        self.state = self.start
        return self.state
    
    def step(self, action):
        # 根据action更新智能体状态
        if action == 0:  # 向上移动
            new_state = (self.state[0], self.state[1] + 1)
        elif action == 1:  # 向下移动
            new_state = (self.state[0], self.state[1] - 1)
        elif action == 2:  # 向左移动
            new_state = (self.state[0] - 1, self.state[1])
        elif action == 3:  # 向右移动
            new_state = (self.state[0] + 1, self.state[1])
        else:
            raise ValueError("Invalid action")
    
        # 检查是否撞墙或到达目标
        if new_state in self.obstacles or new_state[0] < 0 or new_state[0] >= self.map_size[0] or new_state[1] < 0 or new_state[1] >= self.map_size[1]:
            reward = -1  # 撞墙惩罚
            done = True
        elif new_state == self.goal:
            reward = 100  # 到达目标奖励
            done = True
        else:
            reward = -0.1  # 每步的小惩罚,鼓励智能体尽快到达目标
            done = False
    
        self.state = new_state
        return new_state, reward, done
    
    # 定义深度Q-learning智能体
    class DeepQAgent:
    def __init__(self, state_size, action_size, gamma=0.99, epsilon=1.0, epsilon_decay=0.995, epsilon_min=0.01, learning_rate=0.001):
        self.state_size = state_size
        self.action_size = action_size
        self.gamma = gamma
        self.epsilon = epsilon
        self.epsilon_decay = epsilon_decay
        self.epsilon_min = epsilon_min
        self.learning_rate = learning_rate
    
        self.memory = deque(maxlen=2000)
        self.model = self.build_model()
        self.target_model = self.build_model()
        self.update_target_model()
    
    def build_model(self):
        model = tf.keras.Sequential()
        model.add(tf.keras.layers.Dense(64, input_dim=self.state_size, activation='relu'))
        model.add(tf.keras.layers.Dense(64, activation='relu'))
        model.add(tf.keras.layers.Dense(self.action_size, activation='linear'))
        model.compile(loss='mse', optimizer=tf.keras.optimizers.Adam(lr=self.learning_rate))
        return model
    
    def update_target_model(self):
        self.target_model.set_weights(self.model.get_weights())
    
    def remember(self, state, action, reward, next_state, done):
        self.memory.append((state, action, reward, next_state, done))
    
    def act(self, state):
        if np.random.rand() <= self.epsilon:
            return random.randrange(self.action_size)
        act_values = self.model.predict(state)
        return np.argmax(act_values[0])
    
    def replay(self, batch_size):
        minibatch = random.sample(self.memory, batch_size)
        for state, action, reward, next_state, done in minibatch:
            target = self.model.predict(state)
            if done:
                target[0][action] = reward
            else:
                t = self.target_model.predict(next_state)[0]
                target[0][action] = reward + self.gamma * np.amax(t)
            self.model.fit(state, target, epochs=1, verbose=0)
    
        if self.epsilon > self.epsilon_min:
            self.epsilon *= self.epsilon_decay
    
    # 训练智能体
    def train_agent(env, agent, episodes=1000, batch_size=32):
    for e in range(episodes):
        state = env.reset()
        state = np.reshape(state, [1, 2])
        for time in range(500):
            action = agent.act(state)
            next_state, reward, done = env.step(action)
            next_state = np.reshape(next_state, [1, 2])
            agent.remember(state, action, reward, next_state, done)
            state = next_state
            if done:
                agent.update_target_model()
                print("Episode {}/{}, score: {}".format(e, episodes, time))
                break
            if len(agent.memory) > batch_size:
                agent.replay(batch_size)
    
    # 测试智能体
    def test_agent(env, agent, num_tests=10):
    total_rewards = 0
    for _ in range(num_tests):
        state = env.reset()
        state = np.reshape(state, [1, 2])
        done = False
        while not done:
            action = np.argmax(agent.model.predict(state)[0])
            next_state, reward, done = env.step(action)
            next_state = np.reshape(next_state, [1, 2])
            state = next_state
            total_rewards += reward
    print("Average reward: {}".format(total_rewards / num_tests))
    
    # 示例用法
    env = NavigationEnv(map_size=(10, 10), start=(0, 0), goal=(9, 9))
    env.obstacles = [(2, 3), (4, 5), (7, 2)]
    agent = DeepQAgent(state_size=2, action_size=4)
    train_agent(env, agent)
    test_agent(env, agent)
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-18/6LExc9wSrJ5Ffk42UWamv87gHXIM.png)