openYuanrong进阶教程——AI Agent 会话与亲和性调度

AI Agent 会话与亲和性调度AI Agent 会话功能专为交互式应用场景如 AI 智能体、多轮对话设计。它支持函数执行过程中的主动等待与外部唤醒并确保同一会话内的多次请求能够路由到同一个执行实例从而实现低延迟的交互体验。会话调度机制AI Agent 会话的调度逻辑由faasscheduler模块负责具有以下核心特点1. 会话亲和性 (Session Affinity)当一个请求携带sessionId时调度系统会尝试将其路由到该会话已经绑定的实例上。首次请求: 调度系统选择一个合适的实例并建立sessionId与instanceId的绑定关系。后续请求: 只要绑定关系有效所有相同sessionId的请求都会定向到同一实例。2. 弱亲和性与持久化AI Agent 会话默认为弱亲和TTL为0。生命周期: 当会话下的所有租约都释放后绑定关系可能会失效。状态恢复: 会话的上下文数据如历史记录持久化在分布式数据系统中。当新请求到达且原实例不可用时新实例会从数据系统重新加载会话状态。SDK 使用说明在启用了 AI Agent 会话的函数中可以通过Context获取SessionService来操作会话对象。Java SDK核心接口SessionSession对象提供了会话内同步和状态管理的能力。wait(long timeoutMs): 挂起当前执行线程等待输入。notify(JsonObject payload): 唤醒正在wait的线程。getInterrupted(): 检查当前会话是否已被外部中断。Java 使用示例importcom.google.gson.JsonObject;publicObjecthandle(Contextctx,JsonObjectinput){Sessionsessctx.getSessionService().loadSession(ctx.getSessionId());// 检查是否为 notify 请求假设用户在 payload 字段中传入通知内容if(input.has(action)notify.equals(input.get(action).getAsString())){sess.notify(input.getAsJsonObject(payload));returnNotified;}JsonObjectuserInputsess.wait(60000);if(userInputnull)returnTimeout;if(sess.getInterrupted())returnInterrupted;returnGot: userInput.toString();}Python SDK在 Python 函数实例中yr模块提供了类似的会话操作能力。核心接口yr.SessionServicesession.wait_for_notify(timeout_ms): 阻塞当前协程/线程等待通知。session.notify(payload): 发送通知。session.is_interrupted(): 检查会话是否被中断。Python 使用示例importyrdefhandle(ctx,input):session_idctx.get_session_id()sessionctx.get_session_service().load_session(session_id)ifinput.get(action)notify:session.notify(input.get(payload))returnNotified# 等待通知user_inputsession.wait_for_notify(60000)ifuser_inputisNone:returnWait timeoutifsession.is_interrupted():returnSession InterruptedreturnfReceived:{user_input}完整用例多轮对话 AI Agent以下是一个完整的 Java 示例演示了如何利用wait/notify实现一个简单的多轮交互。importorg.yuanrong.services.Context;importorg.yuanrong.services.session.Session;importorg.yuanrong.services.session.SessionService;importcom.google.gson.JsonObject;importjava.util.ArrayList;importjava.util.List;publicclassSimpleAgent{publicObjecthandle(Contextctx,JsonObjectinput){SessionServicesessServicectx.getSessionService();SessionsesssessService.loadSession(ctx.getSessionId());// 1. 处理通知/唤醒请求if(input.has(action)notify.equals(input.get(action).getAsString())){// 提取真正的通知载荷进行唤醒sess.notify(input.getAsJsonObject(payload));returnnull;// Notify 请求通常不需要返回业务结果}// 2. 主执行流程try{ctx.getLogger().log(Agent Started, SessionID: ctx.getSessionId());// 第一轮交互ctx.getStream().write(你好我是 AI 助手请问有什么可以帮您\n);JsonObjectinput1sess.wait(30000);// 等待用户输入 30 秒if(input1null)return等待超时;Stringmsg1input1.get(message).getAsString();ctx.getStream().write(收到指令msg1\n正在为您处理...\n);// 模拟处理过程并更新历史ListStringhistorynewArrayList(sess.getHistories());history.add(User: msg1);sess.setHistories(history);// 第二轮交互ctx.getStream().write(处理完成。您还有其他问题吗\n);JsonObjectinput2sess.wait(30000);if(input2null)return等待超时;if(sess.getInterrupted()){return会话已被外部中断;}Stringmsg2input2.get(message).getAsString();ctx.getStream().write(好的已收到您的进一步要求msg2\n);}catch(Exceptione){ctx.getLogger().log(Error: e.getMessage());}returnSession Completed;}}交互流程示意图第一轮交互逻辑挂起客户端发起POST /invocations(携带SessionID:S1)函数实例接收请求执行业务逻辑直到调用sess.wait()函数实例释放会话锁执行线程进入挂起状态等待唤醒第二轮交互唤醒与继续客户端发起第二个请求POST /invocations(相同SessionID:S1,Action:notify)函数实例接收请求由于会话亲和性该请求进入同一实例函数实例执行sess.notify(payload)唤醒之前挂起的线程函数实例通知请求处理完成并返回200 OK函数实例(原线程) 获取到notify传递的数据继续执行后续逻辑函数实例(原线程) 完成处理向客户端返回第一轮请求的最终结果