文档目录
简单记忆系统
让 Jev 决定哪些请求和答案值得被记住，然后把 Jev 的调用包进一个小函数，在后续调用里加上这些记忆。
非官方中文翻译 · 内容以 英文原文 ↗ 为准 · 同步于 2026-10-15
每次 Jev 请求只看得到你发给它的 state（上下文状态）。客户上个月告诉过你应用的事——比如他们的 edition 或 plan——不会出现在他们的下一次请求里，除非你的代码把它放进去。这里的记忆系统就是这么做的：每次调用多发一条 Jev 请求，外加一个 Python dict。
我们的示例是把 support 工单路由到相应团队。8 月，一位客户说他们在 Windows Server 上自托管。9 月，他们发来消息「Dashboards have been really slow since this morning」。没有记忆时，Jev 会把这张工单派给 cloud 团队，而不是 self-hosted 团队。在下面的例子里，加上记忆后，6 张后续工单全部派到了预期的团队；而单独路由每张工单，6 张里只能对 2 张。
这个简单记忆系统由两部分组成：
- 每次调用结束后，Jev 会同时看到原始请求和它的答案。一个
Noul问题（我们称之为 gate）会问：请求里是否陈述了一个关于客户的重要事实，值得被记住。如果 P(yes) 至少达到 0.5，这张工单和 Jev 的答案就会被保存为该客户的一条记忆。 - 一个包装函数
ask_with_memory()接收与ask()相同的state和questions参数（在 TypeSafe Python SDK 中，ask()是client.system_one方法，已固定模型并启用调用缓存），并返回 Jev 的响应。每次调用之前，它会把该客户已保存的工单以及 Jev 对这些工单的答案一并加到 state 里。
这个包装就是一个基础的 agent harness。harness 指模型外面那一层代码，给模型带来诸如工具调用和循环运行之类的额外能力。这个简易版 harness 只给它一份记忆，存放于一个普通的 Python dict 中，每个客户 ID 对应一个列表。
%%{init: {"fontFamily": "Inter, sans-serif", "flowchart": {"rankSpacing": 25, "wrappingWidth": 300, "subGraphTitleMargin": {"top": 8, "bottom": 20}}}}%%
flowchart TB
IN["ask_with_memory(state, questions)"]
subgraph R["Jev: answer the question"]
Q1["<b>Choice:</b> which support team?"]
end
subgraph G["Jev: judge the routed ticket"]
Q2["<b>Noul:</b> does the request state<br/>an important fact about the customer?"]
end
DEC{"P(yes) ≥ 0.5?<br/>your code"}
SAVE["save the memory<br/>under this customer's ID"]
OUT["handle the support ticket response in your code"]
MEM[("MEMORIES<br/>saved memories per customer")]
IN -- "state with memories<br/>+ routing question" --> R
R -- "original request + answer<br/>+ memory question" --> G
G -- "P(yes)" --> DEC
DEC -- "yes: save" --> SAVE
DEC -- "no: don't save" --> OUT
SAVE --> OUT
SAVE --> MEM
IN -. "added to the state<br/>on this customer's<br/>next support tickets" .- MEM包装会发出两条 Jev 请求，一前一后：先是路由请求，然后是 gate，gate 会结合 Jev 给出的答案一起评判这张工单。无论是否保存了任何东西，包装都返回路由的响应。
由于 ask_with_memory() 接收与 ask() 相同的参数并返回相同的响应，你可以把它们互换，而不用改动任何其他代码。包装会自己找出该客户的记忆，加到 state 里，并保存新的记忆。
这个例子先让两位客户的 5 张早期工单经过包装，然后对每位客户的 3 条后续消息分别路由两次：一次带上包装保存的记忆，一次不带。一张表格把这两种答法并排列出，共 6 条。
设置
pip install ipython 'cooksafe>=0.2.0,<0.3.0'设置 TYPESAFE_API_KEY。每次 API 调用都会缓存到 json_cache.json。把它下载到你运行这段代码的目录里，就能复现文档里给出的数字，而无需调用 API；也可以不放它，让所有调用实时运行。
下面的数字来自 jev-1.13.0，时间为 2026-09-28。0.5 的记忆阈值只是一个起点，你应该结合自己的样例来评估它。
import json
import os
from pathlib import Path
from cooksafe import JsonCache, make_playground_link
from IPython.display import Markdown, display
from typesafe_sdk import Choice, Noul, NoulCriteria, SystemOneResponse, TypeSafeClient
TYPESAFE_MODEL = "jev-1.13.0"
REMEMBER_AT = 0.5 # save when P(lasting customer fact) reaches this starting threshold
MEMORY_KEY = "earlier_tickets_from_this_customer" # where saved tickets go in the state
client = TypeSafeClient(
api_key=os.environ.get("TYPESAFE_API_KEY", "cache-only"), # keyless kernels replay the cache
base_url=os.environ.get("TYPESAFE_ENDPOINT"),
timeout=120.0,
)
json_cache = JsonCache(Path("json_cache.json"))定义工单
两位客户、5 张早期工单和 3 条后续消息，全部是为这份 cookbook 编写的。C-1042（Harbor Logistics）在 Windows Server 上运行 self-hosted edition，签的是 Enterprise 合同。C-2210（Fernwood Studio）使用 cloud edition 的自助 Team plan。5 张早期工单中有 3 张陈述了一个长期事实：一项 edition、一份 plan、一份合同。另外 2 张则没有：在机场用手机、公共假期。没有后续消息提及客户的部署情况，所以没有记忆时，Jev 只能靠猜。
每张工单变成一个由两部分组成的 state：customer（客户是谁）和 ticket（工单会话）。ticket_state() 函数构造一张简化版的工单，就像 helpdesk 里的工单。在你的应用中，state 就是你的 helpdesk 发来的每张工单，ask_with_memory() 会把该客户已保存的工单加到它里面。早期工单与后续消息的拆分只是为了本例的演示。
DATA = {
"customers": {
"C-1042": {"id": "C-1042", "company": "Harbor Logistics"},
"C-2210": {"id": "C-2210", "company": "Fernwood Studio"},
},
"earlier_tickets": [
{"id": "T-101", "customer_id": "C-1042", "received_on": "2026-08-04",
"subject": "Adding an admin",
"text": "Hi, we run your self-hosted edition on Windows Server 2022 in our own data "
"centre. How do I add a new admin user?"},
{"id": "T-102", "customer_id": "C-2210", "received_on": "2026-08-11",
"subject": "Exporting to PDF",
"text": "We're a small design studio on your cloud edition, paying monthly on the "
"self-serve Team plan. How do I export a dashboard as a PDF?"},
{"id": "T-103", "customer_id": "C-1042", "received_on": "2026-08-19",
"subject": "Invoice download",
"text": "Sorry for the typos, typing this on my phone at the airport. Where can I "
"download last month's invoice?"},
{"id": "T-104", "customer_id": "C-2210", "received_on": "2026-09-02",
"subject": "Dark mode?",
"text": "Our whole team is out on Friday for a public holiday, so no rush on this "
"one. Is there a dark mode?"},
{"id": "T-105", "customer_id": "C-1042", "received_on": "2026-09-10",
"subject": "Sandbox environment",
"text": "For context, we signed an Enterprise contract with you in March. Can we get "
"a sandbox environment for testing?"},
],
"later_messages": [
{"subject": "Slow dashboards",
"text": "Dashboards have been really slow since this morning."},
{"subject": "Upgrade failing",
"text": "The upgrade to version 5.2 keeps failing halfway through."},
{"subject": "More seats",
"text": "We'd like to add 20 more seats before next quarter."},
],
"later_received_on": "2026-09-28",
# Labelled by hand from what each customer said in the earlier tickets.
"expected_team": {
"C-1042": {"Slow dashboards": "selfhosted_windows",
"Upgrade failing": "selfhosted_windows", "More seats": "enterprise_desk"},
"C-2210": {"Slow dashboards": "cloud_ops", "Upgrade failing": "cloud_ops",
"More seats": "billing"},
},
}
CUSTOMERS = DATA["customers"]
def ticket_state(customer_id: str, ticket_id: str, received_on: str, subject: str,
text: str) -> dict:
"""The state Jev sees for one ticket: a simplified ticket, like one from a helpdesk."""
return {
"customer": CUSTOMERS[customer_id],
"ticket": {
"id": ticket_id,
"received_on": received_on,
"subject": subject,
"messages": [{"from": "customer", "text": text}],
},
}
EARLIER = [
ticket_state(t["customer_id"], t["id"], t["received_on"], t["subject"], t["text"])
for t in DATA["earlier_tickets"]
]
print(f"{len(CUSTOMERS)} customers, {len(EARLIER)} earlier tickets, "
f"{len(DATA['later_messages'])} later messages. The first ticket as Jev sees it:\n")
print(json.dumps(EARLIER[0], indent=2))2 customers, 5 earlier tickets, 3 later messages. The first ticket as Jev sees it:
{
"customer": {
"id": "C-1042",
"company": "Harbor Logistics"
},
"ticket": {
"id": "T-101",
"received_on": "2026-08-04",
"subject": "Adding an admin",
"messages": [
{
"from": "customer",
"text": "Hi, we run your self-hosted edition on Windows Server 2022 in our own data centre. How do I add a new admin user?"
}
]
}
}为工单选路
一道 Choice（选择题）题型用来决定工单交给哪个支持团队。none_suitable 选项给 Jev 留了一条退路：没有任何团队合适时，可以选它；一张上下文太少的工单也可能落到这一项。
ask() 这个辅助函数相当于 client.system_one，只是把模型版本固定下来，并缓存每一次调用结果。它接受一个 state（上下文状态）和一个题型字典，返回 SDK 自带的响应对象；所以那些已经习惯于读 client.system_one 响应的代码，可以原样使用它。
ROUTE = {
"team": Choice(
instructions="Which support team should handle `ticket`?",
criteria={
"cloud_ops": "Cloud team: problems with the hosted cloud edition, run by us",
"selfhosted_linux": "Self-hosted team, Linux: installs, upgrades and "
"performance of the self-hosted edition on Linux servers",
"selfhosted_windows": "Self-hosted team, Windows: installs, upgrades and "
"performance of the self-hosted edition on Windows Server",
"billing": "Billing: invoices, payments, seats and plan changes for "
"self-serve plans",
"enterprise_desk": "Enterprise desk: seats, contracts and account changes for "
"customers on an Enterprise contract (their technical problems go to the "
"technical teams)",
"general": "General product questions: features, how-to, settings",
"none_suitable": "None of these teams fits the ticket",
},
),
}
@json_cache
def _system_one(model: str, state: str, questions: str) -> dict:
response = client.system_one(
state=json.loads(state), questions=json.loads(questions), model=model
)
return response.model_dump(mode="json")
def ask(state: dict, questions: dict) -> SystemOneResponse:
"""client.system_one, with the model pinned and the call cached."""
as_json = {qid: q.model_dump(mode="json", exclude_none=True) for qid, q in questions.items()}
raw = _system_one(TYPESAFE_MODEL, json.dumps(state), json.dumps(as_json))
return SystemOneResponse.model_validate(raw)
def later_state(customer_id: str, message_index: int) -> dict:
"""One of the later messages, sent by this customer, with a fixed ticket ID."""
n = 201 + 3 * list(CUSTOMERS).index(customer_id) + message_index
message = DATA["later_messages"][message_index]
return ticket_state(customer_id, f"T-{n}", DATA["later_received_on"], message["subject"],
message["text"])
slow = later_state("C-1042", 0)
team = ask(slow, ROUTE).answers["team"]
print(f"C-1042: {slow['ticket']['messages'][0]['text']!r}")
print(f" plain Jev -> {team.choice} {team.probabilities[team.choice]:.2f}")C-1042: 'Dashboards have been really slow since this morning.'
plain Jev -> cloud_ops 0.80直接用 Jev 为 C-1042 选路时，slow dashboards 这条工单会被派给 cloud 团队。仅凭这一张工单，它没有理由把它派到别处。
询问 Jev 这张工单是否值得记住
首先，as_memory() 把原始 state 与 Jev 的答案配成一对——也就是封装层可能会保存的那条 memory。其中 request 键保存的是注入 memory 之前的 state，jev_answered 把每道题映射到它的答案。这样的结构是为了避免把旧的 memory 嵌套进每一条新 memory 里面。
第二次 Jev 调用读取这个配对。闸门要问的是：request 中是否陈述了一个关于该客户的重要事实，例如他们的版本、操作系统、套餐或合同。Jev 的答案也作为上下文提供，但只出现在答案里、而没有出现在 request 中的事实，不算数。这条规则是必需的：否则像 "General product questions" 这种泛泛的答案，会盖过 request 里真实陈述的事实；而那些依赖早期 memory 的答案，又会被当作新信息再存一遍。
GATE = {
"remember": Noul(
instructions=(
"In `request.ticket`, does the thread state a lasting fact about the customer that "
"should change how their future, unrelated tickets are handled? Count what the "
"customer or our support staff wrote in `request.ticket`. `jev_answered` shows how "
"the ticket was routed, which may rely on things learned earlier: do not count "
"anything it says or implies."
),
criteria=NoulCriteria(
true=(
"Yes: in `request.ticket` the customer or our support staff state something "
"about the customer that stays true after this ticket, such as the product "
"edition or operating system they run, their plan or contract, or a standing "
"preference."
),
false=(
"No: `request.ticket` only contains things specific to this one ticket or "
"moment, such as a one-off request, today's schedule, or the device they "
"happen to be typing on; or the lasting fact appears only in `jev_answered`."
),
),
),
}
def answer_text(question, answer):
"""The readable part of an answer: the chosen option's description, or the number."""
if answer.type == "choice":
return question.criteria[answer.choice]
return answer.noul if answer.type == "noul" else answer.score
def as_memory(state: dict, questions: dict, response: SystemOneResponse) -> dict:
return {
"request": state,
"jev_answered": {
q.instructions: answer_text(q, response.answers[qid])
for qid, q in questions.items()
},
}
def remember_probability(memory: dict) -> float:
"""P(the request states an important fact about the customer), with the answer as context."""
return ask(memory, GATE).answers["remember"].noul将调用包成会记住上下文的版本
封装函数 ask_with_memory() 与 ask() 的参数和返回值完全相同。memory 存储就是一个普通字典，每个客户 ID 对应一个列表。封装函数用 state 里的 ID 去挑出该客户的 memory。
MEMORIES: dict[str, list[dict]] = {} # customer ID -> that customer's saved memories
def customer_id(state: dict) -> str:
return state["customer"]["id"]
def ask_with_memory(state: dict, questions: dict) -> SystemOneResponse:
"""ask(), plus memory: same arguments, same return value."""
# 1. Answer the call with this customer's existing memories.
saved = MEMORIES.get(customer_id(state), [])
response = ask({**state, MEMORY_KEY: saved} if saved else state, questions)
# 2. Ask Jev whether the original request states an important fact worth keeping.
memory = as_memory(state, questions, response)
if remember_probability(memory) >= REMEMBER_AT:
MEMORIES.setdefault(customer_id(state), []).append(memory)
return response按这种方式加上 memory，会给你的应用带来四点影响：
- 每次调用包含两次顺序请求：你的调用在前，记忆闸门在后。闸门是把工单连同 Jev 的答案一起来判定的，所以它必须排在选路请求之后。如果你的闸门只看工单本身、不关心答案，它完全可以并入选路请求的另一个问题里（参见 fan-out）。
- memory 只会越积越多。 每一条被保存的 memory 都会追加到该客户之后每一次调用中。这里既没有容量上限，也没有相关性检查，更没有过期机制。下一节会针对每一项给出优化思路。
- 你的问题里完全不需要提到 memory。 memory 被放进 state 下的
earlier_tickets_from_this_customer字段，Jev 会读整个 state。ROUTE完全不用改。 - state 必须有
customer.id。 直接调用 Jev 时，state 也可以是一个字符串，但这个封装函数需要靠 ID 来找到该客户的列表。
依次发送早些时候的工单，观察答案如何变化
按到达顺序，把五张较早的工单通过 ask_with_memory() 送出去。同一位客户还会再发两次后到的消息：slow dashboards，以及追加 20 个席位。这样能看出每一条保存下来的 memory 如何改变 Jev 的答案。重复出现的那两条消息也走封装函数，所以闸门同样会判它们。
PROBES = {"Slow dashboards": 0, "More seats": 2} # later messages to send after each ticket
def memory_ids(cid: str) -> str:
return ", ".join(m["request"]["ticket"]["id"] for m in MEMORIES.get(cid, [])) or "none"
def send_probes(cid: str) -> None:
for subject, i in PROBES.items():
team = ask_with_memory(later_state(cid, i), ROUTE).answers["team"]
print(f" {subject:<16} -> {team.choice:<19} {team.probabilities[team.choice]:.2f}")
for cid in CUSTOMERS:
print(f"{cid}, memories: {memory_ids(cid)}")
send_probes(cid)
for state in EARLIER:
cid = customer_id(state)
response = ask_with_memory(state, ROUTE)
p = remember_probability(as_memory(state, ROUTE, response)) # cached: the wrapper's call
verdict = "saved" if p >= REMEMBER_AT else "not saved"
text = state["ticket"]["messages"][0]["text"]
print(f"\n{cid} {state['ticket']['id']} {response.answers['team'].choice:<19} "
f"P(remember) {p:.2f} {verdict}")
print(f" {text[:84]}")
print(f"{cid}, memories: {memory_ids(cid)}")
send_probes(cid)C-1042, memories: none
Slow dashboards -> cloud_ops 0.80
More seats -> enterprise_desk 0.62
C-2210, memories: none
Slow dashboards -> cloud_ops 0.75
More seats -> enterprise_desk 0.52
C-1042 T-101 selfhosted_windows P(remember) 0.91 saved
Hi, we run your self-hosted edition on Windows Server 2022 in our own data centre. H
C-1042, memories: T-101
Slow dashboards -> selfhosted_windows 0.99
More seats -> enterprise_desk 0.44
C-2210 T-102 general P(remember) 0.91 saved
We're a small design studio on your cloud edition, paying monthly on the self-serve
C-2210, memories: T-102
Slow dashboards -> cloud_ops 0.91
More seats -> billing 0.98
C-1042 T-103 billing P(remember) 0.07 not saved
Sorry for the typos, typing this on my phone at the airport. Where can I download la
C-1042, memories: T-101
Slow dashboards -> selfhosted_windows 0.99
More seats -> enterprise_desk 0.44
C-2210 T-104 general P(remember) 0.10 not saved
Our whole team is out on Friday for a public holiday, so no rush on this one. Is the
C-2210, memories: T-102
Slow dashboards -> cloud_ops 0.91
More seats -> billing 0.98
C-1042 T-105 enterprise_desk P(remember) 0.92 saved
For context, we signed an Enterprise contract with you in March. Can we get a sandbo
C-1042, memories: T-101, T-105
Slow dashboards -> selfhosted_windows 0.94
More seats -> enterprise_desk 0.99陈述了版本、套餐或合同的那三张工单，分别以 0.91 至 0.92 的概率被保存下来。T-102 虽然被 Jev 派给了 general 团队，仍然被保存——因为闸门数的是客户自己说的话，而不是那条答案。机场、公共假期那两张工单分别得到 0.07 和 0.10，不会被保存，所以它们之后答案也不会变。
每一条保存下来的 memory 都会改变它相关的那些答案。C-1042 这边，Windows 那张工单把 slow dashboards 从 cloud 团队（0.80）挪到了 Windows 团队（0.99）。Enterprise 合同那条 memory 落地之后，席位问题稳定在 0.99。C-2210 这边，Team-plan 那条工单把 cloud 团队的概率从 0.75 抬到 0.91，并把席位问题从 Enterprise 团队（0.52）挪到 billing（0.98）。加上 Enterprise 那条 memory 之后，C-1042 的 slow dashboards 概率会从 0.99 略微降到 0.94——这正是为什么要挑选该存哪些 memory（见下一节）。另外，闸门没有保存那两条重复出现的消息。
为两个客户路由相同消息
现在每个客户都会发送后面的三条消息。每条消息既会走普通的 ask()，也会走 ask_with_memory()，使用相同的 state 和同一个问题。
EXPECTED = DATA["expected_team"]
plain_correct = memory_correct = total = 0
example_state = None
def cell(answer, expected: str) -> str:
mark = "✓" if answer.choice == expected else "✗"
return f"{mark} {answer.choice} {answer.probabilities[answer.choice]:.2f}"
print(f"{'customer':<9}{'message':<17}{'expected':<20}{'plain Jev':<27}{'with memory':<27}")
for cid in CUSTOMERS:
for i, message in enumerate(DATA["later_messages"]):
state = later_state(cid, i)
expected = EXPECTED[cid][message["subject"]]
plain = ask(state, ROUTE).answers["team"]
if example_state is None:
# Keep the exact state used for this call for the playground link.
saved = MEMORIES.get(cid, [])
example_state = json.loads(json.dumps({**state, MEMORY_KEY: saved} if saved else state))
response = ask_with_memory(state, ROUTE)
remembered = response.answers["team"]
plain_correct += plain.choice == expected
memory_correct += remembered.choice == expected
total += 1
print(f"{cid:<9}{message['subject']:<17}{expected:<20}{cell(plain, expected):<27}"
f"{cell(remembered, expected):<27}")
print(f"\nExpected routes: {plain_correct}/{total} without memory, "
f"{memory_correct}/{total} with memory.")customer message expected plain Jev with memory
C-1042 Slow dashboards selfhosted_windows ✗ cloud_ops 0.80 ✓ selfhosted_windows 0.94
C-1042 Upgrade failing selfhosted_windows ✗ none_suitable 0.60 ✓ selfhosted_windows 1.00
C-1042 More seats enterprise_desk ✓ enterprise_desk 0.62 ✓ enterprise_desk 0.99
C-2210 Slow dashboards cloud_ops ✓ cloud_ops 0.75 ✓ cloud_ops 0.91
C-2210 Upgrade failing cloud_ops ✗ none_suitable 0.63 ✓ cloud_ops 0.86
C-2210 More seats billing ✗ enterprise_desk 0.52 ✓ billing 0.98
Expected routes: 2/6 without memory, 6/6 with memory.加上记忆后，六条消息全部命中预期团队，而不加记忆只有两条命中。同一条消息现在被路由到不同团队：slow dashboards 对 C-1042 走 Windows 团队，对 C-2210 仍走云团队；多要 20 个席位对 C-1042 走 Enterprise 坐席，对 C-2210 走 billing。原本普通 Jev 已经答对的情况，记忆把概率抬高了，比如 C-1042 的席位从 0.62 升到 0.99。包装函数也评估了后面这六条消息，但一条都没保存：这六条都没有陈述重要事实。
每一行预期的团队都来自这些客户在更早工单中确立的设置：C-1042 的自托管版本与企业版合同，C-2210 的云端 Team 套餐。
在 playground 中打开它
该链接保存的是 C-1042 第一次 slow-dashboards 调用所用的精确路由 state，包括当时的记忆，以及路由问题本身。
playground_link = make_playground_link(example_state, ROUTE, models=[TYPESAFE_MODEL])
display(Markdown(
f"🔗 [Open C-1042's ticket with its memories in the TypeSafe playground]({playground_link})"
))在 TypeSafe playground 中打开 C-1042 的工单及其记忆 →
后续步骤
- 选择你的应用应当记住的内容。 编辑
GATE，描述对你自己的任务而言哪些信息值得保存，并用你标注好的样本来测试REMEMBER_AT。 - 选择要加入哪些记忆。 包装函数会添加所有已保存的记忆。当列表变长时，可以加一道相关性问题（relevance question）来筛出更小的子集。
- 替换过时的记忆。 让 Jev 判断新记忆是否替换了较旧的记忆，再在代码里适时移除旧条目。
- 跨运行保留记忆。 用按客户 ID 索引的数据库替代
MEMORIES。进程退出后，内存里的字典就消失了。 - 按路由采取动作。 在代码中把每个团队映射到一个处理器，例如为 Enterprise 客户预约沟通，或发送对应版本的安装说明文章。参见 意图路由 与 置信度分流。把你的回复以客服消息的形式加到工单线程里，门控就能记住你做了什么。
附录
保存的答案会强化错误吗
会的。门控会忽略只出现在 Jev 答案里的内容，所以单凭答案本身不会把工单保存下来。但工单一经保存，它的答案也会一同被保存，包括错误的推断，之后该客户的每次调用都会读到它。后面的答案于是可能重复这个错误。
阈值该怎么选
这个包装函数里的 0.5 只是个起点。要选你自己的值，先从你的流量里把「请求-答案」对标注成值得记或不值得记，再找出能把两组最好分开的概率。在挑阈值时没有用到的另一批对上做校验。如果一条误导性记忆的代价比漏掉一条记忆更大，就调高阈值；反之就调低。关于阈值选取以及与各动作风险等级的匹配，见 Confidence 和 置信度分流。
本文档由 Mintlify 构建和托管。