互动文字游戏看起来只是“阅读一段文字，再输入一句命令”，实际上同时包含状态理解、目标规划和序列决策。智能体可能需要先拿钥匙、再开门，或者检查容器后才能取得目标物品；某一步即时奖励为零，并不代表该动作没有长期价值。
直接让语言模型自由生成命令，常见问题是输出环境不接受的表达、重复无效操作，或者在上下文增长后偏离目标。只使用表格型 Q-learning 也有明显限制：自然语言观测几乎每一步都不同，状态空间稀疏，随机探索很难迅速发现有意义的动作。
一种更容易验证的做法是把职责拆开：
- 环境提供当前允许执行的候选命令，负责动作合法性。
- 语言模型只对候选命令排序，不直接创造动作。
- 强化学习根据环境返回的真实奖励更新长期价值。
- 轨迹日志保存状态、候选、选择和奖励，供失败复盘与离线评估使用。
这里的重点不是断言混合方案一定胜过某种基线，而是建立可测量、可替换的实验框架。最终效果仍取决于游戏结构、奖励密度、候选动作质量、模型能力和训练预算，必须通过固定任务集验证。
原理：将语言先验与长期回报分离
设状态为 s，环境给出的合法动作集合为 A(s)。表格型 Q-learning 的更新为：
Q(s, a) ← Q(s, a) + α[r + γ max Q(s', a') - Q(s, a)]
其中 α 是学习率，γ 是折扣因子。Q 值来自实际交互，不应由语言模型伪造。语言模型承担的是候选重排：根据当前描述、背包内容和历史动作，从 A(s) 中选出语义上更可能推进目标的一项。
执行时仍采用 ε-greedy 策略：以概率 ε 探索，否则选择当前 Q 值最高的动作。探索分支可以使用模型重排代替完全随机选择。这样既保留探索，又能利用语言先验减少明显无意义的尝试。随着训练推进逐步降低 ε，策略会更多依赖已经获得的回报证据。
自然语言状态不能直接无限存入 Q 表。示例将描述、背包和最近动作规范化后计算摘要，作为实验用状态键。这会产生状态混叠：两段文本相同但隐藏状态不同的场景可能被视为同一状态。更严谨的实验可以加入任务标识、已访问房间、已完成事件或环境公开的结构化字段。
环境与接入准备
先创建隔离环境并安装依赖：
python -m venv .venv
source .venv/bin/activate
pip install textworld requests
TextWorld 的安装条件和可用游戏格式可能随当前发行包及操作系统而异，应以项目当前文档为准。准备一个可由本机 TextWorld 打开的游戏文件，并通过命令行参数传给脚本。
模型服务通过环境变量配置，密钥不写入源码：
export LLM_API_URL='https://your-endpoint.example/v1/chat/completions'
export LLM_MODEL='your-model-id'
export LLM_API_KEY='replace-with-your-secret'
如果需要比较不同模型或中转接口，可以把端点封装在同一个适配层中。例如，在确认其当前接口文档与请求格式符合项目要求后，可将 HaerAPI 作为候选接入端点之一；本文示例只假设一种常见的类 Chat Completions JSON 结构，不代表任意服务都支持该路径或字段。
实现受约束的混合智能体
下面的脚本包含四个关键边界：只允许模型返回候选序号；解析失败时退回随机候选；网络请求设置超时；每一步写入 JSONL 轨迹。不同 TextWorld 包装器的 reset、step 返回结构可能不同，代码针对 textworld.start 的经典状态对象接口，若本地接口不同应按当前文档调整适配处。
import argparse
import hashlib
import json
import os
import random
from collections import defaultdict
import requests
import textworld
from textworld import EnvInfos
ALPHA = 0.2
GAMMA = 0.95
def state_key(state, recent_actions):
payload = {
"description": getattr(state, "description", ""),
"inventory": getattr(state, "inventory", ""),
"recent_actions": recent_actions[-3:],
}
raw = json.dumps(payload, ensure_ascii=False, sort_keys=True)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
def model_pick(state, actions):
url = os.environ.get("LLM_API_URL")
model = os.environ.get("LLM_MODEL")
api_key = os.environ.get("LLM_API_KEY")
if not all([url, model, api_key]):
return random.randrange(len(actions)), "random_no_config"
prompt = {
"goal": "Choose one legal action that is most likely to advance the game.",
"description": getattr(state, "description", ""),
"inventory": getattr(state, "inventory", ""),
"actions": [{
"index": i, "command": a} for i, a in enumerate(actions)],
"output_rule": "Return JSON only: {\"index\": integer}",
}
body = {
"model": model,
"temperature": 0,
"messages": [{
"role": "user", "content": json.dumps(prompt)}],
}
try:
response = requests.post(
url,
headers={
"Authorization": f"Bearer {api_key}"},
json=body,
timeout=(5, 30),
)
response.raise_for_status()
content = response.json()["choices"][0]["message"]["content"]
index = int(json.loads(content)["index"])
if 0 <= index < len(actions):
return index, "model"
except (requests.RequestException, KeyError, ValueError, TypeError, json.JSONDecodeError):
pass
return random.randrange(len(actions)), "random_fallback"
def train(game_file, episodes, max_steps, log_file):
infos = EnvInfos(
admissible_commands=True,
description=True,
inventory=True,
won=True,
lost=True,
)
env = textworld.start(game_file, infos)
q = defaultdict(float)
with open(log_file, "a", encoding="utf-8") as log:
for episode in range(episodes):
state = env.reset()
recent = []
epsilon = max(0.1, 1.0 - episode / max(1, episodes - 1))
for step in range(max_steps):
actions = list(getattr(state, "admissible_commands", []) or [])
if not actions:
break
key = state_key(state, recent)
if random.random() < epsilon:
index, source = model_pick(state, actions)
else:
index = max(range(len(actions)), key=lambda i: q[(key, actions[i])])
source = "q_table"
action = actions[index]
next_state, reward, done = env.step(action)
next_actions = list(getattr(next_state, "admissible_commands", []) or [])
next_key = state_key(next_state, recent + [action])
future = max((q[(next_key, a)] for a in next_actions), default=0.0)
old = q[(key, action)]
q[(key, action)] = old + ALPHA * (reward + GAMMA * future - old)
record = {
"episode": episode,
"step": step,
"state_key": key,
"action": action,
"source": source,
"reward": reward,
"done": done,
"won": bool(getattr(next_state, "won", False)),
"lost": bool(getattr(next_state, "lost", False)),
}
log.write(json.dumps(record, ensure_ascii=False) + "\n")
log.flush()
recent.append(action)
state = next_state
if done:
break
env.close()
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("game_file")
parser.add_argument("--episodes", type=int, default=20)
parser.add_argument("--max-steps", type=int, default=100)
parser.add_argument("--log", default="trajectories.jsonl")
args = parser.parse_args()
train(args.game_file, args.episodes, args.max_steps, args.log)
执行方式如下：
python agent.py ./games/example.ulx --episodes 20 --max-steps 100
默认值只是让流程能够运行的起点，不是经过证明的最优参数。正式实验应固定游戏文件、随机种子、步数上限和评估预算，并把参数写入实验记录。
如何做可复核评估
不要只看某一局是否通关。至少建立三组策略：完全随机候选、纯 ε-greedy Q-learning、模型辅助探索。每组使用相同任务集合与最大步数，并分别记录通关与否、累计环境奖励、执行步数、无效终止、模型调用次数和接口失败次数。
训练轨迹与评估轨迹应分开。评估阶段固定策略，不再更新 Q 表；否则不同策略经历的数据不同，结果很难公平解释。任务数量不足时，不宜依据少量成功案例推出普遍结论。若游戏会暴露候选命令，实验结论也只能说明“候选集合内的决策能力”，不能等同于开放式自然语言规划能力。
接口侧还应记录耗时、HTTP 状态类别和解析结果，但不要把密钥、完整认证头或含敏感数据的原始提示写入日志。涉及真实业务文本时，需要先确定数据分级、脱敏规则、留存周期和供应商处理条款。
常见问题
模型返回了解释文字，JSON 无法解析
提示词约束不能代替程序校验。生产实现可以在服务确实支持时使用结构化输出能力，但仍要验证类型与范围。解析失败必须回退到合法候选，不能把原始文本直接交给环境执行。
智能体不断在两个房间之间往返
状态键缺少历史信息，或者环境奖励过于稀疏。可以增加最近若干动作、已访问位置和关键物品状态，也可以对短周期重复动作加入小幅惩罚。惩罚会改变优化目标，应在实验配置中明确记录，不能把塑形奖励当成环境原始奖励。
Q 表几乎无法复用
描述中的分数、步数或措辞变化会制造大量不同摘要。应先规范化文本，或者改用环境提供的结构化事实。进一步使用神经网络近似 Q 值时，还要处理经验回放、目标网络与训练稳定性，不能直接把表格实现的结论外推过去。
接口超时导致训练停滞
为连接和读取分别设置超时，并采用有限次数、带抖动的退避重试。训练循环应能够降级为随机探索，同时记录降级原因。是否重试 429 或 5xx 要结合服务条款和响应头，避免无上限重放。
每一步都调用模型，成本不可控
只在探索分支、首次遇到某状态或 Q 值差距很小时调用，并缓存“规范化状态加候选集合”的排序结果。缓存键应包含模型标识和提示模板版本，否则模型或提示升级后会混用旧结果。
总结
文字游戏强化学习的工程难点，不只是选择哪种算法，而是把合法动作、语义先验、长期奖励和实验记录放在清晰边界内。环境负责定义可执行命令，模型负责候选重排，Q-learning 只从真实交互更新价值，日志负责让结果可复核。
这套结构的价值在于可替换和可比较：可以关闭模型形成基线，可以替换状态表示，也可以在不改变环境执行边界的前提下测试不同模型。只有固定任务、隔离评估并公开失败条件后，模型辅助探索是否真正改善样本效率，才是一个能够被数据回答的问题。
