OpenHands服务：这AI框架如何应对疑问？

AI Agent 框架探秘：拆解 OpenHands（4）--- 服务 目录AI Agent 框架探秘：拆解 OpenHands（4）--- 服务0x00 概述0x01 服务1.1 API 模式1.1.1 Actions1.1.2 observation1.2 服务器组件session.pysession/agent_session.pysession/conversation_manager/conversation_manager.pylisten.py1.3 服务工作流程描述1.4 listen_socket.py1.4.1 核心特色1.4.2 具体功能1.4.3 流程图1.4.4 会话连接1.4.5 代码0xFF 参考 0x00 概述 本篇结合官方文档进行解读OpenHands的服务器，这是OpenHands系统的立身基础。 因为本系列借鉴的文章过多，可能在参考文献中有遗漏的文章，如果有，还请大家指出。 0x01 服务 OpenHands提供了WebSocket服务器。 1.1 API 模式 可以发送或从服务器接收两种类型的消息： Actions Observations 1.1.1 Actions 一个action 包含三个部分： action：要采取的动作 args：动作的参数 message：可以放在聊天记录中的友好消息 有几种action 。它们的参数列在下面。 随着时间的推移，这个列表可能会增长。 initialize - 初始化代理。仅由客户端发送。 model - 要使用的模型名称 directory - 工作空间的路径 agent_cls - 要使用的代理类 start - 开始一个新的开发任务。仅由客户端发送。 task - 要开始的任务 read - 读取文件内容。 path - 要读取的文件路径 write - 写入内容到文件。 path - 要写入的文件路径 content - 写入文件的内容 run - 运行命令。 command - 要运行的命令 browse - 打开网页。 url - 要打开的URL think - 允许代理制定计划、设定目标或记录想法 thought - 要记录的想法 finish - 代理发出任务完成的信号 1.1.2 observation 一个observation 包含四个部分： observation：观察类型 content：表示观察数据的字符串 extras：额外的结构化数据 message：可以放在聊天记录中的友好消息 有几种observation 。它们的额外信息列在下面。 随着时间的推移，这个列表可能会增长。 read - 文件内容 path - 读取的文件路径 browse - URL的HTML内容 url - 打开的URL run - 命令的输出 command - 运行的命令 exit_code - 命令的退出代码 chat - 用户的消息 1.2 服务器组件 以下部分描述了OpenHands项目的服务器端组件。 session.py session.py 文件定义了Session类，它代表与客户端的WebSocket会话。关键特性包括： 处理WebSocket连接和断开 初始化和管理代理会话 在客户端和代理之间分发事件 向客户端发送消息和错误 session/agent_session.py agent_session.py 文件包含AgentSession类，它管理会话内代理的生命周期。关键特性包括： 创建和管理运行时环境 初始化代理控制器 处理安全分析 管理事件流 session/conversation_manager/conversation_manager.py conversation_manager.py 文件定义了ConversationManager类，它负责管理多个客户端会话。关键特性包括： 添加和重启会话 向特定会话发送消息 清理非活动会话 listen.py listen.py 文件是主服务器文件，它设置FastAPI应用程序并定义各种API端点。关键特性包括： 设置CORS中间件 处理WebSocket连接 管理文件上传 提供代理交互、文件操作和安全分析的API端点 为前端提供静态文件服务 该脚本定义了服务接口，主要分为两个部分： 一部分是通过FastAPI库实现的HTTP接口，其具体实现位于openhands/server/routes目录中； 另一部分是利用socketio库实现的WebSocket接口，其代码实现在openhands/server/listen_socket.py文件中。用户与代理的交互通过WebSocket进行，连接初始化时会触发connect事件，用户发送消息时会触发oh_user_action事件，连接断开时会触发disconnect事件。因此，梳理代理交互逻辑的核心在于对这三个事件的处理流程进行整理。 import socketio from openhands.server.app import app as base_app from openhands.server.listen_socket import sio from openhands.server.middleware import ( CacheControlMiddleware, InMemoryRateLimiter, LocalhostCORSMiddleware, RateLimitMiddleware, ) from openhands.server.static import SPAStaticFiles if os.getenv('SERVE_FRONTEND', 'true').lower() == 'true': base_app.mount( '/', SPAStaticFiles(directory='./frontend/build', html=True), name='dist' ) base_app.add_middleware(LocalhostCORSMiddleware) base_app.add_middleware(CacheControlMiddleware) base_app.add_middleware( RateLimitMiddleware, rate_limiter=InMemoryRateLimiter(requests=10, seconds=1), ) app = socketio.ASGIApp(sio, other_asgi_app=base_app) 1.3 服务工作流程描述 服务的工作流程如下： 服务器初始化： FastAPI应用程序在listen.py中创建和配置。 设置CORS中间件和静态文件服务。 初始化ConversationManager。 客户端连接： 当客户端通过WebSocket连接时，创建新的Session或重启现有一个。 Session初始化AgentSession，设置运行时环境和代理控制器。 代理初始化： 客户端发送初始化请求。 服务器根据提供的参数创建和配置代理。 设置运行时环境，初始化代理控制器。 事件处理： Session管理客户端和代理之间的事件流。 客户端的事件分发到代理。 代理的观察结果发送回客户端。 文件操作： 服务器处理文件上传，确保它们符合大小和类型限制。 通过运行时环境执行文件读取和写入操作。 安全分析： 如果配置了，每个会话初始化安全分析器。 安全相关的API请求转发到安全分析器。 会话管理： ConversationManager定期清理非活动会话。 它还在需要时处理向特定会话发送消息。 API端点： 提供各种API端点，用于代理交互、文件操作和获取配置默认值。 这种服务器架构允许管理多个客户端会话，每个会话都有自己的代理实例、运行时环境和安全分析器。事件驱动设计促进了客户端和代理之间的实时通信，而模块化结构允许轻松扩展和维护不同组件。 1.4 listen_socket.py listen_socket.py是 OpenHands 服务器端的 Socket.IO 事件监听器，负责处理客户端和服务器之间的实时双向通信，包括连接建立、事件回放、用户行动转发和连接断开四大核心场景，是客户端与后端会话、代理系统交互的桥梁。 1.4.1 核心特色 listen_socket.py的核心特色如下： 断点续传的事件回放：支持通过 latest_event_id 参数实现事件断点续传，客户端重连时仅回放未接收的事件，避免重复数据传输，提升连接效率。 严格的身份与权限校验：连接建立时校验会话 ID、API 密钥、用户身份（通过 Cookie 和 Authorization 头），确保会话安全性，防止未授权访问。 向后兼容的事件处理：保留 oh_action 处理器兼容旧版客户端，同时提供 oh_user_action 新版接口，平滑过渡不中断服务。 有序的事件推送逻辑：代理状态变更事件（AgentStateChangedObservation）最后发送，确保客户端先接收历史事件，再同步最新状态，避免状态不一致。 异步高效的事件处理：基于异步 IO（async/await）实现事件回放和转发，支持高并发连接，不阻塞主线程，提升系统吞吐量。 完善的错误处理：连接失败时主动断开无效连接，记录详细日志，便于问题排查；过滤无效事件（如 NullAction），减少不必要的网络传输。 1.4.2 具体功能 listen_socket.py的具体功能如下： 连接管理（connect 事件） 身份验证：验证连接参数中的 conversation_id 和 API 密钥 用户认证：通过 conversation_validator 验证用户身份 会话恢复：为已存在的会话重放事件流历史 事件重播：向新连接的客户端发送历史事件，包括过滤特定事件类型 会话加入：将客户端连接加入到对应的会话中 动作处理（oh_user_action 和 oh_action 事件） 用户动作接收：处理来自客户端的用户操作请求 事件转发：将用户动作转发到会话管理器进行处理 向后兼容：同时支持 oh_user_action 和 oh_action 事件处理（后者为兼容旧客户端保留） 断开连接处理（disconnect 事件） 连接清理：当客户端断开连接时，清理相关会话资源 状态管理：通知会话管理器客户端已断开连接 listen_socket.py的核心工作流程为： 连接建立： 解析查询参数（会话 ID、最新事件 ID等） 验证会话和用户身份 创建事件存储实例 事件历史重播： 为客户端重放会话历史事件 过滤掉 NullAction、NullObservation、RecallAction 等特定事件 确保 AgentStateChangedObservation 事件最后发送 会话加入： 将连接 ID 与会话关联 初始化会话设置 安全机制 API密钥验证：检查 SESSION_API_KEY 环境变量与查询参数中的密钥是否匹配 会话权限控制：通过 conversation_validator 验证用户是否有权访问定会话 错误处理 连接拒绝：在验证失败或出现错误时拒绝连接 异常传播：使用ConnectionRefusedError处理连接错误 异步清理：在连接被拒绝后异步断开连接 listen_socket.py 与其他组件关系 与EventStream紧密配合，负责事件的传输和分发 通过 connection_manager 管理会话状态 使用 event_to_dict 进行事件序列化以便通过网络传输 1.4.3 流程图 1.4.4 会话连接 此处关键一步为与会话管理器 ConversationManager 建立连接。 conversation_init_data = await setup_init_conversation_settings( user_id, conversation_id, providers_set ) agent_loop_info = await conversation_manager.join_conversation( conversation_id, connection_id, conversation_init_data, user_id, ) 1.4.5 代码 listen_socket.py的代码举例如下： @sio.event async def connect(connection_id: str, environ: dict) -> None: """ SocketIO连接事件处理器：客户端建立连接时触发，完成会话验证、事件回放、会话加入等初始化流程。 参数： connection_id: 客户端连接唯一标识（SocketIO分配） environ: WSGI环境变量字典，包含请求头、查询参数等信息 """ try: logger.info(f"SocketIO连接建立：connection_id={connection_id}") # 解析查询参数（从WSGI环境变量中提取QUERY_STRING） query_params = parse_qs(environ.get('QUERY_STRING', '')) # 解析最新事件ID（用于断点续传，默认-1表示从最开始回放） latest_event_id_str = query_params.get('latest_event_id', [-1])[0] try: latest_event_id = int(latest_event_id_str) except ValueError: logger.debug(f"无效的latest_event_id值：{latest_event_id_str}，默认设为-1") latest_event_id = -1 # 解析会话ID（必需参数，用于关联特定对话） conversation_id = query_params.get('conversation_id', [None])[0] logger.info(f"会话连接请求：conversation_id={conversation_id}, connection_id={connection_id}") # 解析提供者集合（如支持的LLM提供商列表，用于限制可用资源） raw_list = query_params.get('providers_set', []) providers_list = [] for item in raw_list: # 拆分逗号分隔的提供者名称，过滤空值 providers_list.extend(item.split(',') if isinstance(item, str) else []) providers_list = [p for p in providers_list if p] providers_set = [ProviderType(p) for p in providers_list] # 转换为ProviderType枚举类型 # 校验会话ID是否存在 if not conversation_id: logger.error("查询参数中缺少conversation_id") raise ConnectionRefusedError("缺少会话ID（conversation_id）") # 校验会话API密钥是否有效 if _invalid_session_api_key(query_params): raise ConnectionRefusedError("无效的会话API密钥") # 提取请求中的Cookie和Authorization头（用于用户身份验证） cookies_str = environ.get('HTTP_COOKIE', '') # WSGI环境中，HTTP头会转为"HTTP_前缀+下划线替换短横线"格式 authorization_header = environ.get('HTTP_AUTHORIZATION', None) # 创建会话验证器，校验用户身份（关联会话ID、Cookie、授权头） conversation_validator = create_conversation_validator() user_id = await conversation_validator.validate( conversation_id, cookies_str, authorization_header ) # 创建事件存储实例（用于读取会话历史事件） try: event_store = EventStore( conversation_id, conversation_manager.file_store, user_id ) except FileNotFoundError as e: logger.error(f"创建会话事件存储失败：conversation_id={conversation_id}, 错误={e}") raise ConnectionRefusedError(f"无法访问会话事件：{e}") agent_state_changed = None # 存储代理状态变更事件（最后单独发送） # 创建异步事件存储包装器，从latest_event_id+1开始回放事件（避免重复） async_store = AsyncEventStoreWrapper(event_store, latest_event_id + 1) # 异步回放历史事件（向客户端推送未接收过的事件） async for event in async_store: logger.debug(f"回放事件：{event.__class__.__name__}") # 跳过无效/召回类事件（无需推送给客户端） if isinstance( event, (NullAction, NullObservation, RecallAction), ): continue # 暂存代理状态变更事件（最后发送，确保客户端状态同步） elif isinstance(event, AgentStateChangedObservation): agent_state_changed = event # 其他事件直接推送给客户端 else: await sio.emit('oh_event', event_to_dict(event), to=connection_id) # 最后发送代理状态变更事件（确保客户端获取最新状态） if agent_state_changed: await sio.emit( 'oh_event', event_to_dict(agent_state_changed), to=connection_id ) logger.info(f"会话事件回放完成：conversation_id={conversation_id}") # 初始化会话设置（用户偏好、提供者配置等） conversation_init_data = await setup_init_conversation_settings( user_id, conversation_id, providers_set ) # 加入会话：关联connection_id与会话，启动代理循环 agent_loop_info = await conversation_manager.join_conversation( conversation_id, connection_id, conversation_init_data, user_id, ) # 校验会话加入结果 if agent_loop_info is None: raise ConnectionRefusedError("加入会话失败") logger.info(f"会话加入成功：conversation_id={conversation_id}, connection_id={connection_id}") except ConnectionRefusedError: # 发送错误后断开无效连接 asyncio.create_task(sio.disconnect(connection_id)) raise @sio.event async def oh_user_action(connection_id: str, data: dict[str, Any]) -> None: """ 处理客户端发送的用户行动事件（如用户输入、操作指令）。 参数： connection_id: 客户端连接ID data: 用户行动数据（字典格式，包含行动类型、内容等） """ # 将用户行动转发到事件流，由会话管理器处理 await conversation_manager.send_to_event_stream(connection_id, data) @sio.event async def oh_action(connection_id: str, data: dict[str, Any]) -> None: """ 兼容旧版客户端的行动事件处理器（保留用于向后兼容）。 注意：待所有客户端升级为使用oh_user_action后，可移除该处理器 目前用于支持正在进行中的旧会话，避免中断服务 """ await conversation_manager.send_to_event_stream(connection_id, data) @sio.event async def disconnect(connection_id: str) -> None: """ SocketIO断开连接事件处理器：客户端断开连接时触发。 参数： connection_id: 断开连接的客户端ID """ logger.info(f"SocketIO连接断开：connection_id={connection_id}") # 通知会话管理器，断开该连接与会话的关联 await conversation_manager.disconnect_from_session(connection_id) 0xFF 参考 https://docs.all-hands.dev/openhands/usage/architecture/backend 当AI Agent从“玩具”走向“工具”，我们该关注什么？Openhands架构解析【第二篇：Agent 相关核心概念】 克里 当AI Agent从“玩具”走向“工具”，我们该关注什么？Openhands架构解析【第一篇：系列导读】 克里 Coding Agent之Openhands解析(含代码) Arrow OpenHands 源码解读 一力辉