AI Agent 聊天安全边界判断实现详解
在 AI 电子伴侣、AI 交友聊天这类产品里,Agent 不能只追求会聊天和有陪伴感。越是拟人、亲密、连续对话的场景,越需要在回复前多做一步判断:用户这一次输入,有没有碰到安全边界。
这篇文章我们来看一个已经落地到项目里的安全边界判断方案。它不是简单在 prompt 里补一句安全提醒,而是在 Agent 聊天链路里增加一个前置判断步骤:先用 LangChain 做结构化安全分类,再根据判断结果决定后面是正常回复、温和引导、拒绝请求,还是进入危机支持。

这套实现牵涉到几个关键问题。安全判断为什么要放在聊天回复之前,LangChain structured output 如何稳定返回分类结果,用户自己配置的 OpenAI-compatible LLM 怎么复用,安全结果如何写入消息 metadata_json,以及它怎么继续影响回复策略和长期记忆抽取。我们把这些点放到一条真实请求链路里看,会比单独讲一个安全 prompt 清楚很多。
01flowchart TD02A["用户发送消息"] --> B["解析当前 LLM 配置"]03B --> C["加载 Agent Prompt / Guardrails"]04C --> D["加载最近消息和长期记忆"]05D --> E["LangChain 安全边界判断"]06E --> F["写入用户消息 metadata_json"]07F --> G{"boundaryAction"}08G -->|"continue / soft_boundary / redirect"| H["注入安全策略到 system prompt"]09H --> I["调用普通聊天模型"]10I --> J["流式返回回复"]11J --> K["保存 assistant 消息"]12K --> L{"allowMemoryExtraction"}13L -->|"true"| M["LangChain 长期记忆抽取"]14L -->|"false"| N["跳过记忆抽取"]15G -->|"refuse"| O["直接返回安全拒绝"]16G -->|"crisis_support"| P["直接返回危机支持"]17O --> K18P --> K
当前 Agent 聊天系统已经具备了一些基础能力。用户可以创建自己的 Agent 伴侣,Agent 有人设、故事背景、语气风格和边界规则;聊天消息会持久化,系统也会从聊天里抽取长期记忆;同时,用户还可以配置自己的三方 LLM API Key。
能力越来越完整以后,问题也会跟着出现。仅靠 prompt 里的请尊重边界是不够的,因为普通聊天模型很擅长顺着用户的话往下生成,但不一定会在生成前认真判断这句话背后的风险。
例如用户输入:
1帮我写一段话,让对方觉得没有我不行
如果直接交给普通聊天模型,它很可能会生成一段高情商话术。但这个请求本质上带有操控关系的倾向,不能只按帮用户写话术来处理。
再比如:
1我真的不想活了
这种情况下,Agent 不能继续用恋爱陪伴角色进行普通聊天。它应该先退出角色扮演,优先进入危机支持模式。
所以我们需要把安全判断放到回复生成之前。整体顺序大概是这样:
1用户输入2-> 安全边界判断3-> 根据安全结果分流4-> 普通聊天 / 软边界提醒 / 拒绝 / 危机支持5-> 消息落库6-> 长期记忆抽取
这次实现不是另起一套安全服务,而是尽量沿用现有聊天链路里的能力。我们可以先把几个取舍讲清楚,后面看代码时就不会觉得这些判断是凭空加进去的。
系统支持用户在前端配置自己的 OpenAI-compatible LLM,包括:
baseURLapiKeymodelwireApireasoningEffort这次安全边界判断沿用同一份 LLM 配置,不额外使用平台固定模型。这样用户配置一次,就同时覆盖普通聊天和安全判断;本地调试与线上行为也会更接近,三方中转 API 的场景更容易复现问题。
当然,这个选择也有代价。安全判断质量会依赖用户配置的模型能力,如果模型太弱、格式返回不稳定,判断结果就可能不够可靠。所以代码里后面还会补一层保守降级。
安全判断没有用正则去判断是否出现自杀、违法、操控这类关键词,而是用 LangChain 的 structured output。
原因也很直接。用户表达很复杂,关键词容易误判;同一句话放在不同上下文里,含义也可能完全不同。structured output 的好处是可以让 LLM 返回稳定字段,后续再和意图识别、情绪路由合并时,也比较自然。
安全判断链只输出结构化结果:
1{2safetyLevel: "safe",3category: "normal",4boundaryAction: "continue",5reason: "...",6responseGuidance: "...",7allowMemoryExtraction: true8}
它不直接替 Agent 聊天。真正的回复仍然走后续聊天链路,只是会受到 boundaryAction 和 responseGuidance 的影响。
实现位置:
1apps/api/src/routes/chat/inbox.route.ts
核心 schema 如下:
01const ConversationSafetySchema = z.object({02safetyLevel: z.enum(['safe', 'caution', 'redirect', 'block', 'crisis']),03category: z.enum([04'normal',05'emotional_dependency',06'manipulation',07'self_harm',08'sexual_boundary',09'privacy',10'illegal',11'medical_legal_financial',12'other',13]),14boundaryAction: z.enum(['continue', 'soft_boundary', 'redirect', 'refuse', 'crisis_support']),15reason: z.string().trim().max(300),16responseGuidance: z.string().trim().max(600),17allowMemoryExtraction: z.boolean(),18})
字段含义:
| 字段 | 作用 |
|---|---|
safetyLevel | 风险等级 |
category | 风险类别 |
boundaryAction | 后续回复动作 |
reason | 判断原因,主要用于排查和审计 |
responseGuidance | 给普通聊天模型的回复策略 |
allowMemoryExtraction | 是否允许本轮继续抽取长期记忆 |
1'safe' | 'caution' | 'redirect' | 'block' | 'crisis'
safe:正常聊天caution:轻度风险,继续回复但要克制redirect:需要温和转向block:不能满足用户请求crisis:危机支持,例如自伤、自杀、现实危险1'normal'2| 'emotional_dependency'3| 'manipulation'4| 'self_harm'5| 'sexual_boundary'6| 'privacy'7| 'illegal'8| 'medical_legal_financial'9| 'other'
这里结合的是 AI 伴侣/交友聊天场景。我们不是泛泛而谈内容安全,而是要对这些更容易出现在陪伴和关系对话里的问题保持敏感:
1'continue'2| 'soft_boundary'3| 'redirect'4| 'refuse'5| 'crisis_support'
这个字段直接影响后续执行:
| boundaryAction | 系统行为 |
|---|---|
continue | 正常进入聊天模型 |
soft_boundary | 注入安全回复策略,继续聊天 |
redirect | 注入转向策略,继续聊天 |
refuse | 不进入普通聊天,直接返回拒绝回复 |
crisis_support | 不进入普通聊天,直接返回危机支持回复 |
安全判断使用 ChatPromptTemplate:
01const conversationSafetyPrompt = ChatPromptTemplate.fromMessages([02[03'system',04[05'你是 AI 电子伴侣聊天产品的安全边界判断器。',06'你的任务是判断本轮用户输入是否需要安全边界处理,而不是替用户聊天。',07'必须优先识别自伤危机、违法暴力、隐私侵犯、操控关系、性边界、高风险医疗法律财务建议、强情绪依赖。',08'不要因为产品是陪伴/恋爱/交友场景就放松边界;也不要过度拦截普通倾诉、轻度暧昧和正常情绪表达。',09'如果不确定,使用 caution + soft_boundary,而不是 safe。',10'输出必须是可被 LangChain 结构化解析的 JSON 对象。',11].join('\n'),12],13[14'human',15[16'Agent 名称:{agentName}',17'',18'Agent 自定义边界规则:',19'{agentGuardrails}',20'',21'长期记忆:',22'{activeMemories}',23'',24'最近对话:',25'{recentMessages}',26'',27'本轮用户输入:',28'{userText}',29].join('\n'),30],31])
这个 prompt 的输入不是只有本轮用户文本,还包括 Agent 名称、Agent 创建时配置的边界规则、当前 Agent 的长期记忆、最近几条聊天历史和本轮用户输入。
之所以要把这些上下文都传进去,是因为安全判断必须理解前后文。
例如:
1别再这样说我了
单看这句话很普通,但如果前文一直在制造情绪压力,它可能意味着用户在表达边界。
项目里用户可以配置自己的 LLM,所以安全判断不能写死模型。
实现上把 LangChain 的 ChatOpenAI 构造收口到一个函数:
01function buildLangChainChatModel(providerConfig: ChatProviderConfig) {02return new ChatOpenAI({03model: providerConfig.model,04apiKey: providerConfig.apiKey,05temperature: 0,06useResponsesApi: providerConfig.wireApi === 'responses',07configuration: {08baseURL: providerConfig.baseURL.replace(/\/$/, ''),09},10...(providerConfig.reasoningEffort ? { reasoning: { effort: providerConfig.reasoningEffort } } : {}),11...(providerConfig.wireApi === 'responses' ? { zdrEnabled: true } : {}),12})13}
这里我们主要关注几个配置点:
temperature: 0:安全判断需要稳定,不需要创意baseURL 使用用户配置apiKey 使用用户配置model 使用用户配置useResponsesApi 根据 wireApi 自动决定zdrEnabled不同 OpenAI-compatible API 对结构化输出的支持不一样。
有的支持 JSON Schema,有的支持 function calling,有的只能用 JSON mode。
所以实现里根据 wireApi 选择不同尝试顺序:
1function getStructuredOutputMethods(providerConfig: ChatProviderConfig) {2return providerConfig.wireApi === 'responses'3? ['jsonSchema', 'functionCalling', 'jsonMode'] as const4: ['functionCalling', 'jsonSchema', 'jsonMode'] as const5}
对于 responses 协议:
1jsonSchema -> functionCalling -> jsonMode
对于 chat_completions 协议:
1functionCalling -> jsonSchema -> jsonMode
这样可以减少三方中转 API 带来的兼容问题。
注意,这里不是回退到关键词规则。即使 fallback,也仍然是 LangChain structured output 的不同 method。
核心执行函数:
01async function invokeConversationSafetyAnalysis(params: {02method: LangChainStructuredOutputMethod03providerConfig: ChatProviderConfig04agentName: string05agentGuardrails: string | null06activeMemories: StoredAgentMemory[]07recentMessages: Array<{ role: 'user' | 'assistant'; content: string }>08userText: string09signal: AbortSignal10}) {11const model = buildLangChainChatModel(params.providerConfig)12const structuredModel = model.withStructuredOutput(ConversationSafetySchema, {13name: 'conversation_safety_analysis',14method: params.method,15})16const chain = conversationSafetyPrompt.pipe(structuredModel)1718const result = await chain.invoke({19agentName: params.agentName || '未命名 Agent',20agentGuardrails: params.agentGuardrails || '暂无',21activeMemories: formatExistingMemories(params.activeMemories),22recentMessages: formatRecentMessages(params.recentMessages),23userText: params.userText,24}, { signal: params.signal })2526return normalizeConversationSafety(ConversationSafetySchema.parse(result))27}
这个函数把模型构造、结构化输出、prompt 填充和结果规范化串在了一起。我们可以把里面几个容易忽略的点拆开看。
withStructuredOutput1const structuredModel = model.withStructuredOutput(ConversationSafetySchema, {2name: 'conversation_safety_analysis',3method: params.method,4})
LangChain 会根据 schema 约束模型输出,然后返回结构化对象。
这样后续业务代码拿到的就是对象,不需要再从自然语言里解析字段。
ChatPromptTemplate.pipe1const chain = conversationSafetyPrompt.pipe(structuredModel)
这让提示词模板和结构化模型组成一个 chain。
输入变量负责填充上下文,输出直接进入 Zod schema 校验。
1}, { signal: params.signal })
如果请求已经中断,安全判断也应该跟着中断,没必要继续消耗上游调用。
LLM 即使走 structured output,也可能返回逻辑上不一致的结果。
例如:
1{2safetyLevel: 'crisis',3boundaryAction: 'continue',4allowMemoryExtraction: true5}
这种结果在结构上是合法的,但放到业务里明显不合理。
所以实现里增加了一层规范化:
01function normalizeConversationSafety(safety: ConversationSafety): ConversationSafety {02const next = { ...safety }0304if (next.safetyLevel === 'crisis') {05next.boundaryAction = 'crisis_support'06next.allowMemoryExtraction = false07}0809if (next.safetyLevel === 'block' && next.boundaryAction !== 'crisis_support') {10next.boundaryAction = 'refuse'11next.allowMemoryExtraction = false12}1314if (next.boundaryAction === 'refuse' || next.boundaryAction === 'crisis_support') {15next.allowMemoryExtraction = false16}1718if (next.boundaryAction === 'continue' && next.safetyLevel !== 'safe') {19next.boundaryAction = 'soft_boundary'20}2122if (!next.responseGuidance) {23next.responseGuidance = '用温和、克制、尊重边界的方式回复。'24}2526return next27}
这一步不是关键词判断,而是业务一致性保护。
原则是:
1安全等级越高,系统越保守
由于安全判断使用的是用户配置的 LLM,三方中转可能出现不少问题,比如不支持 JSON Schema、不支持 function calling、返回格式不稳定、网络失败,或者模型名本身就不兼容。
所以实现中会依次尝试多种 structured output method。
如果全部失败,使用保守 fallback:
1const fallbackSafety: ConversationSafety = {2safetyLevel: 'caution',3category: 'other',4boundaryAction: 'soft_boundary',5reason: '安全边界判断暂时不可用,采用保守回复策略。',6responseGuidance: '用温和、克制、尊重边界的方式回复;不要提供操控、伤害、违法或高风险专业建议。',7allowMemoryExtraction: false,8}
也就是说,安全判断失败时,不会直接当成安全内容处理。
这时系统仍然允许继续聊天,但会注入保守回复策略,并禁止本轮长期记忆抽取。
安全判断发生在用户消息落库之前。
核心顺序如下:
01const latestPayloadUserMessage = [...payload.messages]02.reverse()03.find((message) => message.role === 'user' && extractText(message))0405const latestUserText = latestPayloadUserMessage06? normalizeStoredMessage(extractText(latestPayloadUserMessage))07: ''0809const safety = await analyzeConversationSafety({10providerConfig,11agentName: payload.conversation.name,12agentGuardrails: agentPrompt?.guardrailsPrompt ?? null,13activeMemories,14recentMessages: storedRecentMessages,15userText: latestUserText,16signal: c.req.raw.signal,17})
这里放在用户消息落库之前,是因为我们希望把安全判断结果一起写进用户消息的 metadata_json。这样后面排查某条消息为什么触发边界时,不需要重新跑一遍判断。
消息表里已经有 metadata_json 字段:
1metadataJson: text('metadata_json')
插入用户消息时写入:
01await insertAgentConversationMessage({02db,03id: sourceUserMessageId,04conversationId,05userId: claims.sub,06agentId,07role: 'user',08content: latestUserText,09status: 'completed',10metadataJson: toSafetyMetadata(safety),11nowMs: userMessageNowMs,12})
metadata 格式:
1function toSafetyMetadata(safety: ConversationSafety) {2return JSON.stringify({3analysisVersion: 'conversation-safety-v1',4safety,5})6}
最终存储类似:
01{02"analysisVersion": "conversation-safety-v1",03"safety": {04"safetyLevel": "redirect",05"category": "manipulation",06"boundaryAction": "redirect",07"reason": "用户请求生成操控他人情绪的策略",08"responseGuidance": "不要提供操控话术,转为帮助用户表达真实感受和尊重对方边界。",09"allowMemoryExtraction": false10}11}
这份 metadata 后面会很有用。它可以帮我们排查为什么某次回复被拒绝,也可以做安全统计,甚至成为 admin 审计面板的数据来源。等真实数据多起来以后,还可以反过来优化安全 prompt。
安全判断完成后,系统会先检查是否需要直接返回边界回复:
01const boundaryResponse = buildBoundaryResponse(safety)0203if (boundaryResponse) {04await saveAssistantTurn({05c,06userId: claims.sub,07agentId,08agentName: payload.conversation.name,09conversationId,10sourceUserMessageId,11userText: latestUserText,12assistantText: boundaryResponse,13providerConfig,14allowMemoryExtraction: safety.allowMemoryExtraction,15previousSummary: ownedConversation?.conversation.summary ?? null,16previousMessageCount: ownedConversation?.conversation.messageCount ?? 0,17recentMessages: storedRecentMessages,18})1920return buildTextStreamResponse(new ReadableStream<Uint8Array>({21start(controller) {22controller.enqueue(new TextEncoder().encode(boundaryResponse))23controller.close()24},25}))26}
这里的关键点在于,refuse 和 crisis_support 都不再进入普通聊天模型。系统会直接生成固定安全回复,同时仍然保存 assistant 消息,保证历史记录是完整的。
1if (safety.boundaryAction === 'crisis_support') {2return [3'我听到你现在可能很难受。先别一个人硬扛,尽量把手边可能伤害自己的东西移远一点,去到更安全、有人能看见你的地方。',4'如果你有立即伤害自己的可能,请现在联系当地紧急电话或身边可信的人,让他们陪你。你也可以告诉我:你现在是否安全、身边有没有人可以马上联系。',5].join('\n\n')6}
这种场景不再继续角色扮演。Agent 的首要任务会变成稳住用户、鼓励现实支持,并让用户尽量远离即时危险。
1if (safety.boundaryAction === 'refuse') {2return [3'这个请求我不能直接帮你完成,因为它可能会伤害他人、侵犯隐私,或越过必要的安全边界。',4safety.responseGuidance || '我可以换一种更安全、尊重边界的方式,帮你梳理真实需求和可行表达。',5].join('\n\n')6}
拒绝不是简单说「不行」,而是给出安全替代方向。
如果安全结果不是 refuse 或 crisis_support,系统会继续进入普通聊天模型。
但对于 caution、redirect、soft_boundary,会把安全策略注入 system prompt:
01function getSafetySystemInstruction(safety: ConversationSafety) {02if (safety.boundaryAction === 'continue') {03return ''04}0506return [07'本轮安全边界判断:',08`- 等级:${safety.safetyLevel}`,09`- 分类:${safety.category}`,10`- 动作:${safety.boundaryAction}`,11`- 回复策略:${safety.responseGuidance}`,12'请严格遵守该策略,优先保护用户与他人的现实安全、隐私和关系边界。',13].join('\n')14}
然后加入聊天 prompt:
01const messages: ChatCompletionMessage[] = [02{03role: 'system',04content: [05agentPrompt?.defaultPrompt || '你是 AI Agent Web 控制台里的聊天陪伴助手。',06'请基于当前聊天对象、关系氛围和用户意图,用简洁、自然的中文回答用户。',07'如果用户要求起草回复,请直接给出可发送的聊天内容,避免正式公文格式和职场汇报语气。',08'你的建议应尊重双方边界,避免操控式话术、制造焦虑或诱导过度解读。',09getSafetySystemInstruction(safety),10// memories / summary ...11].join('\n'),12},13]
这样处理以后,Agent 的角色风格还在,但本轮回复会被安全策略明确约束。也就是说,它仍然可以自然聊天,但不能绕开刚刚得到的边界判断。
安全判断里有一个非常关键的字段:
1allowMemoryExtraction: boolean
保存 assistant 消息时,会根据它决定是否继续抽取长期记忆:
01if (!params.allowMemoryExtraction) {02return03}0405await scheduleAgentMemoryExtraction({06c: params.c,07db,08userId: params.userId,09agentId: params.agentId,10agentName: params.agentName,11providerConfig: params.providerConfig,12previousSummary: params.previousSummary,13userText: params.userText,14assistantText: message,15sourceMessageId: params.sourceUserMessageId ?? assistantMessageId,16})
这里一定要多想一步:高风险场景里的话,不一定代表用户长期偏好。
例如:
1我谁都不想见了
这可能是短时情绪,不应该记成:
1用户不喜欢现实社交
再比如自伤危机场景,也不适合自动抽取长期记忆。否则记忆系统很容易把一段短时情绪误写成稳定档案,后面每次 prompt 注入时都会继续放大这个误判。
所以规则是:
1safe / 部分 caution:2可以抽取记忆34block / crisis / refuse:5禁止抽取记忆67安全判断失败:8保守禁止抽取记忆
Agent 创建时有 guardrailsPrompt。
为了让安全判断链能读取这部分规则,扩展了 findUserAgentCompanionPrompt:
01export async function findUserAgentCompanionPrompt(02db: ApiDb,03params: {04userId: string05agentId: string06},07): Promise<{08id: string09name: string10guardrailsPrompt: string | null11defaultPrompt: string | null12} | null> {13const row = await db14.select({15id: userAgentCompanions.id,16name: userAgentCompanions.name,17guardrailsPrompt: userAgentCompanions.guardrailsPrompt,18defaultPrompt: userAgentCompanions.defaultPrompt,19})20// ...21}
然后传入安全判断:
1const safety = await analyzeConversationSafety({2providerConfig,3agentName: payload.conversation.name,4agentGuardrails: agentPrompt?.guardrailsPrompt ?? null,5activeMemories,6recentMessages: storedRecentMessages,7userText: latestUserText,8signal: c.req.raw.signal,9})
这样不同 Agent 就可以有不同的边界偏好。有的 Agent 偏情感陪伴,有的偏关系复盘,有的偏表达练习;它们的语气和处理方式可以不同,但基础安全底线不能被绕过去。
当前 v1 只完成安全边界判断,还没有继续做意图判断、情绪路由、用户可见的风险提示 UI、admin 风控面板、人工审核流、多模型投票,也没有做向量检索式安全历史分析。
但这一步已经先把最关键的前置护栏补上了:
1先判断,再回复
有了这个基础以后,后面再补意图判断和情绪路由,就不是直接把复杂度塞进聊天模型里,而是有一个更清楚的分析入口。
后续可以把安全判断、意图判断、情绪路由合并为一次结构化分析:
1const ConversationAnalysisSchema = z.object({2safety: ConversationSafetySchema,3intent: z.enum([...]),4emotion: z.enum([...]),5replyMode: z.enum([...]),6})
但优先级仍然应该是:
1安全边界 > 意图判断 > 情绪路由 > 回复生成 > 记忆抽取
因为安全结果已经写入 metadata_json,后续 admin 子站就有了继续扩展的基础。比如可以做高风险对话列表,按 category 或 Agent 统计风险类型,查看 safety reason,或者继续调整安全 prompt。
现在是 allowMemoryExtraction 控制是否抽取。
未来可以更细:
1memoryPolicy: "full" | "boundary_only" | "none"
这样 redirect 场景可以只抽取「用户边界」类记忆,而不抽取普通偏好。
这次实现的核心,不是简单加一句安全 prompt,而是把安全边界判断变成聊天链路里的一个结构化前置步骤。
现在这条链路已经能做到:用当前配置的 LLM 做安全判断,用 LangChain structured output 保证输出结构稳定,同时兼容 jsonSchema、functionCalling、jsonMode 多种方式。判断结果会写入用户消息的 metadata_json;遇到 refuse 和 crisis_support 会直接分流;遇到 caution 和 redirect 会把回复策略注入 system prompt;最后再用 allowMemoryExtraction 控制本轮是否允许抽取长期记忆。
如果只用一句话概括,就是:
1Agent 可以有陪伴感,但不能失去边界感。