AI 学习笔记:独立练习答案册
先提交自己的运行结果、失败记录与解释,再来对照。这里给出规则说明与可运行实现;各专题课程的新练习以对应章答案为准。
01|Agent 核心循环
- 停止条件:至少包含成功、不可恢复错误、步数上限与预算耗尽;不能只依赖模型说“完成”。
- 重复工具调用:用规范化后的工具名与参数生成调用键,重复出现时拒绝、复用缓存或转人工,并把选择写入轨迹。
- 副作用工具:拆成准备与提交两步;提交前校验权限、幂等键和人工批准,执行后记录不可抵赖的结果。
可运行答案:澄清、截止时间与审批提交
时钟在每个工具前检查,超时后不会开始下一项;无法强制终止已开始的外部调用,仍需工具自身超时。审批绑定同一计划字段,真实支付需要持久账本和提供商幂等键。 保存为独立Python文件执行。
import json
# 注入时钟可证明预算分支,不依赖sleep与偶然调度。
def run(question,clock,deadline,decisions):
trace=[];seen=set();clarified=False;calls=0
for step in decisions:
if clock()>=deadline:return 'timeout',trace,calls
name=step['name'];args=step.get('args',{})
if name=='final':return 'completed',trace,calls
if name!='lookup':raise ValueError('unknown tool')
key=json.dumps([name,args],sort_keys=True)
if key in seen:raise ValueError('repeated tool')
seen.add(key);calls+=1
result='未找到' if args['topic']=='missing' else 'policy'
trace.append((name,result))
if result=='未找到':
if clarified:return 'insufficient',trace,calls
clarified=True;trace.append(('clarify','请补充政策名称'))
return 'budget_exhausted',trace,calls
assert run('',lambda:0,10,[{'name':'lookup','args':{'topic':'missing'}}])[1][-1][0]=='clarify'
times=iter([0,11]);status,trace,calls=run('',lambda:next(times),10,[{'name':'lookup','args':{'topic':'policy'}},{'name':'lookup','args':{'topic':'other'}}])
assert status=='timeout' and calls==1
for steps in [[{'name':'delete'}],[{'name':'lookup','args':{'topic':'x'}}]*2]:
try:run('',lambda:0,10,steps)
except ValueError:pass
else:raise AssertionError('invalid action accepted')
ledger={};audit=[]
def prepare(actor,order,amount):return {'actor':actor,'order':order,'amount':amount,'key':f'{actor}:{order}:{amount}'}
def commit(plan,approval):
if approval!=plan or plan['actor']!='owner' or plan['amount']<=0:raise PermissionError('approval or scope')
if plan['key'] not in ledger:
ledger[plan['key']]='refund-receipt';audit.append({'key':plan['key'],'status':'committed'})
return ledger[plan['key']]
p=prepare('owner','o1',10)
assert commit(p,dict(p))==commit(p,dict(p)) and len(audit)==1
try:commit(dict(p,amount=11),p)
except PermissionError:pass
else:raise AssertionError('changed plan accepted')
print('clarification, deadline, approval and dedup passed')02|推理与思考
- 多数投票失效:多个样本可能共享同一错误来源;应按独立性、证据与校验器得分,而不是只数答案。
- 搜索预算:同时限制宽度、深度、时间和模型调用成本;达到上限时返回当前最佳候选及不确定性。
- 外显过程边界:保存可审计的结构化决策与证据即可,不要求持久化模型的私有推理文本。
可运行实现:分歧、预算与同源错误
保存为 reasoning_answers.py,Python 3.11+ 运行。检查器分别检查输入类型、任务约束和可执行关系;它们有不同失败条件,但并不是三个统计独立的模型。这是确定性任务验证,不能用“三个函数”证明三份独立证据。
from collections import Counter
def checks(x):
return (type(x) is int, type(x) is int and 0 <= x <= 10,
type(x) is int and 6*x == 24)
def choose(values, width, deadline, clock):
if width <= 0:
raise ValueError('positive width required')
seen = []
for x in values[:width]:
if clock() >= deadline:
return {'answer': seen[0] if seen else None, 'uncertain': True}
if all(checks(x)):
seen.append(x)
return {'answer': seen[0] if seen else None, 'uncertain': not bool(seen)}
assert checks(True) == (False, False, False)
assert choose([5,4],2,10,lambda:0) == {'answer':4,'uncertain':False}
times = iter([0,10])
assert choose([4,5],2,10,lambda:next(times)) == {'answer':4,'uncertain':True}
assert choose([5,4],1,10,lambda:0)['answer'] is None
assert Counter([5]*9+[4]).most_common(1)[0][0] == 5
assert choose([5]*9+[4],10,10,lambda:0)['answer'] == 4
print('verification and limits ok')输出 verification and limits ok。到达时限后不再进入下一候选检查,保留已有结果但明确本轮未检查完整集合。它不抢占正在执行的远程请求;在线调用还需要传递剩余截止时间并处理取消。若检查器分歧,返回可定位的各项结果并转人工或补证据,不用多数票覆盖硬约束。
Self-Ask 与 rollout 变体
将正文保存为 selfask_protocol.py、mcts_rollout.py 后,同目录保存运行以下答案。这里只导入本节已给出的完整文件,无隐藏辅助函数。
from selfask_protocol import self_ask, scripted, follow, facts
from mcts_rollout import search, verify
assert self_ask('?',scripted(follow('unknown')),facts.get)['evidence']==[]
assert search(max_depth=2)['path'] is None
assert search(max_depth=3)['path']==('start','bridge','garden','goal')
assert not verify(('start','goal'))
for world in (lambda state, action: state, lambda state, action: ('start',),
lambda state, action: state + ('missing',)):
try:search(iterations=1,max_depth=2,world=world)
except ValueError:pass
else:raise AssertionError('invalid transition accepted')
print('protocol failure and complete-path budget variants passed')深度计算的是边数,可行路径需要3条边;把预算2时没有候选写成“搜索答案错误”会混淆没有证据与反证。
03|自主、反思与多 Agent
- 角色拆分依据:权限、工具、上下文或验证责任确实不同;仅仅换三段提示词不构成有效分工。
- 共享状态冲突:使用单一 owner、版本号或事件日志;冲突必须显式失败或合并,不能最后写入静默覆盖。
- 反思停止:分数无改善、重复建议、预算耗尽或校验通过时停止,避免“反思”成为无限循环。
可运行答案:角色冲突、无进展与权限
信任边界:writer → 本地格式化;reviewer → 只读证据代理;二者经版本化工作项交换结果,不能互相继承工具权限。Python策略用于控制流实验,真实网络隔离还需独立进程/凭据/沙箱。 保存为独立Python文件执行。
from dataclasses import dataclass
@dataclass
class WorkItem:
text:str
version:int=0
def update(item,expected,text):
if item.version!=expected:raise RuntimeError('conflict')
item.text=text;item.version+=1
item=WorkItem('draft');writer_version=reviewer_version=item.version
update(item,writer_version,'new draft')
try:update(item,reviewer_version,'stale review overwrite')
except RuntimeError:pass
else:raise AssertionError('lost update')
assert item.text=='new draft'
def reviews(suggestions,budget=5):
previous=None;same=0
for count,suggestion in enumerate(suggestions,1):
same=same+1 if suggestion==previous else 1
previous=suggestion
if same>=2:return 'no_new_advice'
if count>=budget:return 'budget_exhausted'
return 'incomplete'
assert reviews(['cite source','cite source'])=='no_new_advice'
assert reviews(['a','b','c'],2)=='budget_exhausted'
def tool(role,name):
allowed={'writer':{'format'},'reviewer':{'read_evidence'}}
if name not in allowed[role]:raise PermissionError('role boundary')
return 'local-format' if name=='format' else 'approved evidence'
try:tool('writer','read_evidence')
except PermissionError:pass
else:raise AssertionError('writer got network')
assert tool('reviewer','read_evidence')=='approved evidence'
print('version conflict, stop condition, role policy passed')04|记忆与知识
- 记忆层次:工作记忆服务当前任务,情节记忆保存经历,语义记忆保存稳定事实;三者保留期和更新权不同。
- 删除语义:删除原文后还要处理切片、向量、缓存、摘要、备份与派生索引,并保留合规审计记录。
- 过期知识:元数据带来源时间和有效期;检索时过滤或降权,生成时明确引用日期。
可运行实现:主体、冲突与遗忘
下面是完整答案,保存 memory_answer.py,用Python3.11+运行。先重建首次实验状态,再增加同租户另一主体和新旧冲突记录。
from dataclasses import dataclass, replace
@dataclass(frozen=True)
class Memory:
id: str
tenant: str
subject: str
text: str
source: str
observed: int
expires: int
confidence: float
deleted: bool = False
items = [
Memory('a', 'A', 'u1', '偏好简短', 'explicit', 10, 100, 1.0),
Memory('b', 'B', 'u1', '偏好详细', 'explicit', 11, 100, 1.0),
Memory('c', 'A', 'u1', '偏好长文', 'inferred', 9, 100, 0.5),
Memory('d', 'A', 'u1', '订单未发货', 'order-api', 8, 15, 1.0),
]
def retrieve(tenant, subject, query, now):
visible = [m for m in items if m.tenant == tenant and m.subject == subject
and not m.deleted and m.expires > now and query in m.text]
return sorted(visible, key=lambda m: (m.source == 'explicit', m.observed, m.confidence), reverse=True)
assert [m.id for m in retrieve('A', 'u1', '偏好', 20)] == ['a','c']
assert retrieve('A', 'u2', '偏好', 20) == []
assert retrieve('A', 'u1', '订单', 20) == []
# 删除墓碑优先于相关性;保留身份供派生缓存失效使用。
items[0] = replace(items[0], deleted=True)
assert [m.id for m in retrieve('A', 'u1', '偏好', 20)] == ['c']
# 模拟换页:只选能装入预算的条目,归档仍存在但没有全部进入上下文。
def page(memories, capacity):
chosen = []
for memory in memories:
size = len(memory.text)
if size <= capacity:
chosen.append(memory.id)
capacity -= size
return chosen
assert page(retrieve('A','u1','偏好',20), 4) == ['c']
assert page(retrieve('A','u1','偏好',20), 3) == []
print('isolation, expiry, deletion and paging ok')
items[0] = replace(items[0], deleted=False)
items.extend([
Memory('e','A','u2','偏好简短','explicit',30,100,1.),
Memory('f','A','u1','偏好长文','inferred',40,100,.9),
Memory('g','A','u1','偏好清楚','explicit',35,100,1.),
])
assert [m.id for m in retrieve('A','u2','偏好',50)]==['e']
assert [m.id for m in retrieve('A','u1','偏好',50)]==['g','a','f','c']
# 新的推断f仍不覆盖明确确认的g与a;本例排序不是自动事实合并。
items[:] = [replace(m,deleted=True) if (m.tenant,m.subject)==('A','u1') else m for m in items]
assert retrieve('A','u1','偏好',50)==[]
assert retrieve('A','u2','偏好',50)[0].id=='e'
assert retrieve('B','u1','偏好',50)[0].id=='b'
print('forget scoped subject ok')输出 forget scoped subject ok。实际删除清单按来源ID追踪:原文→切片→向量→摘要→缓存→异步队列与备份恢复流程。在线读取先检查删除墓碑;派生副本清除后运行相同主体的检索,结果必须为空。备份按独立保留期清理,恢复时重放删除标记,不能用“检索不到”推断所有介质已经物理擦除。
05|AutoGen
- 安全执行:默认隔离容器、只读输入、临时工作区、网络关闭、CPU/内存/时间限制与输出审查。
- 迁移旧 API:先锁旧版回归轨迹,再按当前 AgentChat/Core API 重写最小路径,最后比较消息、终止与工具事件。
- 团队终止:业务完成、不可恢复错误、最大轮次、预算与人工取消都应成为可测试事件。
可运行答案:真实团队与旧版维护
下面使用正文 0.7.4 环境独立运行;每个变体都新建客户端与团队。
import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.replay import ReplayChatCompletionClient
async def test(limit):
client = ReplayChatCompletionClient(['draft','checked'])
try:
team = RoundRobinGroupChat([
AssistantAgent('writer',model_client=client),
AssistantAgent('reviewer',model_client=client)
],termination_condition=MaxMessageTermination(limit))
result = await team.run(task='title')
expected = ['user','writer'] if limit == 2 else ['user','writer','reviewer']
assert [m.source for m in result.messages] == expected
assert result.stop_reason is not None
finally:
await client.close()
async def main():
await test(2)
await test(3)
print('message limit is not approval')
asyncio.run(main())旧版环境把正文 reply 替换为以下函数,其余注册保持不变,在原断言后追加空输入断言:
def reply(recipient, messages=None, sender=None, config=None):
if not messages:
return True, 'empty'
return True, 'checked:' + messages[-1]['content'].strip()
# 此函数必须在 agent.register_reply 之前定义;注册后运行:
# assert agent.generate_reply(messages=[]) == 'empty'审批与停滞用标准库独立运行:
def allowed(version, approval):
return approval is not None and approval == version
assert not allowed(1,None)
assert allowed(1,1)
assert not allowed(2,1)
def status(drafts):
return 'stalled' if len(drafts)>=2 and drafts[-1]==drafts[-2] else 'working'
assert status(['a']) == 'working'
assert status(['a','a']) == 'stalled'
assert status(['a','b']) == 'working'
print('approval binding and stall ok')版本批准在本例代表已认证服务端的授权记录,不能接收模型自报 approved。相同草稿的检测只证明没有文字变化;实际任务还需外部评分、预算与业务完成条件,避免措辞改变却毫无进展。
新工具、Selector 与自定义 Agent 的完整变体
把正文3个文件保存在同目录,使用现代0.7.4解释器运行:
import asyncio
from autogen_tools_real import case as tool_case
from autogen_selector import case as team_case
from autogen_custom import Reviewer
from autogen_core import CancellationToken
async def main():
assert (await tool_case(120,'missing')).is_error
assert await team_case('another title')==['user','writer','editor']
agent=Reviewer()
before=await agent.save_state()
for bad in [{'version':2,'count':3},{'version':1,'count':-1}]:
try:await agent.load_state(bad)
except ValueError:pass
else:raise AssertionError('invalid state accepted')
assert await agent.save_state()==before
token=CancellationToken();token.cancel()
for _ in range(2):
try:await agent.on_messages([],token)
except asyncio.CancelledError:pass
else:raise AssertionError('cancel ignored')
assert agent.count==0
print('tool, selector, state rejection and repeated cancellation passed')
asyncio.run(main())06|LangChain
- 最小边界:Prompt、模型、工具与状态分别可替换;先测纯函数工具和输入输出 schema,再接真实模型。
- 版本迁移:LangChain v1 的 Agent 主入口是
langchain.agents.create_agent;旧教程导入路径要标注版本而非混用。 - 可观测性:一次运行至少关联输入、模型配置、工具调用、延迟、token/成本、错误和最终校验结果。
可运行答案:真实旧Agent迁移与离线工具测试
固定环境沿用LangChain新课,旧式initialize_agent使用langchain-classic 1.0.1,不宣称复现所有历史0.x。旧结果output/intermediate_steps与新messages不同;测试最终文本、工具观察和调用ID,不直接比较整个字典。模型替身只验证执行协议,未评价真实模型质量。 保存为独立Python文件执行。
from typing import Any
from langchain_classic.agents import initialize_agent,AgentType
from langchain_core.language_models.fake import FakeListLLM
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.tools import tool
from langchain_core.messages import AIMessage,ToolMessage
from langchain_core.outputs import ChatResult,ChatGeneration
from langchain.agents import create_agent
calls=[]
@tool
def lookup(key:str)->str:
"""Read the sample policy."""
calls.append(key);return '2 days'
old=initialize_agent([lookup],FakeListLLM(responses=['Action: lookup\nAction Input: policy','Final Answer: 2 days']),agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,return_intermediate_steps=True,max_iterations=3)
a=old.invoke({'input':'policy'})
assert a['output']=='2 days' and a['intermediate_steps'][0][1]=='2 days'
class Chat(BaseChatModel):
name_to_call:str='lookup'
@property
def _llm_type(self):return 'fixture'
def bind_tools(self,tools,**kwargs):return self
def _generate(self,messages,stop=None,run_manager=None,**kwargs):
if any(isinstance(m,ToolMessage) for m in messages):message=AIMessage(content='2 days')
else:message=AIMessage(content='',tool_calls=[{'id':'c1','name':self.name_to_call,'args':{'key':'policy'},'type':'tool_call'}])
return ChatResult(generations=[ChatGeneration(message=message)])
new=create_agent(Chat(),[lookup]);b=new.invoke({'messages':[{'role':'user','content':'policy'}]})
assert b['messages'][-1].content==a['output']
assert any(isinstance(m,ToolMessage) and m.tool_call_id=='c1' for m in b['messages'])
before=len(calls)
invalid=create_agent(Chat(name_to_call='unknown'),[lookup]).invoke({'messages':[{'role':'user','content':'policy'}]})
assert len(calls)==before
assert any(isinstance(m,ToolMessage) and m.status=='error' for m in invalid['messages'])
# 输入与预算在下游前判断;超时不是用等待模拟。
def guarded(text,now,deadline,downstream):
if now>=deadline:return {'ok':False,'error':'timeout'}
if not text.strip():return {'ok':False,'error':'empty'}
return {'ok':True,'value':downstream(text.strip())}
seen=[]
assert guarded(' ',0,10,seen.append)['error']=='empty'
assert guarded('x',10,10,seen.append)['error']=='timeout'
assert seen==[]
print('legacy output/tool observation matches modern messages; unknown tool rejected')07|LLM 基础
保存 answer.py,Python 3.11+ 运行 python3 answer.py。均匀分布每项0.25,top-p阈值0.5需两项,1.0需四项;因果mask在位置0只能看到第一个值。
from math import exp
def nucleus(logits, p=0.9, temperature=1.):
if not 0 < p <= 1 or temperature <= 0:
raise ValueError('invalid sampling parameters')
scaled = [x/temperature for x in logits]
peak = max(scaled)
values = [exp(x-peak) for x in scaled]
probabilities = [x/sum(values) for x in values]
ordered = sorted(enumerate(probabilities), key=lambda x:-x[1])
kept, mass = [], 0.
for token, probability in ordered:
kept.append(token); mass += probability
if mass >= p:
break
return kept
assert nucleus([0.,0.,0.,0.],0.6) == [0,1,2]
assert nucleus([100.,0.],0.9) == [0]
print(nucleus([0.,0.,0.,0.],0.6))
from math import exp
def attend(scores, values, position):
visible = scores[:position+1]
peak = max(visible)
exps = [exp(s-peak) for s in visible]
weights = [e/sum(exps) for e in exps]
return sum(w*v for w,v in zip(weights,values)), weights
out0, weights0 = attend([0.,0.],[10.,20.],0)
out1, weights1 = attend([0.,0.],[10.,20.],1)
assert out0 == 10. and weights0 == [1.]
assert out1 == 15. and weights1 == [0.5,0.5]
print(out0,out1)
assert len(nucleus([0.,0.,0.,0.],0.5)) == 2
assert len(nucleus([0.,0.,0.,0.],1.0)) == 4
assert attend([0.,0.],[10.,20.],0)[0] == 10.
print('sampling and causal mask passed')如果位置0输出15,意味着混入未来值20;这不是更准确,而是违反生成时的信息约束。
张量形状与构造边界
沿用正文decoder.py的固定环境。将以下代码追加到文件末尾运行;子类仅加入构造合同,其余前向与训练仍使用正文Decoder。
class CheckedDecoder(Decoder):
def __init__(self,vocab,width=16,heads=2):
if type(heads) is not int or type(width) is not int or heads<=0 or width<=0 or width%heads:
raise ValueError('positive width must be divisible by positive heads')
super().__init__(vocab,width,heads)
four=CheckedDecoder(tok.get_vocab_size(),heads=4)
y,w=four(ids)
assert w.shape==(1,4,ids.shape[1],ids.shape[1])
assert y.shape==(1,ids.shape[1],tok.get_vocab_size())
for width,heads in [(15,2),(16,0)]:
try:CheckedDecoder(tok.get_vocab_size(),width,heads)
except ValueError:pass
else:raise AssertionError('invalid head shape accepted')
print('four heads and construction boundary passed')16维分成4头后每头4维,输出重新合并成16维;不能把头数增加误认为hidden width增加。先拒绝零头数,再求余,避免校验自己抛除零错误。
08|Prompt 工程
保存 answer.py,Python 3.11+ 运行 python3 answer.py。搜索目标和排序准则必须一致,不能只改任务说明。
def search(target):
frontier = [(0, [])]
for _ in range(2):
expanded = [(v+d,path+[d]) for v,path in frontier for d in [-1,1]]
frontier = sorted(expanded,key=lambda item:abs(target-item[0]))[:2]
return frontier[0]
assert search(-2) == (-2,[-1,-1])
assert search(2) == (2,[1,1])
assert search(2)[0] != -2
print('objective and evaluator agree')最后一条模拟任务要求-2、评审仍偏好2的失配。更多采样不会自动修复错误评审标准。
09|RAG
保存 answer.py,Python 3.11+ 运行 python3 answer.py。输入ID是回填依据;缺失或重复输出应在发布前失败。
def join(requests,responses):
keys=[key for key,_ in responses]
if len(keys)!=len(set(keys)):
raise ValueError('duplicate embedding id')
by_id=dict(responses)
missing={key for key,_ in requests}-set(by_id)
if missing:
raise ValueError('missing embedding ids: '+','.join(sorted(missing)))
return [(key,by_id[key]) for key,_ in requests]
requests=[('a','first'),('b','second')]
assert join(requests,[('b',[0.,1.]),('a',[1.,0.])])[0]==('a',[1.,0.])
try:
join(requests,[('b',[0.,1.])])
except ValueError as e:
assert str(e)=='missing embedding ids: a'
else:
raise AssertionError('partial index accepted')
print('missing id prevents publication')实际批任务可单独重试失败项,但发布的索引版本必须满足其完整性契约;不能把后一个成功向量错配到前一个文档。
10|向量数据库
以下完整文件保存 answer.py,Python 3.11+ 运行 python3 answer.py,先重建正文初始状态,再检验失败不改变数据。
from math import isfinite
class Store:
def __init__(self, dimension):
self.dimension = dimension
self.rows = {}
def vector(self, value):
if len(value) != self.dimension or not all(isinstance(x,(int,float)) and isfinite(x) for x in value):
raise ValueError('invalid vector')
return list(value)
def upsert(self, tenant, doc_id, vector, text):
self.rows[(tenant,doc_id)] = {'vector':self.vector(vector),'text':text}
def delete(self, tenant, doc_id):
self.rows.pop((tenant,doc_id),None)
def search(self, tenant, query, k):
query = self.vector(query)
if k < 0:
raise ValueError('negative k')
candidates = []
for (owner,doc_id),row in self.rows.items():
if owner == tenant:
distance = sum((a-b)**2 for a,b in zip(query,row['vector']))
candidates.append((distance,doc_id,row['text']))
return sorted(candidates)[:k]
s = Store(2)
s.upsert('A','1',[1.,0.],'old')
s.upsert('B','1',[1.,0.],'secret')
s.upsert('A','1',[0.,1.],'new')
assert s.search('A',[0.,1.],1)==[(0.,'1','new')]
s.delete('A','1'); s.delete('A','1')
assert s.search('A',[1.,0.],10)==[]
assert s.search('B',[1.,0.],1)==[(0.,'1','secret')]
print('CRUD contract passed')
from hashlib import sha256
def shard(key, count):
if count <= 0:
raise ValueError('count')
return int.from_bytes(sha256(key.encode()).digest()[:8],'big') % count
assert shard('tenant-A',3)==shard('tenant-A',3)
assert 0 <= shard('tenant-A',3) < 3
print(shard('tenant-A',3))
before = dict(s.rows)
try:
s.upsert('A','bad',[1.],'bad')
except ValueError:
pass
else:
raise AssertionError('bad dimension accepted')
assert s.rows == before
moved=sum(shard(str(i),3)!=shard(str(i),4) for i in range(100))
assert 0 < moved < 100
assert s.search('B',[1.,0.],1)==[(0.,'1','secret')]
print('moved',moved)此固定样本的搬迁数量只说明取模扩容会重映射;不代表一致性哈希或实际数据库分片性能。失败写入前先校验维数,确保没有部分更新。
答案册验收方式
- 自己的结论与上述方向不同时,保留能够复现的反例;证据优先于“标准答案”。
- 代码输出、依赖版本、随机种子和数据样本应能让另一位学习者复现。
- 涉及真实模型、外部数据或工具副作用时,同时记录成本、权限、超时与清理责任。