让同一个检查任务走完
思想篇里最值得亲手试的一幕，是程序已经输出42，循环却仍返回call_limit。先运行公开例子，再回来看停止条件所在的位置。
源码在第二十课，核心兼容Python 3.10+，本机验证于2026-10-20。默认标准库路径使用固定模型与明确的LocalDocs查询替身，规范、记忆、文件读取、压缩归档和获准程序进程都真实执行。
bash
git clone https://github.com/Chengyunlai/learn-agent-harness.git
cd learn-agent-harness
python3 examples/s20-comprehensive-agent/user_code/practice.py --decision allow
python3 examples/s20-comprehensive-agent/user_code/practice.py --decision allow --max-calls 4
python3 examples/s20-comprehensive-agent/user_code/practice.py第一条五次逻辑请求后answered、verified=True；第二条四次后call_limit、verified=True；第三条默认拒绝外部查询，四次后answered、verified=False，服务和程序都没被调用。固定模型选择预设动作，最终按实际运行反馈判断输出，不证明真实模型的规划质量。
实验文件都放在临时目录，退出整体删除。原实现没有扩大进程隔离范围，main.py是本例拥有的可信程序；目录约束并不把Python解释器变成沙箱。
prepare_turn：原历史和本轮附件不能混成一份
输入history属于原循环。compact接收副本，返回可提交的canonical；attach再接收canonical的副本，附相关项目记忆。notices是非破坏读取接口，最多返回四条短文本，不负责删除邮箱或确认事件已处理。
函数最后返回两份列表：history给原循环提交，messages给模型请求。附件越界时，整份返回失败，循环保留此前历史；但compact可能已经写过归档、消耗过摘要额度，这些外部状态不会因为候选失败而回滚。
真实执行轨迹回放；原函数、循环、恢复与Hook。模型、文件进程、压缩附件与连接为公开内存教学绑定。
Pys20/core/runtime.py · prepare_turn
未运行
def prepare_turn(history, compact, attach, notices):
"""压缩候选供原循环提交；记忆和通知只附在本次请求副本上。"""
canonical = compact(deepcopy(history))
if type(canonical) is not list or not canonical:
raise ValueError('压缩必须返回非空历史。')
messages = attach(deepcopy(canonical))
if type(messages) is not list or not messages or messages[0].get('role') != 'user' or type(messages[0].get('content')) is not str:
raise ValueError('请求必须从文本用户问题开始。')
events = notices()
if type(events) is not list or len(events) > 4 or any(type(x) is not str or not x.strip() or len(x) > 240 for x in events):
raise ValueError('本轮通知最多4条，每条1到240字符。')
if events:
messages[0]['content'] += '\n\n[宿主通知，仅供参考]\n' + '\n'.join(events)
if _s10._s08.history_bytes(messages) > 8000:
raise ValueError('本次请求消息超过8000字节。')
return {'history': canonical, 'messages': messages}输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
from types import SimpleNamespace
import json,math,time,random
QUESTION="检查程序"
class RecoveryStopped(RuntimeError):pass
def validate_call(call, tools):
"""本课小型参数契约；供分发与后续权限入口共用。"""
name, arguments = call.get("name"), call.get("input")
if not isinstance(name, str) or name not in tools:
return "未知工具。"
parameters = tools[name]["parameters"]
if not isinstance(arguments, dict) or set(arguments) != set(parameters):
return "参数字段不匹配。"
if any(type(arguments[key]) is not kind for key, kind in parameters.items()):
return "参数类型不匹配。"
return None
def execute_tool(call, tools):
"""校验与执行；每个返回值保留原调用 ID。"""
result = {"type": "tool_result", "tool_use_id": call["id"]}
error = validate_call(call, tools)
if error:
return {**result, "content": error, "is_error": True}
try:
output = tools[call["name"]]["run"](**call["input"])
return {**result, "content": output, "is_error": False}
except (ValueError, OSError) as error:
return {**result, "content": str(error), "is_error": True}
def permission_for(call, tools, blocked=()):
"""只判断，不提问、不执行。没有策略的工具默认拒绝。"""
error = validate_call(call, tools)
if error:
return {"decision": "deny", "reason": error}
name = call["name"]
if name in blocked:
return {"decision": "deny", "reason": "宿主策略禁止这个工具。"}
if name == "read_file":
return {"decision": "allow", "reason": "允许读取；目录约束仍由读取函数检查。"}
if name == "run_main":
return {"decision": "ask", "reason": "运行 main.py 前需要用户允许。"}
return {"decision": "deny", "reason": "这个工具尚未配置权限策略。"}
def authorize_call(call, tools, ask_user, blocked=()):
"""只有 ask 才调用宿主；必须收到布尔 True 才允许。"""
verdict = permission_for(call, tools, blocked)
if verdict["decision"] != "ask":
return verdict
# 宿主看到调用副本；不能通过修改显示对象改变后续执行参数。
answer = ask_user(deepcopy(call), verdict["reason"])
if answer is True:
return {"decision": "allow", "reason": "用户允许本次调用。"}
return {"decision": "deny", "reason": "本次调用未获用户允许。"}
def execute_guarded(call, tools, ask_user, blocked=()):
"""验证、显示与执行使用同一份调用快照；拒绝结果不冒充执行结果。"""
snapshot = deepcopy(call)
verdict = authorize_call(snapshot, tools, ask_user, blocked)
if verdict["decision"] != "allow":
return {
"type": "tool_result", "tool_use_id": snapshot["id"],
"content": "未执行：" + verdict["reason"], "is_error": True,
}
return execute_tool(snapshot, tools)
def run_hooks(hooks, event, payload):
"""before 按顺序遇阻即停；after 故障只记警告，并继续其他观察者。"""
if event not in ("before", "after"):
raise ValueError("未知 hook 事件")
warnings = []
for index, callback in enumerate(tuple(hooks.get(event, ()))):
try:
reason = callback(deepcopy(payload))
if reason is not None:
if event != "before" or not isinstance(reason, str) or not reason.strip():
raise ValueError("before 只能返回 None 或非空拒绝理由；after 必须返回 None")
return {"blocked": reason, "warnings": warnings}
except Exception as error:
message = type(error).__name__ + ": " + str(error)
if event == "before":
return {"blocked": "前置检查异常：" + message, "warnings": warnings}
warnings.append(message)
return {"blocked": None, "warnings": warnings}
def execute_hooked(call, tools, ask_user, hooks, blocked=(), guard=None):
"""包住上一课权限入口；后置观察故障不覆盖实际结果，也不重试工具。"""
snapshot = deepcopy(call)
before = run_hooks(hooks, "before", {"call": snapshot})
if before["blocked"] is not None:
result = {"type": "tool_result", "tool_use_id": snapshot["id"],
"content": "未执行：" + before["blocked"], "is_error": True}
else:
result = (guard or execute_guarded)(snapshot, tools, ask_user, blocked)
after = run_hooks(hooks, "after", {"call": snapshot, "result": result})
return {"result": result, "warnings": after["warnings"]}
def result_text(reply):
if not isinstance(reply, dict) or type(reply.get('isError',False)) is not bool:
raise ValueError('MCP结果格式不合法。')
blocks = reply.get('content')
if not isinstance(blocks, list) or not blocks or len(blocks) > 4 or reply.get('structuredContent') is not None:
raise ValueError('本课只接收一到四块纯文本结果。')
if any(not isinstance(b,dict) or b.get('type') != 'text' or not isinstance(b.get('text'),str) for b in blocks):
raise ValueError('结果包含本课不支持的内容类型。')
text = '\n'.join(block['text'] for block in blocks)
if not text.strip() or len(text) > 2000:
raise ValueError('外部结果为空或超过2000字符。')
return text, reply.get('isError', False)
def call_external(call, pool, ask_user):
"""宿主确认独立于server标注；一次调用，无自动工具重试。"""
snapshot = deepcopy(call)
envelope = {'type':'tool_result', 'tool_use_id':snapshot['id']}
try:
route = pool['routes'].get(snapshot.get('name'))
if route is None:
raise ValueError('本次工具快照中没有这个路由。')
arguments = snapshot.get('input')
if (not isinstance(arguments,dict) or set(arguments) != set(route['schema']['properties'])
or any(type(value) is not str or len(value) > 120 for value in arguments.values())):
raise ValueError('外部工具参数字段、类型或长度不符。')
if ask_user(deepcopy(snapshot), route['server']) is not True:
return {**envelope, 'is_error':True, 'content':'未执行：本次外部调用未获批准。'}
reply = route['invoke'](route['remote'], deepcopy(arguments))
text, failed = result_text(reply)
return {**envelope, 'is_error':failed, 'content':text}
except Exception as error:
return {**envelope, 'is_error':True, 'content':'外部调用未能完成确认：' + type(error).__name__ + ': ' + str(error)[:180]}
def history_bytes(messages):
return len(json.dumps(messages, ensure_ascii=False, separators=(',', ':')).encode('utf-8'))
def checked_state(state):
if type(state) is not dict or set(state) != {'remaining', 'waited', 'events'}:
raise ValueError('恢复状态必须由宿主持有。')
if type(state['remaining']) is not int or not 0 <= state['remaining'] <= 20 or type(state['waited']) not in (int, float) or not math.isfinite(state['waited']) or not 0 <= state['waited'] <= 3 or type(state['events']) is not list or len(state['events']) > 20:
raise ValueError('请求或等待额度无效。')
def classify_failure(outcome):
"""接收适配器的基本数据，不匹配任意SDK异常文字。"""
if type(outcome) is not dict:
return {'action': 'stop', 'reason': 'invalid_response'}
if 'failure' in outcome:
failure = outcome['failure']
if type(failure) is not dict:
return {'action': 'stop', 'reason': 'invalid_failure'}
status = failure.get('status')
if type(status) is int and status in (429, 500, 502, 503, 504, 529):
return {'action': 'wait', 'reason': str(status), 'retry_after': failure.get('retry_after')}
if failure.get('transport') in ('connection', 'timeout') and status is None:
return {'action': 'wait', 'reason': failure['transport']}
return {'action': 'stop', 'reason': 'request_error:' + str(status)}
reason = outcome.get('stop_reason')
if reason == 'max_tokens':
return {'action': 'expand', 'reason': 'max_tokens'}
if reason in ('end_turn', 'tool_use') and type(outcome.get('content')) is list and outcome['content']:
return {'action': 'accept', 'reason': reason}
return {'action': 'stop', 'reason': 'unsupported_response:' + str(reason)}
def retry_delay(attempt, retry_after=None, jitter=0.0):
"""优先有效秒数；无效/日期格式回退，过长等待由调用者停下而非缩短。"""
if type(attempt) is not int or not 0 <= attempt < 4:
raise ValueError('尝试序号必须在0到3之间。')
if type(jitter) not in (int, float) or not math.isfinite(jitter) or not 0 <= jitter <= 1:
raise ValueError('抖动比例必须在0到1之间。')
seconds = None
if type(retry_after) in (str, int, float):
try:
value = float(retry_after)
if math.isfinite(value) and value >= 0:
seconds = value
except ValueError:
pass
if seconds is not None:
return seconds
base = min(0.25 * 2 ** attempt, 2.0)
return base + base * 0.25 * jitter
def request_with_recovery(send, request, state, pause=time.sleep, random_unit=random.random):
"""每次逻辑请求最多4次尝试；父会话共享20次请求/3秒计划等待额度。"""
checked_state(state)
pending = deepcopy(request)
if pending.get('max_tokens') != 1024 or type(pending.get('max_tokens')) is not int:
raise ValueError('教学输出预算从1024开始。')
expanded = False
for attempt in range(1, 5):
if state['remaining'] == 0:
raise RecoveryStopped('request_budget')
state['remaining'] -= 1
outcome = send(deepcopy(pending))
decision = classify_failure(outcome)
state['events'].append({'attempt': attempt, 'action': decision['action'], 'reason': decision['reason']})
if decision['action'] == 'accept':
return outcome
if decision['action'] == 'stop':
raise RecoveryStopped(decision['reason'])
if attempt == 4 or state['remaining'] == 0:
raise RecoveryStopped('request_budget')
if decision['action'] == 'expand':
if expanded:
raise RecoveryStopped('output_still_truncated')
pending['max_tokens'] = 2048
expanded = True
continue
delay = retry_delay(attempt - 1, decision.get('retry_after'), random_unit())
if state['waited'] + delay > 3:
raise RecoveryStopped('wait_budget')
state['waited'] += delay
state['events'][-1]['delay'] = delay
pause(delay)
raise RecoveryStopped('request_budget')
def run_loop(ask, execute, max_calls=4, emit=print, question=QUESTION, prepare=None):
"""ask 读完整记录，execute 读工具请求；返回状态和证据，避免把停下当作成功。"""
if max_calls < 1:
raise ValueError("max_calls 必须至少为 1")
messages = [{"role": "user", "content": question}]
def finish(status, text, calls):
return {"status": status, "text": text, "calls": calls, "messages": messages}
for number in range(1, max_calls + 1):
if prepare is not None:
try:
candidate = prepare(deepcopy(messages))
if type(candidate) is not list:
raise ValueError("prepare必须返回候选消息列表")
except Exception as error:
detail = str(error)[:160] if isinstance(error, ValueError) else type(error).__name__
return finish("context_error", f"上下文准备失败：{detail}；原历史保留。", number - 1)
messages[:] = candidate
emit(f"模型调用 {number}")
try:
response = ask(messages)
except Exception as error:
# API 异常时交回控制权；不自动重试，也不伪造工具结果。
return finish("model_error", f"模型请求失败：{type(error).__name__}", number)
blocks = response.get("content")
reason = response.get("stop_reason")
if not isinstance(blocks, list) or any(
not isinstance(block, dict) or block.get("type") not in ("text", "tool_use")
for block in blocks
):
return finish("protocol_error", "响应不符合本课支持的格式。", number)
messages.append({"role": "assistant", "content": blocks})
calls = [block for block in blocks if block["type"] == "tool_use"]
text = "\n".join(block.get("text", "") for block in blocks if block["type"] == "text")
if reason == "end_turn" and not calls and text.strip():
return finish("answered", text, number)
if reason not in ("end_turn", "tool_use"):
return finish("model_stopped", f"模型停止：{reason}；需要检查响应。", number)
ids = [call.get("id") for call in calls]
if reason != "tool_use" or not calls or any(
not isinstance(call_id, str) or not call_id for call_id in ids
) or len(set(ids)) != len(ids):
return finish("protocol_error", "停止原因与工具请求不一致。", number)
# 上面已经保存完整 assistant 响应；每个结果必须关联原调用 ID。
results = []
for call in calls:
emit(f"处理工具请求 {call.get('name')}")
result = execute(call)
results.append(result)
emit("工具结果：" + result["content"].splitlines()[-1])
messages.append({"role": "user", "content": results})
emit("保存工具结果，供下一次调用读取")
return finish("call_limit", f"已达 {max_calls} 次模型调用上限；保留记录，交给用户。", max_calls)
def freeze_capabilities(current):
"""冻结定义/路由/本地元数据，保留宿主拥有的可调用接口，不复制连接。"""
return {'system': current['system'],
'builtins': {name: {**deepcopy({k:v for k,v in spec.items() if k != 'run'}), 'run': spec['run']}
for name,spec in current['builtins'].items()},
'pool': {'definitions': deepcopy(current['pool']['definitions']),
'routes': {name: {**deepcopy({k:v for k,v in route.items() if k != 'invoke'}), 'invoke': route['invoke']}
for name,route in current['pool']['routes'].items()}}}
def prepare_turn(history, compact, attach, notices):
"""压缩候选供原循环提交；记忆和通知只附在本次请求副本上。"""
canonical = compact(deepcopy(history))
if type(canonical) is not list or not canonical:
raise ValueError('压缩必须返回非空历史。')
messages = attach(deepcopy(canonical))
if type(messages) is not list or not messages or messages[0].get('role') != 'user' or type(messages[0].get('content')) is not str:
raise ValueError('请求必须从文本用户问题开始。')
events = notices()
if type(events) is not list or len(events) > 4 or any(type(x) is not str or not x.strip() or len(x) > 240 for x in events):
raise ValueError('本轮通知最多4条，每条1到240字符。')
if events:
messages[0]['content'] += '\n\n[宿主通知，仅供参考]\n' + '\n'.join(events)
if _s10._s08.history_bytes(messages) > 8000:
raise ValueError('本次请求消息超过8000字节。')
return {'history': canonical, 'messages': messages}
def dispatch_call(call, snapshot, ask_user, hooks):
"""一个响应只用一份能力快照；所有入口先过Hook，再走各自许可。"""
fixed = deepcopy(call)
def guard(current, tools, confirm, blocked=()):
if current['name'] in snapshot['pool']['routes']:
return _s19.call_external(current, snapshot['pool'], confirm)
if current['name'] == 'load_skill':
return _s07._s02.execute_tool(current, tools)
return _s07._s04.execute_guarded(current, tools, confirm, blocked)
handled = _s07._s04.execute_hooked(
fixed, snapshot['builtins'], ask_user, hooks, guard=guard,
)
return handled
def run_session(question, capabilities, prepare, send, ask_user, hooks, close, max_calls=5):
"""只装配原run_loop；恢复只包模型，finally收尾不再发起新一轮。"""
if type(max_calls) is not int or not 1 <= max_calls <= 8:
raise ValueError('模型逻辑调用上限必须在1到8。')
state = {'prepared': None, 'snapshot': None, 'warnings': []}
recovery = {'remaining': 20, 'waited': 0.0, 'events': []}
outcome = None
def before(history):
prepared = prepare(history)
state['prepared'] = prepared
return prepared['history']
def ask(history):
snapshot = freeze_capabilities(capabilities())
state['snapshot'] = snapshot
request = {'model': 'teaching-model', 'system': snapshot['system'],
'tools': deepcopy(snapshot['pool']['definitions']),
'messages': deepcopy(state['prepared']['messages']), 'max_tokens': 1024}
return _s11.request_with_recovery(send, request, recovery)
def execute(call):
handled = dispatch_call(call, state['snapshot'], ask_user, hooks)
state['warnings'].extend(handled['warnings'])
return handled['result']
try:
outcome = run_loop(ask, execute, max_calls=max_calls, emit=lambda _: None,
question=question, prepare=before)
except Exception as error:
outcome = {'status': 'host_error', 'text': type(error).__name__, 'calls': None, 'messages': None}
finally:
try:
close()
except Exception as error:
state['warnings'].append('收尾失败：' + type(error).__name__)
return {'outcome': outcome, 'warnings': state['warnings'],
'recovery': recovery}
_s10=SimpleNamespace(_s08=SimpleNamespace(history_bytes=history_bytes))
_s07=SimpleNamespace(_s02=SimpleNamespace(execute_tool=execute_tool),_s04=SimpleNamespace(execute_guarded=execute_guarded,execute_hooked=execute_hooked))
_s19=SimpleNamespace(call_external=call_external)
_s11=SimpleNamespace(request_with_recovery=request_with_recovery)
scenario='attached'
history=[{'role':'user','content':'检查main.py，核对42。'}];original=deepcopy(history)
notices_log=[];invoked=[];closed=[];sent=[];approvals=[]
def compact(history):
if scenario in ('compact_error','context_error'):raise ValueError('归档失败')
return deepcopy(history)
def attach(messages):
if scenario!='empty':messages[0]['content']+='\n[项目记忆] 按本次输出核对。'
if scenario=='oversize':messages[0]['content']+='x'*8001
return messages
def notices():return ['x'*241] if scenario=='bad_notices' else ([] if scenario=='empty' else ['收到检查任务。'])
def program():invoked.append('run_main');return 'exit_code=0\n42'
def remote(name,args):invoked.append(name);return {'content':[{'type':'text','text':'需要本次许可。'}],'isError':False}
def confirm(*_):approvals.append('asked');return scenario!='denied'
def before(_):return '前置阻止' if scenario=='blocked' else None
def after(_):
if scenario=='warning':raise ValueError('观察故障')
tools={'run_main':{'description':'固定运行main.py','parameters':{},'run':program}}
pool={'definitions':[{'name':'run_main','description':'固定运行main.py','input_schema':{'type':'object','properties':{},'additionalProperties':False}}],
'routes':{'mcp__docs__search_docs':{'server':'docs','remote':'search.docs','schema':{'properties':{'query':{'type':'string'}}},'invoke':remote}}}
snapshot={'builtins':tools,'pool':pool,'system':'按本次证据核对。'}
call={'type':'tool_use','id':'step_1','name':'mcp__docs__search_docs' if scenario=='external' else 'missing' if scenario=='unknown' else 'run_main','input':{'query':'运行许可'} if scenario=='external' else {}}
hooks={'before':[before],'after':[after]}
def capabilities():return snapshot
def prepare(history):return prepare_turn(history,compact,attach,notices)
def send(request):
sent.append(deepcopy(request))
if scenario=='model_error':raise ValueError('模型接口故障')
if scenario=='retry' and len(sent)==2:return {'failure':{'status':529,'retry_after':'0'}}
if len(sent)==1:return {'stop_reason':'tool_use','content':[call]}
return {'stop_reason':'end_turn','content':[{'type':'text','text':'收到实际工具反馈。'}]}
def close():
closed.append('closed')
if scenario=='close_error':raise OSError('关闭故障')
# 调用入口
try:
observed=prepare_turn(history,compact,attach,notices)
except Exception as error:
observed={"error":str(error)}
print(observed)
- 压缩完整历史候选。
- 在请求副本附记忆和通知，校验后分别交出。
在messages赋值前暂停，canonical已经生成；单步后才有带记忆的请求副本。停在通知追加行，原历史仍没有通知。选择“通知超限”或“请求超限”，观察候选怎样停下，而不是截掉半句继续发模型。
本课最终消息预算是8000个UTF-8字节，通知最多四条、每条240字符，都是教学选择。s08仍负责压缩完整工具交接；没有把它改成可任意混入通知的通用历史处理器。
实际例子没有后台工作，通知接口返回空列表。教学绑定增加一条明确的内存通知帮助观察附件位置。生产宿主要先保存事件，再设计何时确认；如果在notices里先pop队列，后续请求失败就可能把未处理通知丢掉。
dispatch_call：共用入口，保留各自的执行策略
输入是一个模型调用、该响应冻结的snapshot、宿主确认接口和Hook。snapshot同时含本地注册和外部pool，外部pool又把definitions与routes放在一起。
分发器先复制调用，随后进入原s04.execute_hooked。before检查通过后，guard选择外部MCP、显式只读的load_skill，或原本地权限入口。这个顺序让新增工具也经过同一组Hook，同时保留各自的参数与许可规则。
真实执行轨迹回放；原函数、循环、恢复与Hook。模型、文件进程、压缩附件与连接为公开内存教学绑定。
Pys20/core/runtime.py · dispatch_call
未运行
def dispatch_call(call, snapshot, ask_user, hooks):
"""一个响应只用一份能力快照；所有入口先过Hook，再走各自许可。"""
fixed = deepcopy(call)
def guard(current, tools, confirm, blocked=()):
if current['name'] in snapshot['pool']['routes']:
return _s19.call_external(current, snapshot['pool'], confirm)
if current['name'] == 'load_skill':
return _s07._s02.execute_tool(current, tools)
return _s07._s04.execute_guarded(current, tools, confirm, blocked)
handled = _s07._s04.execute_hooked(
fixed, snapshot['builtins'], ask_user, hooks, guard=guard,
)
return handled输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
from types import SimpleNamespace
import json,math,time,random
QUESTION="检查程序"
class RecoveryStopped(RuntimeError):pass
def validate_call(call, tools):
"""本课小型参数契约；供分发与后续权限入口共用。"""
name, arguments = call.get("name"), call.get("input")
if not isinstance(name, str) or name not in tools:
return "未知工具。"
parameters = tools[name]["parameters"]
if not isinstance(arguments, dict) or set(arguments) != set(parameters):
return "参数字段不匹配。"
if any(type(arguments[key]) is not kind for key, kind in parameters.items()):
return "参数类型不匹配。"
return None
def execute_tool(call, tools):
"""校验与执行；每个返回值保留原调用 ID。"""
result = {"type": "tool_result", "tool_use_id": call["id"]}
error = validate_call(call, tools)
if error:
return {**result, "content": error, "is_error": True}
try:
output = tools[call["name"]]["run"](**call["input"])
return {**result, "content": output, "is_error": False}
except (ValueError, OSError) as error:
return {**result, "content": str(error), "is_error": True}
def permission_for(call, tools, blocked=()):
"""只判断，不提问、不执行。没有策略的工具默认拒绝。"""
error = validate_call(call, tools)
if error:
return {"decision": "deny", "reason": error}
name = call["name"]
if name in blocked:
return {"decision": "deny", "reason": "宿主策略禁止这个工具。"}
if name == "read_file":
return {"decision": "allow", "reason": "允许读取；目录约束仍由读取函数检查。"}
if name == "run_main":
return {"decision": "ask", "reason": "运行 main.py 前需要用户允许。"}
return {"decision": "deny", "reason": "这个工具尚未配置权限策略。"}
def authorize_call(call, tools, ask_user, blocked=()):
"""只有 ask 才调用宿主；必须收到布尔 True 才允许。"""
verdict = permission_for(call, tools, blocked)
if verdict["decision"] != "ask":
return verdict
# 宿主看到调用副本；不能通过修改显示对象改变后续执行参数。
answer = ask_user(deepcopy(call), verdict["reason"])
if answer is True:
return {"decision": "allow", "reason": "用户允许本次调用。"}
return {"decision": "deny", "reason": "本次调用未获用户允许。"}
def execute_guarded(call, tools, ask_user, blocked=()):
"""验证、显示与执行使用同一份调用快照；拒绝结果不冒充执行结果。"""
snapshot = deepcopy(call)
verdict = authorize_call(snapshot, tools, ask_user, blocked)
if verdict["decision"] != "allow":
return {
"type": "tool_result", "tool_use_id": snapshot["id"],
"content": "未执行：" + verdict["reason"], "is_error": True,
}
return execute_tool(snapshot, tools)
def run_hooks(hooks, event, payload):
"""before 按顺序遇阻即停；after 故障只记警告，并继续其他观察者。"""
if event not in ("before", "after"):
raise ValueError("未知 hook 事件")
warnings = []
for index, callback in enumerate(tuple(hooks.get(event, ()))):
try:
reason = callback(deepcopy(payload))
if reason is not None:
if event != "before" or not isinstance(reason, str) or not reason.strip():
raise ValueError("before 只能返回 None 或非空拒绝理由；after 必须返回 None")
return {"blocked": reason, "warnings": warnings}
except Exception as error:
message = type(error).__name__ + ": " + str(error)
if event == "before":
return {"blocked": "前置检查异常：" + message, "warnings": warnings}
warnings.append(message)
return {"blocked": None, "warnings": warnings}
def execute_hooked(call, tools, ask_user, hooks, blocked=(), guard=None):
"""包住上一课权限入口；后置观察故障不覆盖实际结果，也不重试工具。"""
snapshot = deepcopy(call)
before = run_hooks(hooks, "before", {"call": snapshot})
if before["blocked"] is not None:
result = {"type": "tool_result", "tool_use_id": snapshot["id"],
"content": "未执行：" + before["blocked"], "is_error": True}
else:
result = (guard or execute_guarded)(snapshot, tools, ask_user, blocked)
after = run_hooks(hooks, "after", {"call": snapshot, "result": result})
return {"result": result, "warnings": after["warnings"]}
def result_text(reply):
if not isinstance(reply, dict) or type(reply.get('isError',False)) is not bool:
raise ValueError('MCP结果格式不合法。')
blocks = reply.get('content')
if not isinstance(blocks, list) or not blocks or len(blocks) > 4 or reply.get('structuredContent') is not None:
raise ValueError('本课只接收一到四块纯文本结果。')
if any(not isinstance(b,dict) or b.get('type') != 'text' or not isinstance(b.get('text'),str) for b in blocks):
raise ValueError('结果包含本课不支持的内容类型。')
text = '\n'.join(block['text'] for block in blocks)
if not text.strip() or len(text) > 2000:
raise ValueError('外部结果为空或超过2000字符。')
return text, reply.get('isError', False)
def call_external(call, pool, ask_user):
"""宿主确认独立于server标注；一次调用，无自动工具重试。"""
snapshot = deepcopy(call)
envelope = {'type':'tool_result', 'tool_use_id':snapshot['id']}
try:
route = pool['routes'].get(snapshot.get('name'))
if route is None:
raise ValueError('本次工具快照中没有这个路由。')
arguments = snapshot.get('input')
if (not isinstance(arguments,dict) or set(arguments) != set(route['schema']['properties'])
or any(type(value) is not str or len(value) > 120 for value in arguments.values())):
raise ValueError('外部工具参数字段、类型或长度不符。')
if ask_user(deepcopy(snapshot), route['server']) is not True:
return {**envelope, 'is_error':True, 'content':'未执行：本次外部调用未获批准。'}
reply = route['invoke'](route['remote'], deepcopy(arguments))
text, failed = result_text(reply)
return {**envelope, 'is_error':failed, 'content':text}
except Exception as error:
return {**envelope, 'is_error':True, 'content':'外部调用未能完成确认：' + type(error).__name__ + ': ' + str(error)[:180]}
def history_bytes(messages):
return len(json.dumps(messages, ensure_ascii=False, separators=(',', ':')).encode('utf-8'))
def checked_state(state):
if type(state) is not dict or set(state) != {'remaining', 'waited', 'events'}:
raise ValueError('恢复状态必须由宿主持有。')
if type(state['remaining']) is not int or not 0 <= state['remaining'] <= 20 or type(state['waited']) not in (int, float) or not math.isfinite(state['waited']) or not 0 <= state['waited'] <= 3 or type(state['events']) is not list or len(state['events']) > 20:
raise ValueError('请求或等待额度无效。')
def classify_failure(outcome):
"""接收适配器的基本数据，不匹配任意SDK异常文字。"""
if type(outcome) is not dict:
return {'action': 'stop', 'reason': 'invalid_response'}
if 'failure' in outcome:
failure = outcome['failure']
if type(failure) is not dict:
return {'action': 'stop', 'reason': 'invalid_failure'}
status = failure.get('status')
if type(status) is int and status in (429, 500, 502, 503, 504, 529):
return {'action': 'wait', 'reason': str(status), 'retry_after': failure.get('retry_after')}
if failure.get('transport') in ('connection', 'timeout') and status is None:
return {'action': 'wait', 'reason': failure['transport']}
return {'action': 'stop', 'reason': 'request_error:' + str(status)}
reason = outcome.get('stop_reason')
if reason == 'max_tokens':
return {'action': 'expand', 'reason': 'max_tokens'}
if reason in ('end_turn', 'tool_use') and type(outcome.get('content')) is list and outcome['content']:
return {'action': 'accept', 'reason': reason}
return {'action': 'stop', 'reason': 'unsupported_response:' + str(reason)}
def retry_delay(attempt, retry_after=None, jitter=0.0):
"""优先有效秒数；无效/日期格式回退，过长等待由调用者停下而非缩短。"""
if type(attempt) is not int or not 0 <= attempt < 4:
raise ValueError('尝试序号必须在0到3之间。')
if type(jitter) not in (int, float) or not math.isfinite(jitter) or not 0 <= jitter <= 1:
raise ValueError('抖动比例必须在0到1之间。')
seconds = None
if type(retry_after) in (str, int, float):
try:
value = float(retry_after)
if math.isfinite(value) and value >= 0:
seconds = value
except ValueError:
pass
if seconds is not None:
return seconds
base = min(0.25 * 2 ** attempt, 2.0)
return base + base * 0.25 * jitter
def request_with_recovery(send, request, state, pause=time.sleep, random_unit=random.random):
"""每次逻辑请求最多4次尝试；父会话共享20次请求/3秒计划等待额度。"""
checked_state(state)
pending = deepcopy(request)
if pending.get('max_tokens') != 1024 or type(pending.get('max_tokens')) is not int:
raise ValueError('教学输出预算从1024开始。')
expanded = False
for attempt in range(1, 5):
if state['remaining'] == 0:
raise RecoveryStopped('request_budget')
state['remaining'] -= 1
outcome = send(deepcopy(pending))
decision = classify_failure(outcome)
state['events'].append({'attempt': attempt, 'action': decision['action'], 'reason': decision['reason']})
if decision['action'] == 'accept':
return outcome
if decision['action'] == 'stop':
raise RecoveryStopped(decision['reason'])
if attempt == 4 or state['remaining'] == 0:
raise RecoveryStopped('request_budget')
if decision['action'] == 'expand':
if expanded:
raise RecoveryStopped('output_still_truncated')
pending['max_tokens'] = 2048
expanded = True
continue
delay = retry_delay(attempt - 1, decision.get('retry_after'), random_unit())
if state['waited'] + delay > 3:
raise RecoveryStopped('wait_budget')
state['waited'] += delay
state['events'][-1]['delay'] = delay
pause(delay)
raise RecoveryStopped('request_budget')
def run_loop(ask, execute, max_calls=4, emit=print, question=QUESTION, prepare=None):
"""ask 读完整记录，execute 读工具请求；返回状态和证据，避免把停下当作成功。"""
if max_calls < 1:
raise ValueError("max_calls 必须至少为 1")
messages = [{"role": "user", "content": question}]
def finish(status, text, calls):
return {"status": status, "text": text, "calls": calls, "messages": messages}
for number in range(1, max_calls + 1):
if prepare is not None:
try:
candidate = prepare(deepcopy(messages))
if type(candidate) is not list:
raise ValueError("prepare必须返回候选消息列表")
except Exception as error:
detail = str(error)[:160] if isinstance(error, ValueError) else type(error).__name__
return finish("context_error", f"上下文准备失败：{detail}；原历史保留。", number - 1)
messages[:] = candidate
emit(f"模型调用 {number}")
try:
response = ask(messages)
except Exception as error:
# API 异常时交回控制权；不自动重试，也不伪造工具结果。
return finish("model_error", f"模型请求失败：{type(error).__name__}", number)
blocks = response.get("content")
reason = response.get("stop_reason")
if not isinstance(blocks, list) or any(
not isinstance(block, dict) or block.get("type") not in ("text", "tool_use")
for block in blocks
):
return finish("protocol_error", "响应不符合本课支持的格式。", number)
messages.append({"role": "assistant", "content": blocks})
calls = [block for block in blocks if block["type"] == "tool_use"]
text = "\n".join(block.get("text", "") for block in blocks if block["type"] == "text")
if reason == "end_turn" and not calls and text.strip():
return finish("answered", text, number)
if reason not in ("end_turn", "tool_use"):
return finish("model_stopped", f"模型停止：{reason}；需要检查响应。", number)
ids = [call.get("id") for call in calls]
if reason != "tool_use" or not calls or any(
not isinstance(call_id, str) or not call_id for call_id in ids
) or len(set(ids)) != len(ids):
return finish("protocol_error", "停止原因与工具请求不一致。", number)
# 上面已经保存完整 assistant 响应；每个结果必须关联原调用 ID。
results = []
for call in calls:
emit(f"处理工具请求 {call.get('name')}")
result = execute(call)
results.append(result)
emit("工具结果：" + result["content"].splitlines()[-1])
messages.append({"role": "user", "content": results})
emit("保存工具结果，供下一次调用读取")
return finish("call_limit", f"已达 {max_calls} 次模型调用上限；保留记录，交给用户。", max_calls)
def freeze_capabilities(current):
"""冻结定义/路由/本地元数据，保留宿主拥有的可调用接口，不复制连接。"""
return {'system': current['system'],
'builtins': {name: {**deepcopy({k:v for k,v in spec.items() if k != 'run'}), 'run': spec['run']}
for name,spec in current['builtins'].items()},
'pool': {'definitions': deepcopy(current['pool']['definitions']),
'routes': {name: {**deepcopy({k:v for k,v in route.items() if k != 'invoke'}), 'invoke': route['invoke']}
for name,route in current['pool']['routes'].items()}}}
def prepare_turn(history, compact, attach, notices):
"""压缩候选供原循环提交；记忆和通知只附在本次请求副本上。"""
canonical = compact(deepcopy(history))
if type(canonical) is not list or not canonical:
raise ValueError('压缩必须返回非空历史。')
messages = attach(deepcopy(canonical))
if type(messages) is not list or not messages or messages[0].get('role') != 'user' or type(messages[0].get('content')) is not str:
raise ValueError('请求必须从文本用户问题开始。')
events = notices()
if type(events) is not list or len(events) > 4 or any(type(x) is not str or not x.strip() or len(x) > 240 for x in events):
raise ValueError('本轮通知最多4条，每条1到240字符。')
if events:
messages[0]['content'] += '\n\n[宿主通知，仅供参考]\n' + '\n'.join(events)
if _s10._s08.history_bytes(messages) > 8000:
raise ValueError('本次请求消息超过8000字节。')
return {'history': canonical, 'messages': messages}
def dispatch_call(call, snapshot, ask_user, hooks):
"""一个响应只用一份能力快照；所有入口先过Hook，再走各自许可。"""
fixed = deepcopy(call)
def guard(current, tools, confirm, blocked=()):
if current['name'] in snapshot['pool']['routes']:
return _s19.call_external(current, snapshot['pool'], confirm)
if current['name'] == 'load_skill':
return _s07._s02.execute_tool(current, tools)
return _s07._s04.execute_guarded(current, tools, confirm, blocked)
handled = _s07._s04.execute_hooked(
fixed, snapshot['builtins'], ask_user, hooks, guard=guard,
)
return handled
def run_session(question, capabilities, prepare, send, ask_user, hooks, close, max_calls=5):
"""只装配原run_loop；恢复只包模型，finally收尾不再发起新一轮。"""
if type(max_calls) is not int or not 1 <= max_calls <= 8:
raise ValueError('模型逻辑调用上限必须在1到8。')
state = {'prepared': None, 'snapshot': None, 'warnings': []}
recovery = {'remaining': 20, 'waited': 0.0, 'events': []}
outcome = None
def before(history):
prepared = prepare(history)
state['prepared'] = prepared
return prepared['history']
def ask(history):
snapshot = freeze_capabilities(capabilities())
state['snapshot'] = snapshot
request = {'model': 'teaching-model', 'system': snapshot['system'],
'tools': deepcopy(snapshot['pool']['definitions']),
'messages': deepcopy(state['prepared']['messages']), 'max_tokens': 1024}
return _s11.request_with_recovery(send, request, recovery)
def execute(call):
handled = dispatch_call(call, state['snapshot'], ask_user, hooks)
state['warnings'].extend(handled['warnings'])
return handled['result']
try:
outcome = run_loop(ask, execute, max_calls=max_calls, emit=lambda _: None,
question=question, prepare=before)
except Exception as error:
outcome = {'status': 'host_error', 'text': type(error).__name__, 'calls': None, 'messages': None}
finally:
try:
close()
except Exception as error:
state['warnings'].append('收尾失败：' + type(error).__name__)
return {'outcome': outcome, 'warnings': state['warnings'],
'recovery': recovery}
_s10=SimpleNamespace(_s08=SimpleNamespace(history_bytes=history_bytes))
_s07=SimpleNamespace(_s02=SimpleNamespace(execute_tool=execute_tool),_s04=SimpleNamespace(execute_guarded=execute_guarded,execute_hooked=execute_hooked))
_s19=SimpleNamespace(call_external=call_external)
_s11=SimpleNamespace(request_with_recovery=request_with_recovery)
scenario='builtin'
history=[{'role':'user','content':'检查main.py，核对42。'}];original=deepcopy(history)
notices_log=[];invoked=[];closed=[];sent=[];approvals=[]
def compact(history):
if scenario in ('compact_error','context_error'):raise ValueError('归档失败')
return deepcopy(history)
def attach(messages):
if scenario!='empty':messages[0]['content']+='\n[项目记忆] 按本次输出核对。'
if scenario=='oversize':messages[0]['content']+='x'*8001
return messages
def notices():return ['x'*241] if scenario=='bad_notices' else ([] if scenario=='empty' else ['收到检查任务。'])
def program():invoked.append('run_main');return 'exit_code=0\n42'
def remote(name,args):invoked.append(name);return {'content':[{'type':'text','text':'需要本次许可。'}],'isError':False}
def confirm(*_):approvals.append('asked');return scenario!='denied'
def before(_):return '前置阻止' if scenario=='blocked' else None
def after(_):
if scenario=='warning':raise ValueError('观察故障')
tools={'run_main':{'description':'固定运行main.py','parameters':{},'run':program}}
pool={'definitions':[{'name':'run_main','description':'固定运行main.py','input_schema':{'type':'object','properties':{},'additionalProperties':False}}],
'routes':{'mcp__docs__search_docs':{'server':'docs','remote':'search.docs','schema':{'properties':{'query':{'type':'string'}}},'invoke':remote}}}
snapshot={'builtins':tools,'pool':pool,'system':'按本次证据核对。'}
call={'type':'tool_use','id':'step_1','name':'mcp__docs__search_docs' if scenario=='external' else 'missing' if scenario=='unknown' else 'run_main','input':{'query':'运行许可'} if scenario=='external' else {}}
hooks={'before':[before],'after':[after]}
def capabilities():return snapshot
def prepare(history):return prepare_turn(history,compact,attach,notices)
def send(request):
sent.append(deepcopy(request))
if scenario=='model_error':raise ValueError('模型接口故障')
if scenario=='retry' and len(sent)==2:return {'failure':{'status':529,'retry_after':'0'}}
if len(sent)==1:return {'stop_reason':'tool_use','content':[call]}
return {'stop_reason':'end_turn','content':[{'type':'text','text':'收到实际工具反馈。'}]}
def close():
closed.append('closed')
if scenario=='close_error':raise OSError('关闭故障')
# 调用入口
try:
observed=dispatch_call(call,snapshot,confirm,hooks)
except Exception as error:
observed={"error":str(error)}
print(observed)
- 固定调用与能力快照。
- 共同Hook包住各自路由和许可。
- 关联原调用ID交回结果，观察警告另存。
“本地运行”会经过s03：run_main每次确认，未知工具默认拒绝。“外部查询”走s19.call_external，验证参数、询问宿主，再以服务原名调用。load_skill沿用s07已注册名称的只读加载，不把规范文字当成程序运行许可。
在handled赋值前暂停，工具尚未执行；进入guard之后才选中具体入口。选“前置阻止”，调用不会到达guard；选“宿主拒绝”，Hook通过也仍不能运行。选“观察故障”，工具已经完成，after异常只出现在warnings，不会改变结果或重做工具。
返回值仍保留handled.result与handled.warnings两部分。原循环只把result按模型调用ID保存，宿主另存警告；让两者分别可见，才能区分“工具失败”和“日志观察失败”。
run_session：回调装配，原循环继续独占历史
会话接收当前问题、能力提供者、请求准备器、模型发送器、宿主确认、Hook和收尾回调。函数没有再写一条while循环，而是把before、ask、execute交给第一课的run_loop。
before先保存prepare返回的请求副本，随后交出history，原循环收到它才提交。ask从同一次能力快照生成tools，把带附件的messages放进request；execute使用state中该响应的快照。freeze_capabilities复制定义、路由和本地元数据，保留宿主拥有的可调用接口，不复制SDK连接。
真实执行轨迹回放；原函数、循环、恢复与Hook。模型、文件进程、压缩附件与连接为公开内存教学绑定。
Pys20/core/runtime.py · run_session
未运行
def run_session(question, capabilities, prepare, send, ask_user, hooks, close, max_calls=5):
"""只装配原run_loop；恢复只包模型，finally收尾不再发起新一轮。"""
if type(max_calls) is not int or not 1 <= max_calls <= 8:
raise ValueError('模型逻辑调用上限必须在1到8。')
state = {'prepared': None, 'snapshot': None, 'warnings': []}
recovery = {'remaining': 20, 'waited': 0.0, 'events': []}
outcome = None
def before(history):
prepared = prepare(history)
state['prepared'] = prepared
return prepared['history']
def ask(history):
snapshot = freeze_capabilities(capabilities())
state['snapshot'] = snapshot
request = {'model': 'teaching-model', 'system': snapshot['system'],
'tools': deepcopy(snapshot['pool']['definitions']),
'messages': deepcopy(state['prepared']['messages']), 'max_tokens': 1024}
return _s11.request_with_recovery(send, request, recovery)
def execute(call):
handled = dispatch_call(call, state['snapshot'], ask_user, hooks)
state['warnings'].extend(handled['warnings'])
return handled['result']
try:
outcome = run_loop(ask, execute, max_calls=max_calls, emit=lambda _: None,
question=question, prepare=before)
except Exception as error:
outcome = {'status': 'host_error', 'text': type(error).__name__, 'calls': None, 'messages': None}
finally:
try:
close()
except Exception as error:
state['warnings'].append('收尾失败：' + type(error).__name__)
return {'outcome': outcome, 'warnings': state['warnings'],
'recovery': recovery}输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
from types import SimpleNamespace
import json,math,time,random
QUESTION="检查程序"
class RecoveryStopped(RuntimeError):pass
def validate_call(call, tools):
"""本课小型参数契约；供分发与后续权限入口共用。"""
name, arguments = call.get("name"), call.get("input")
if not isinstance(name, str) or name not in tools:
return "未知工具。"
parameters = tools[name]["parameters"]
if not isinstance(arguments, dict) or set(arguments) != set(parameters):
return "参数字段不匹配。"
if any(type(arguments[key]) is not kind for key, kind in parameters.items()):
return "参数类型不匹配。"
return None
def execute_tool(call, tools):
"""校验与执行；每个返回值保留原调用 ID。"""
result = {"type": "tool_result", "tool_use_id": call["id"]}
error = validate_call(call, tools)
if error:
return {**result, "content": error, "is_error": True}
try:
output = tools[call["name"]]["run"](**call["input"])
return {**result, "content": output, "is_error": False}
except (ValueError, OSError) as error:
return {**result, "content": str(error), "is_error": True}
def permission_for(call, tools, blocked=()):
"""只判断，不提问、不执行。没有策略的工具默认拒绝。"""
error = validate_call(call, tools)
if error:
return {"decision": "deny", "reason": error}
name = call["name"]
if name in blocked:
return {"decision": "deny", "reason": "宿主策略禁止这个工具。"}
if name == "read_file":
return {"decision": "allow", "reason": "允许读取；目录约束仍由读取函数检查。"}
if name == "run_main":
return {"decision": "ask", "reason": "运行 main.py 前需要用户允许。"}
return {"decision": "deny", "reason": "这个工具尚未配置权限策略。"}
def authorize_call(call, tools, ask_user, blocked=()):
"""只有 ask 才调用宿主；必须收到布尔 True 才允许。"""
verdict = permission_for(call, tools, blocked)
if verdict["decision"] != "ask":
return verdict
# 宿主看到调用副本；不能通过修改显示对象改变后续执行参数。
answer = ask_user(deepcopy(call), verdict["reason"])
if answer is True:
return {"decision": "allow", "reason": "用户允许本次调用。"}
return {"decision": "deny", "reason": "本次调用未获用户允许。"}
def execute_guarded(call, tools, ask_user, blocked=()):
"""验证、显示与执行使用同一份调用快照；拒绝结果不冒充执行结果。"""
snapshot = deepcopy(call)
verdict = authorize_call(snapshot, tools, ask_user, blocked)
if verdict["decision"] != "allow":
return {
"type": "tool_result", "tool_use_id": snapshot["id"],
"content": "未执行：" + verdict["reason"], "is_error": True,
}
return execute_tool(snapshot, tools)
def run_hooks(hooks, event, payload):
"""before 按顺序遇阻即停；after 故障只记警告，并继续其他观察者。"""
if event not in ("before", "after"):
raise ValueError("未知 hook 事件")
warnings = []
for index, callback in enumerate(tuple(hooks.get(event, ()))):
try:
reason = callback(deepcopy(payload))
if reason is not None:
if event != "before" or not isinstance(reason, str) or not reason.strip():
raise ValueError("before 只能返回 None 或非空拒绝理由；after 必须返回 None")
return {"blocked": reason, "warnings": warnings}
except Exception as error:
message = type(error).__name__ + ": " + str(error)
if event == "before":
return {"blocked": "前置检查异常：" + message, "warnings": warnings}
warnings.append(message)
return {"blocked": None, "warnings": warnings}
def execute_hooked(call, tools, ask_user, hooks, blocked=(), guard=None):
"""包住上一课权限入口；后置观察故障不覆盖实际结果，也不重试工具。"""
snapshot = deepcopy(call)
before = run_hooks(hooks, "before", {"call": snapshot})
if before["blocked"] is not None:
result = {"type": "tool_result", "tool_use_id": snapshot["id"],
"content": "未执行：" + before["blocked"], "is_error": True}
else:
result = (guard or execute_guarded)(snapshot, tools, ask_user, blocked)
after = run_hooks(hooks, "after", {"call": snapshot, "result": result})
return {"result": result, "warnings": after["warnings"]}
def result_text(reply):
if not isinstance(reply, dict) or type(reply.get('isError',False)) is not bool:
raise ValueError('MCP结果格式不合法。')
blocks = reply.get('content')
if not isinstance(blocks, list) or not blocks or len(blocks) > 4 or reply.get('structuredContent') is not None:
raise ValueError('本课只接收一到四块纯文本结果。')
if any(not isinstance(b,dict) or b.get('type') != 'text' or not isinstance(b.get('text'),str) for b in blocks):
raise ValueError('结果包含本课不支持的内容类型。')
text = '\n'.join(block['text'] for block in blocks)
if not text.strip() or len(text) > 2000:
raise ValueError('外部结果为空或超过2000字符。')
return text, reply.get('isError', False)
def call_external(call, pool, ask_user):
"""宿主确认独立于server标注；一次调用，无自动工具重试。"""
snapshot = deepcopy(call)
envelope = {'type':'tool_result', 'tool_use_id':snapshot['id']}
try:
route = pool['routes'].get(snapshot.get('name'))
if route is None:
raise ValueError('本次工具快照中没有这个路由。')
arguments = snapshot.get('input')
if (not isinstance(arguments,dict) or set(arguments) != set(route['schema']['properties'])
or any(type(value) is not str or len(value) > 120 for value in arguments.values())):
raise ValueError('外部工具参数字段、类型或长度不符。')
if ask_user(deepcopy(snapshot), route['server']) is not True:
return {**envelope, 'is_error':True, 'content':'未执行：本次外部调用未获批准。'}
reply = route['invoke'](route['remote'], deepcopy(arguments))
text, failed = result_text(reply)
return {**envelope, 'is_error':failed, 'content':text}
except Exception as error:
return {**envelope, 'is_error':True, 'content':'外部调用未能完成确认：' + type(error).__name__ + ': ' + str(error)[:180]}
def history_bytes(messages):
return len(json.dumps(messages, ensure_ascii=False, separators=(',', ':')).encode('utf-8'))
def checked_state(state):
if type(state) is not dict or set(state) != {'remaining', 'waited', 'events'}:
raise ValueError('恢复状态必须由宿主持有。')
if type(state['remaining']) is not int or not 0 <= state['remaining'] <= 20 or type(state['waited']) not in (int, float) or not math.isfinite(state['waited']) or not 0 <= state['waited'] <= 3 or type(state['events']) is not list or len(state['events']) > 20:
raise ValueError('请求或等待额度无效。')
def classify_failure(outcome):
"""接收适配器的基本数据，不匹配任意SDK异常文字。"""
if type(outcome) is not dict:
return {'action': 'stop', 'reason': 'invalid_response'}
if 'failure' in outcome:
failure = outcome['failure']
if type(failure) is not dict:
return {'action': 'stop', 'reason': 'invalid_failure'}
status = failure.get('status')
if type(status) is int and status in (429, 500, 502, 503, 504, 529):
return {'action': 'wait', 'reason': str(status), 'retry_after': failure.get('retry_after')}
if failure.get('transport') in ('connection', 'timeout') and status is None:
return {'action': 'wait', 'reason': failure['transport']}
return {'action': 'stop', 'reason': 'request_error:' + str(status)}
reason = outcome.get('stop_reason')
if reason == 'max_tokens':
return {'action': 'expand', 'reason': 'max_tokens'}
if reason in ('end_turn', 'tool_use') and type(outcome.get('content')) is list and outcome['content']:
return {'action': 'accept', 'reason': reason}
return {'action': 'stop', 'reason': 'unsupported_response:' + str(reason)}
def retry_delay(attempt, retry_after=None, jitter=0.0):
"""优先有效秒数；无效/日期格式回退，过长等待由调用者停下而非缩短。"""
if type(attempt) is not int or not 0 <= attempt < 4:
raise ValueError('尝试序号必须在0到3之间。')
if type(jitter) not in (int, float) or not math.isfinite(jitter) or not 0 <= jitter <= 1:
raise ValueError('抖动比例必须在0到1之间。')
seconds = None
if type(retry_after) in (str, int, float):
try:
value = float(retry_after)
if math.isfinite(value) and value >= 0:
seconds = value
except ValueError:
pass
if seconds is not None:
return seconds
base = min(0.25 * 2 ** attempt, 2.0)
return base + base * 0.25 * jitter
def request_with_recovery(send, request, state, pause=time.sleep, random_unit=random.random):
"""每次逻辑请求最多4次尝试；父会话共享20次请求/3秒计划等待额度。"""
checked_state(state)
pending = deepcopy(request)
if pending.get('max_tokens') != 1024 or type(pending.get('max_tokens')) is not int:
raise ValueError('教学输出预算从1024开始。')
expanded = False
for attempt in range(1, 5):
if state['remaining'] == 0:
raise RecoveryStopped('request_budget')
state['remaining'] -= 1
outcome = send(deepcopy(pending))
decision = classify_failure(outcome)
state['events'].append({'attempt': attempt, 'action': decision['action'], 'reason': decision['reason']})
if decision['action'] == 'accept':
return outcome
if decision['action'] == 'stop':
raise RecoveryStopped(decision['reason'])
if attempt == 4 or state['remaining'] == 0:
raise RecoveryStopped('request_budget')
if decision['action'] == 'expand':
if expanded:
raise RecoveryStopped('output_still_truncated')
pending['max_tokens'] = 2048
expanded = True
continue
delay = retry_delay(attempt - 1, decision.get('retry_after'), random_unit())
if state['waited'] + delay > 3:
raise RecoveryStopped('wait_budget')
state['waited'] += delay
state['events'][-1]['delay'] = delay
pause(delay)
raise RecoveryStopped('request_budget')
def run_loop(ask, execute, max_calls=4, emit=print, question=QUESTION, prepare=None):
"""ask 读完整记录，execute 读工具请求；返回状态和证据，避免把停下当作成功。"""
if max_calls < 1:
raise ValueError("max_calls 必须至少为 1")
messages = [{"role": "user", "content": question}]
def finish(status, text, calls):
return {"status": status, "text": text, "calls": calls, "messages": messages}
for number in range(1, max_calls + 1):
if prepare is not None:
try:
candidate = prepare(deepcopy(messages))
if type(candidate) is not list:
raise ValueError("prepare必须返回候选消息列表")
except Exception as error:
detail = str(error)[:160] if isinstance(error, ValueError) else type(error).__name__
return finish("context_error", f"上下文准备失败：{detail}；原历史保留。", number - 1)
messages[:] = candidate
emit(f"模型调用 {number}")
try:
response = ask(messages)
except Exception as error:
# API 异常时交回控制权；不自动重试，也不伪造工具结果。
return finish("model_error", f"模型请求失败：{type(error).__name__}", number)
blocks = response.get("content")
reason = response.get("stop_reason")
if not isinstance(blocks, list) or any(
not isinstance(block, dict) or block.get("type") not in ("text", "tool_use")
for block in blocks
):
return finish("protocol_error", "响应不符合本课支持的格式。", number)
messages.append({"role": "assistant", "content": blocks})
calls = [block for block in blocks if block["type"] == "tool_use"]
text = "\n".join(block.get("text", "") for block in blocks if block["type"] == "text")
if reason == "end_turn" and not calls and text.strip():
return finish("answered", text, number)
if reason not in ("end_turn", "tool_use"):
return finish("model_stopped", f"模型停止：{reason}；需要检查响应。", number)
ids = [call.get("id") for call in calls]
if reason != "tool_use" or not calls or any(
not isinstance(call_id, str) or not call_id for call_id in ids
) or len(set(ids)) != len(ids):
return finish("protocol_error", "停止原因与工具请求不一致。", number)
# 上面已经保存完整 assistant 响应；每个结果必须关联原调用 ID。
results = []
for call in calls:
emit(f"处理工具请求 {call.get('name')}")
result = execute(call)
results.append(result)
emit("工具结果：" + result["content"].splitlines()[-1])
messages.append({"role": "user", "content": results})
emit("保存工具结果，供下一次调用读取")
return finish("call_limit", f"已达 {max_calls} 次模型调用上限；保留记录，交给用户。", max_calls)
def freeze_capabilities(current):
"""冻结定义/路由/本地元数据，保留宿主拥有的可调用接口，不复制连接。"""
return {'system': current['system'],
'builtins': {name: {**deepcopy({k:v for k,v in spec.items() if k != 'run'}), 'run': spec['run']}
for name,spec in current['builtins'].items()},
'pool': {'definitions': deepcopy(current['pool']['definitions']),
'routes': {name: {**deepcopy({k:v for k,v in route.items() if k != 'invoke'}), 'invoke': route['invoke']}
for name,route in current['pool']['routes'].items()}}}
def prepare_turn(history, compact, attach, notices):
"""压缩候选供原循环提交；记忆和通知只附在本次请求副本上。"""
canonical = compact(deepcopy(history))
if type(canonical) is not list or not canonical:
raise ValueError('压缩必须返回非空历史。')
messages = attach(deepcopy(canonical))
if type(messages) is not list or not messages or messages[0].get('role') != 'user' or type(messages[0].get('content')) is not str:
raise ValueError('请求必须从文本用户问题开始。')
events = notices()
if type(events) is not list or len(events) > 4 or any(type(x) is not str or not x.strip() or len(x) > 240 for x in events):
raise ValueError('本轮通知最多4条，每条1到240字符。')
if events:
messages[0]['content'] += '\n\n[宿主通知，仅供参考]\n' + '\n'.join(events)
if _s10._s08.history_bytes(messages) > 8000:
raise ValueError('本次请求消息超过8000字节。')
return {'history': canonical, 'messages': messages}
def dispatch_call(call, snapshot, ask_user, hooks):
"""一个响应只用一份能力快照；所有入口先过Hook，再走各自许可。"""
fixed = deepcopy(call)
def guard(current, tools, confirm, blocked=()):
if current['name'] in snapshot['pool']['routes']:
return _s19.call_external(current, snapshot['pool'], confirm)
if current['name'] == 'load_skill':
return _s07._s02.execute_tool(current, tools)
return _s07._s04.execute_guarded(current, tools, confirm, blocked)
handled = _s07._s04.execute_hooked(
fixed, snapshot['builtins'], ask_user, hooks, guard=guard,
)
return handled
def run_session(question, capabilities, prepare, send, ask_user, hooks, close, max_calls=5):
"""只装配原run_loop；恢复只包模型，finally收尾不再发起新一轮。"""
if type(max_calls) is not int or not 1 <= max_calls <= 8:
raise ValueError('模型逻辑调用上限必须在1到8。')
state = {'prepared': None, 'snapshot': None, 'warnings': []}
recovery = {'remaining': 20, 'waited': 0.0, 'events': []}
outcome = None
def before(history):
prepared = prepare(history)
state['prepared'] = prepared
return prepared['history']
def ask(history):
snapshot = freeze_capabilities(capabilities())
state['snapshot'] = snapshot
request = {'model': 'teaching-model', 'system': snapshot['system'],
'tools': deepcopy(snapshot['pool']['definitions']),
'messages': deepcopy(state['prepared']['messages']), 'max_tokens': 1024}
return _s11.request_with_recovery(send, request, recovery)
def execute(call):
handled = dispatch_call(call, state['snapshot'], ask_user, hooks)
state['warnings'].extend(handled['warnings'])
return handled['result']
try:
outcome = run_loop(ask, execute, max_calls=max_calls, emit=lambda _: None,
question=question, prepare=before)
except Exception as error:
outcome = {'status': 'host_error', 'text': type(error).__name__, 'calls': None, 'messages': None}
finally:
try:
close()
except Exception as error:
state['warnings'].append('收尾失败：' + type(error).__name__)
return {'outcome': outcome, 'warnings': state['warnings'],
'recovery': recovery}
_s10=SimpleNamespace(_s08=SimpleNamespace(history_bytes=history_bytes))
_s07=SimpleNamespace(_s02=SimpleNamespace(execute_tool=execute_tool),_s04=SimpleNamespace(execute_guarded=execute_guarded,execute_hooked=execute_hooked))
_s19=SimpleNamespace(call_external=call_external)
_s11=SimpleNamespace(request_with_recovery=request_with_recovery)
scenario='answered'
history=[{'role':'user','content':'检查main.py，核对42。'}];original=deepcopy(history)
notices_log=[];invoked=[];closed=[];sent=[];approvals=[]
def compact(history):
if scenario in ('compact_error','context_error'):raise ValueError('归档失败')
return deepcopy(history)
def attach(messages):
if scenario!='empty':messages[0]['content']+='\n[项目记忆] 按本次输出核对。'
if scenario=='oversize':messages[0]['content']+='x'*8001
return messages
def notices():return ['x'*241] if scenario=='bad_notices' else ([] if scenario=='empty' else ['收到检查任务。'])
def program():invoked.append('run_main');return 'exit_code=0\n42'
def remote(name,args):invoked.append(name);return {'content':[{'type':'text','text':'需要本次许可。'}],'isError':False}
def confirm(*_):approvals.append('asked');return scenario!='denied'
def before(_):return '前置阻止' if scenario=='blocked' else None
def after(_):
if scenario=='warning':raise ValueError('观察故障')
tools={'run_main':{'description':'固定运行main.py','parameters':{},'run':program}}
pool={'definitions':[{'name':'run_main','description':'固定运行main.py','input_schema':{'type':'object','properties':{},'additionalProperties':False}}],
'routes':{'mcp__docs__search_docs':{'server':'docs','remote':'search.docs','schema':{'properties':{'query':{'type':'string'}}},'invoke':remote}}}
snapshot={'builtins':tools,'pool':pool,'system':'按本次证据核对。'}
call={'type':'tool_use','id':'step_1','name':'mcp__docs__search_docs' if scenario=='external' else 'missing' if scenario=='unknown' else 'run_main','input':{'query':'运行许可'} if scenario=='external' else {}}
hooks={'before':[before],'after':[after]}
def capabilities():return snapshot
def prepare(history):return prepare_turn(history,compact,attach,notices)
def send(request):
sent.append(deepcopy(request))
if scenario=='model_error':raise ValueError('模型接口故障')
if scenario=='retry' and len(sent)==2:return {'failure':{'status':529,'retry_after':'0'}}
if len(sent)==1:return {'stop_reason':'tool_use','content':[call]}
return {'stop_reason':'end_turn','content':[{'type':'text','text':'收到实际工具反馈。'}]}
def close():
closed.append('closed')
if scenario=='close_error':raise OSError('关闭故障')
# 调用入口
try:
observed=run_session('检查main.py',capabilities,prepare,send,confirm,hooks,close,max_calls=2)
except Exception as error:
observed={"error":str(error)}
print(observed)
- prepare/ask/execute装配原唯一循环。
- 恢复只包模型请求，工具反馈由原循环保存。
- 有限停止后finally收尾，不再发起任务。
恢复只围住request_with_recovery(send, request, recovery)。选择“只重试模型”，第一次工具反馈已保存，随后一次529使同一request重发；执行轨迹不会再进入工具回调。恢复预算沿用s11：每逻辑请求最多四尝试，会话共享20次实际请求、三秒计划等待，SDK内置重试关闭。
run_loop仍检查响应、保存完整assistant、按ID保存工具结果，并决定answered、call_limit、context_error、model_error或协议停止。本课允许一到八次逻辑调用，默认五次；不会因为工具名称多就取消上限。
finally在循环退出后执行一次close。选“收尾故障”，原answered仍保留，警告另存；选“准备失败”，模型一次也没调用，仍然收尾。连接的实际关闭由外层async上下文管理，close是宿主结束回调，不假称它负责所有资源。
如果分发边界发生未预料的普通异常，run_session返回host_error。原run_loop没有交出结果，此时messages和calls设为None，不能宣称拿到了完整内部历史；可用的审计记录另由宿主持有。非法调用额度在资源接管前校验，外层创建连接的上下文仍应负责关闭。
真正接到文件与MCP的地方
run_example把之前的模块原样接进来：s07扫描并加载SKILL.md，s09读取宿主确认过的项目记忆，s08把超预算历史写进真实归档，s10根据实际工具组装system，s11恢复模型请求。准备与执行的实现都能在公开入口中逐项找到。
默认LocalDocs只是查询替身。要验证真实协议通信，安装固定依赖，使用--mcp：
bash
python3 -m pip install -r requirements.txt -r examples/s19-mcp-tools/requirements.txt
python3 examples/s20-comprehensive-agent/user_code/practice.py --mcp --decision allow
python3 examples/s20-comprehensive-agent/user_code/verify_sdk.py这里用官方mcp==1.26.0启动仓库拥有的s19服务，实际初始化、发现目录、查询与关闭，协商2025-11-25。旧同步会话由asyncio.to_thread运行，沿用s19.SessionBridge把SDK协程投回原事件循环；没有在事件循环中同步等待自己。
Anthropic==1.13.0使用MockTransport模拟HTTP，没有在线模型调用。允许路径六次HTTP，含工具结果后的529；MCP与main.py实际各执行一次。拒绝路径五次HTTP，同样验证恢复，但MCP与程序都是零次。重试前后的请求相同，工具入口没有重入。
不要把任务调度写进ask
任务图、Cron和自主认领决定“接下来交付哪项工作”；团队协议、邮箱和Worktree决定“哪个成员在哪个目录工作”。它们可以调用同一个run_session，却不该在send失败时重新认领任务或创建目录。
本课的贯穿问题没有并发编辑、定时触发或慢程序需求，因此没有启动这些服务，也没有通过空占位函数宣称所有机制都已运行。要加一个工作来源，先在会话入口前接好负责人和目录；要加一个工具，在能力池和许可入口明确注册；要加通知，先确定事件保存和确认语义。
这种分层还保留了一个容易观察的边界：模型停下时，后台工作可以仍在运行，成员也可能仍是idle。任务完成、会话停止和资源关闭分别用自己的事实判断，不能统一改成一个done。
bash
python3 examples/s20-comprehensive-agent/user_code/verify_practice.py
python3 examples/s20-comprehensive-agent/user_code/verify_fixtures.py十四组标准库回归覆盖真实归档与获准进程、拒绝、输出不符、Hook观察故障、请求失败、额度用完、收尾故障，以及附件不累加、能力快照固定、模型重试不重做工具。十八种公开教学输入原样执行三个函数及原循环、恢复、Hook和权限；浏览器里的文件进程、模型、压缩附件与连接接口明确为内存绑定。
最后重跑最初那条任务：用户想知道本次程序是否输出42。应当交回的是本次运行证据、会话怎样停止、还有哪些警告。把这些事实保持分开，新增一个机制时，就能清楚决定它该进入哪一个函数。
学习方向参考社区s20；MCP生命周期沿用官方2025-11-25规范。2026-10-20核对，以上边界与额度均为本册实现选择。