PowerCore 集成（PE Agent 实现说明）¶

当前章节还未完善

本文面向 PE Agent 本仓库开发者，说明 app/integrations/ai_power/ 如何在项目内落地 AI Power 平台协议：鉴权、HTTP、WebSocket、SSE 转发、业务 flow 与排错约定。

简化版说明：当前仓库已裁剪为 auth / user / agent 核心能力；智能体在 AI Power 平台侧预置，本地 Agent 表仅维护展示元数据。下文配置表与 flow 列表以当前代码为准。

若要把协议交给外部系统自行对接 AI Power，请参阅 PE Agent 集成（平台对接指南）。开放 API 见接口 API。

1. 集成边界¶

AI Power 是 PE Agent 依赖的外部智能体运行时。项目内约定：

业务模块优先从 app.integrations.ai_power 导入核心能力（鉴权、流式、任务 HTTP）；扩展 HTTP（read_directory、LLM Keys、LightRAG）可直接引用 http/ 子模块。
协议字段以 AI Power 平台为准，PE Agent 的实现说明只描述本仓库当前封装方式。
新任务必须先建 WebSocket，再调用 enter_agent，否则平台在 enter_agent 后立即推送的帧可能丢失。
WebSocket 按用户持有物理连接，按 link 隔离并发请求；最后一条 link 结束后立即关闭物理连接，不做跨请求常连。

业务模块 (auth / user / agent / tasks / scheduler)
        │
        ▼
app/integrations/ai_power/              ← 对外唯一入口 (__init__.py)
   ├── auth/          App token、user token、add_user / user_login
   ├── http/          REST 客户端、文件上传下载/目录读取、用户 LLM Keys、LightRAG
   ├── agents/        enter / refresh / modify / pause（tasks.py）
   ├── streaming/     上游 WS 帧 → 前端 SSE 事件
   ├── ws/            AiPowerChatWS，单连接多 link 路由
   ├── ws_pool/       按 user.id 管理活跃 WS 会话
   ├── parser/        executing / 交付物解析
   └── task_ids.py    task_id 规范化比较
        │
        ▼
AI Power 平台 (HTTP + WebSocket)

2. 配置¶

配置由 app/core/config.py 聚合成 settings.ai_power_paths。AI Power 根地址和应用凭据由 .env 提供，固定 Path 默认写在 config.py；如确需临时覆盖，仍可使用同名环境变量。

2.1 连接与路径¶

配置项	当前默认	用途
`AI_POWER_API_URL`	环境配置	AI Power HTTP 根地址
`AI_POWER_APP_ID` / `AI_POWER_APP_SKEY`	环境配置	应用级签名凭据
`AI_POWER_GET_ACCESS_TOKEN_PATH`	`/peagent/api/get_token`	获取 App `access_token`
`AI_POWER_GENERATE_ADD_USER_PATH`	`/peagent/api/add_user`	注册 AI Power 用户
`AI_POWER_GENERATE_USER_LOGIN_PATH`	`/peagent/api/user_login`	获取用户 `user_token`
`AI_POWER_ENTER_AGENT_PATH`	`/peagent/api/enter_agent`	创建任务
`AI_POWER_SOCKET_PATH`	`/peagent/api/chat_ws`	WebSocket 路径
`AI_POWER_AGENT_RE_GENERATE_PATH`	`/peagent/api/re_generate`	重新生成
`AI_POWER_AGENT_MODIFY_LLM_RESPONSE_PATH`	`/peagent/api/modify_llm_response`	编辑 LLM 回复后继续收流
`AI_POWER_PAUSE_TASK_PATH`	`/peagent/api/pause_task`	暂停任务
`AI_POWER_UPLOAD_FILE_PATH`	`/peagent/api/common/upload_file`	上传文件
`AI_POWER_DOWNLOAD_FILE_PATH`	`/peagent/api/common/download_file`	下载文件
`AI_POWER_REPLACE_MY_FILE_PATH`	`/peagent/api/replace_my_file`	替换个人文件内容（支持子目录）
`AI_POWER_READ_DIRECTORY_PATH`	`/peagent/common/read-directory`	读取目录内容（App token，见 §9）
`AI_POWER_USER_LLM_KEYS_PATH`	`/peagent/user-llm-keys`	用户大模型密钥（见 §9.2）
`AI_POWER_LIGHTRAG_OPEN_GRAPHS_PATH`	`/lightrag/open_graphs`	知识图谱（见 §9.3）

本地 Agent 表维护展示用元数据；运行时使用库中 agent_id 调用 enter_agent，不在 PE Agent 内创建/更新 AI Power 侧 Agent。

2.2 特殊业务 Agent（本地 category）¶

不再配置 AI Power Agent ID。将本地 agent.category 设为：

`category`	含义
`current`	即时任务清单
`scheduled`	定时任务清单
`experts`	专家团队（总入口；异常续跑接入点、隐藏消息落库）

task_note 由 task_note_type_for_agent_category 识别；专家团队 / 总入口逻辑由 is_experts_agent_category 识别。所有 Agent 的「结束」节点正文均不向前端透传（见 should_skip_agent_content）。

已移除的配置项（勿再添加）：AI_POWER_HUMAN_MACHINE_DELIVERY_AGENT_ID、AI_POWER_CURRENT_TASK_AGENT_ID、AI_POWER_SCHEDULED_TASK_AGENT_ID、hide_end_ids / AI_POWER_HIDE_END_AGENT_IDS、settings.ai_power_agents 等历史 Agent ID 配置。

3. 鉴权¶

AI Power 使用两级 token：

Token	获取方式	使用位置
`access_token`	`app_id + timestamp + app_skey` 计算 HMAC-SHA256 签名，调用 `get_token`	`add_user`、`user_login` 等系统级接口
`user_token`	用 PE Agent 用户 `external_user_id=str(user.id)` 调用 `user_login`	`enter_agent`、文件接口、LLM Keys、WS query token

3.1 App Token¶

实现位置：auth/app_token.py、auth/token_manager.py。

签名规则与平台指南一致：

raw = app_id + timestamp + app_skey
signature = HMAC_SHA256(key=app_skey, message=raw).hexdigest().lower()

请求头：

X-App-Id: {AI_POWER_APP_ID}
X-Timestamp: {unix_seconds}
X-Signature: {signature}

TokenManager 使用 Redis 缓存，key 为 ai_power:access_token。缓存写入时会把业务过期时间提前 100~300 秒，并额外保留约 1 小时陈旧缓存；刷新失败且仍有缓存时，会降级使用旧 token。

3.2 用户 Token 与用户注册¶

实现位置：auth/user.py、modules/auth/service.py。

注册 AI Power 用户：

{
  "external_user_id": "PE Agent user.id",
  "nickname": "user.name 或 PE_AGENT{id}",
  "tel": "user.phone"
}

注册返回的 user_id 会写入 User.ai_power_user_id。用户登录请求：

{
  "external_user_id": "PE Agent user.id",
  "nickname": "user.name"
}

TokenManager.get_user_token(user) 按 user.id 缓存 Redis key：ai_power:user_token:{user.id}，并按用户粒度加锁，避免并发刷新同一用户 token。

3.3 请求头约定¶

系统级接口：

{
    "Authorization": f"Bearer {access_token}",
    "Content-Type": "application/json",
}

用户级接口：

{
    "Authorization": f"Bearer {user_token}",
    "Content-Type": "application/json",
}

WebSocket 建连 URL 由 ws/protocol.py 生成：

wss://{AI_POWER_API_URL.host}{AI_POWER_SOCKET_PATH}?token={user_token}

4. HTTP 客户端规范¶

统一入口：http/client.py。

核心行为：

使用线程本地 requests.Session 复用 HTTP 连接。
所有 JSON 接口只认响应体 code == 200 为成功（get_direct() 除外，见下）。
非 JSON、HTTP 4xx/5xx、业务 code != 200 都会记录日志，并转换为统一前端文案 USER_FACING_ERROR = "服务暂时不可用，请稍后重试"。
文件上传使用 post_file()，会移除 Content-Type，让 requests 自动生成 multipart/form-data boundary。
文件下载使用 get_raw()，直接返回二进制 bytes。
get_lenient()：GET 请求，失败时不抛异常、返回 None；用于 read_directory 等可选资源加载。
get_direct()：GET 请求，上游直接返回 JSON 体（无 code/data 包装）；根地址与其它接口相同，走 AI_POWER_API_URL。

业务层不应直接拼 URL 或直接调用 requests，应使用 app.integrations.ai_power 导出的函数。

5. 任务级 HTTP 能力¶

实现位置：agents/tasks.py。

5.1 `enter_agent`¶

task_id = enter_agent(user, agent_id, message="")

请求体：

{
  "agent_id": "2157",
  "message": ""
}

注意：当前实现会把 agent_id 转为字符串提交。返回体必须包含 data.task_id，否则抛业务异常。

5.2 `task_refresh`¶

用于重新生成：

new_task_id = task_refresh(
    user,
    task_id,
    message_id,
    agent_id,
    node_id,
    content="",
)

请求体：

{
  "task_id": 987654321,
  "content": "用户补充说明",
  "message_id": "msg_xxx",
  "agent_id": 2157,
  "node_id": 100
}

平台会返回新的 task_id。调用方必须立刻执行：

ws.associate_task_with_link(new_task_id, link_id)

否则后续 WS 帧会因为无路由被丢弃。

5.3 `modify_llm_response`¶

用于编辑已有 LLM 回复：

modify_llm_response(user, task_id, content, message_id)

请求体：

{
  "task_id": 987654321,
  "content": "编辑后的全文",
  "message_id": "msg_xxx"
}

SSE 编辑入口使用 modify_llm_response_stream()：先建立 WS session 并绑定原 task_id，再调用 HTTP 修改接口，随后继续从 WS 接收 stream / refine。收到 refine 后，modules/agent/service/user_agent/modify.py 会更新本地消息与报告来源内容，并下发 message_full。

5.4 `pause_task`¶

pause_task(user, task_id)

请求体：

{
  "task_id": 987654321
}

入口在 modules/agent/service/user_agent/run_stream/context.py（SSE 异常或客户端断开时调用，用于终止当前执行）。

6. WebSocket 架构¶

6.1 核心类¶

AiPowerChatWS 位于 ws/chat_ws.py，由多个 mixin 组合：

Mixin	文件	职责
`ConnectionMixin`	`ws/connection.py`	建连、断连、重连
`ReceiverMixin`	`ws/receiver.py`	后台接收线程
`TransportMixin`	`ws/transport.py`	`send` 与 `_safe_recv`
`IncomingRouterMixin`	`ws/incoming_router.py`	入站帧校验、`task_id` 路由、任务切换
`LinkOpsMixin`	`ws/link_ops.py`	link 队列、`task_id -> link_id` 映射
`LinkErrorsMixin`	`ws/link_errors.py`	链路级错误传播
`CommandsMixin`	`ws/commands.py`	`dialog` / `confirm` / `recv_stream`

不要在接收线程外直接 ws.recv()；业务只消费 recv_stream()、dialog()、confirm() 这些按 link 隔离后的迭代器。

6.2 物理 WS 与 link¶

AiPowerWSPool 位于 ws_pool/pool.py。它按 user.id 维护一条当前活跃物理 WS，但这不是跨请求常连池：

with get_ai_power_ws_pool().session(user) as sess:
  → entry.ws 不存在时 connect()
  → begin_link(link_id)
  → yield sess
  → end_link(link_id)
  → 如果 open_link_count() == 0，立即关闭 socket，entry.ws = None

只有同一用户的多个 session() 在时间上重叠时，才会复用同一条物理连接并挂多个 link：

user.id
  └─ AiPowerChatWS（活跃物理连接）
       ├─ link-A → Queue(maxsize=2000)
       ├─ link-B → Queue(maxsize=2000)
       └─ link-C → Queue(maxsize=2000)

link_id 默认取当前 trace_id，没有 trace 时使用 uuid4().hex。SSE 入口会通过 link_id_holder 把当前 link 回写给断连收尾逻辑。

6.3 新任务标准顺序¶

SSE 主链路在 streaming/raw.py 的 run_agent_stream_raw() 中实现，标准顺序为：

1. pool.session(user) → WS connect + begin_link
2. enter_agent(user, agent_id, message) → task_id
3. ws.associate_task_with_link(task_id, link_id)
4. yield task_start，再消费 recv_stream(link_id=link_id)
5. session 退出 → end_link；最后一个 link 结束时关闭物理 WS

ws/session.py 仅保留 fill_link_id_holder()、iter_dict_messages() 等轻量辅助；历史上用于同步 flow 的 agent_task_session() / wait_until_confirm() 已随 flows/ 目录移除。

续聊 / 确认 / 刷新等场景见 §7.1 run_agent_stream_raw 行为表。

6.4 出站消息¶

dialog() 会发送：

{
  "type": "dialog",
  "data": {
    "task_id": 987654321,
    "content": "用户输入",
    "files": []
  }
}

约束：

task_id 必填。
content 与 files 至少一项非空。
files 存在时才会写入 data；仅附件对话可传 content=""。

confirm() 会发送：

{
  "type": "confirm",
  "data": {
    "task_id": 987654321
  }
}

6.5 入站路由¶

IncomingRouterMixin 的处理规则：

除 call_agent 外，入站帧必须能解析出 task_id，否则忽略。
active_task 用 pre_task_id 找原 link，再把新的 data.task_id 绑定到同一 link。
call_agent 用 caller 找原 link，再把 callee 绑定到同一 link。
普通帧用 data.task_id 或顶层 task_id 查 task_id -> link_id。
查不到 link 的帧会记录“无路由丢弃”并丢弃。

因此任何能产生新 task_id 的动作，都必须立即绑定路由：

enter_agent 返回初始 task_id
task_refresh 返回新 task_id
收到 active_task
收到 call_agent

6.6 超时与重连¶

关键常量位于 ws/constants.py：

常量	当前值	说明
`_INCOMING_QUEUE_MAXSIZE`	`2000`	每个 link 队列容量
`_TASK_TO_LINK_MAX`	`10000`	task 路由表上限
`_RECV_STALL_SECONDS`	`3600`	接收线程长期无任何帧时判定异常
`_RECONNECT_BACKOFF_CAP_SEC`	`5`	重连退避上限
`_RECONNECT_MIN_INTERVAL_SEC`	`2`	最小重连间隔

recv_stream() 默认：

首包超时 first_msg_timeout=60
后续空闲超时 subsequent_timeout=300
每 0.1s 轮询 link 队列，并传播链路错误

7. SSE 流式链路¶

入口：modules/agent/service/user_agent/run_stream/service.py 的 run_stream_service()。

run_stream_service
  ├─ run_agent_stream_raw()       # 生成上游 WS 原始帧
  ├─ AgentStreamAdapter.handle()  # 上游 type → 前端事件
  └─ sse_event(...)               # 输出 SSE

7.1 `run_agent_stream_raw`¶

实现位置：streaming/raw.py。

场景	条件	行为
新任务	无 `task_id`	WS session → `enter_agent` → 绑定 task → 先 yield `task_start` → 消费 enter 后上游帧
重新生成	`request_type == "refresh"`	WS session → `task_refresh` → 绑定新 task → `recv_stream`
确认	`request_type == "confirm"`	WS session → `ws.confirm(task_id)`
续聊 / 普通输入	其它有 `task_id` 的请求	WS session → `ws.dialog(task_id, message, files)`
编辑回复	`modify_llm_response_stream`	WS session → 绑定原 task → HTTP modify → `recv_stream`

若不是 confirm / init / refresh，且没有 message 和 files，会直接返回 message不能为空。

7.2 上游 type 映射¶

streaming/handlers/__init__.py 当前注册：

HANDLERS = {
    "stream-start": handle_stream_start,
    "stream": handle_stream,
    "executing": handle_executing,
    "confirm": handle_confirm,
    "clarify": handle_clarify,
    "executed": handle_executed,
    "stream-end": handle_stream_end,
    "active_task": handle_active_task,
    "call_agent": handle_call_agent,
    "rate_limit_blocked": handle_rate_limit_blocked,
    "del_message": handle_del_message,
    "user_message": handle_user_message,
}

处理器文件分布：

文件	处理的 type
`handlers/stream.py`	`stream-start` / `stream` / `stream-end`
`handlers/executing.py`	`executing`
`handlers/task_lifecycle.py`	`confirm` / `clarify` / `executed` / `rate_limit_blocked`
`handlers/system.py`	`active_task` / `call_agent`
`handlers/user_message.py`	`user_message`
`handlers/del_message.py`	`del_message`

前端常见 SSE 事件：

SSE type	来源	内容
`task_start`	`run_agent_stream_raw`	新任务 `task_id`
`message_chunk`	`stream`	主任务增量文本
`note_chunk`	`stream`	子任务 / 外部任务增量笔记
`message_full`	`executing` 或 `refine`	当前完整消息及按钮状态（含 `show_actions`）
`note_full`	`executing.task_generated_data` 或外部任务消息	笔记全文
`task_update`	`executing.task_progress/task_context`	节点进度
`message_display`	`executing` 当前消息 `display`	控制正文展示
`deliverables_for_node`	`executing` 当前消息文件元数据	节点交付物文件（内部事件）
`deliverables_for_agent`	`executing`，`node_name == "运行代码"`	内部 adapter 事件（不直接下发前端）；经 `events.py` 转为 `deliverable_data` + 成功提示流
`deliverable_data`	`deliverables_for_node` / `deliverables_for_agent` 落库后	节点/Agent 交付物，写入 `agent_run_deliverable` 并下发前端
`active_task`	`active_task`	活动任务切换（subagent 续跑时会替换 `external_task_ids` 中的旧 task）
`user_message`	`user_message`	上游逻辑自动产生的用户侧消息（非真实用户输入）
`del_message`	`del_message`	上游通知删除某条 assistant 消息（本地隐藏对应记录）
`rate_limit_blocked`	`rate_limit_blocked`	上游限流，含 `reason` 正文后结束本轮
`done`	`confirm` / `executed` / 编辑完成 / 限流	本轮结束

内部事件（不下发前端）：clarify_waiting — 上游 clarify 通知需补充信息，仅设置运行记录的 waiting_clarify 标志并结束本轮 SSE。

已移除的产品侧形态：message_full.deliverable_confirm 确认按钮（SSE 不再下发该字段）。

须保留的活跃逻辑（勿当死代码清理）：

deliverables_for_agent 内部分支：handlers/executing.py 在 node_name == "运行代码" 时产出内部事件 → run_stream/events.py → deliverables.py 落库并下发 deliverable_data 与模拟成功消息流。
skills 占位：adapter_helpers.py（start_skills_placeholder、absorb_skills_placeholder_stream_chunk、tick_skills_placeholder_tokens 等）+ handlers/stream.py / executing.py + run_stream/service.py token tick（0.5s）；上游 skills 执行等待期间下发占位 message_chunk（有工具状态时带 status.type / status.token_text + 增量正文；无工具状态时可见消息不带 status）。display=false / 子线 hidden 时与原先一致，改为 message_full 只下发 token。真实回复不再走假流，直接下发 message_full。
clarify 续跑占位：start_clarify_skills_placeholder() 在 clarify 续跑首帧启动 skills 占位（不置 in_skills），由 handlers/stream.py / executing.py 调用。
user_message 抑制：should_suppress_auto_user_message() — 最后一条可见消息仍是 user/tools、尚无 assistant 回复时，跳过上游自动 user_message 的落库与 SSE 下发。

7.3 `executing` 解析¶

主要解析函数：parser/executing.py。

当前消息定位路径：

data.dialog.dialog.agents[0].current_state.last_message_id
data.dialog.dialog.agents[0].messages[]

parse_run_message_snapshot() 返回：

content, message_id, role, status, node_name, node_id, ai_power_agent_id, display

其中：

status 来自 message.metadata.status，缺省为 pending。
display 缺省为 True，显式 false 会保留；路由节点 会被强制视为不展示。
file_name / file_size 由 parse_last_message_file_meta() 从当前消息读取，用于 deliverables_for_node。

7.4 展示与按钮规则¶

平台指南说明 display 是上游消息字段；PE Agent 当前实现只用它控制消息正文展示：

message_display = (display is not false) and node_name != "路由节点"

按钮由 streaming/handlers/executing.py 根据当前消息计算：

role == "system"：next / refresh / modify 全部关闭。
request_type == "init" 且本次请求无用户文本/附件：交互按钮关闭。
status == "completed"：next 关闭，但 refresh / modify 仍可按其它条件展示。
内容含 点击“确认”：next 强制开启。
内容含 这次有什么可以帮您？：next 关闭。
正文含 <HRP_no_next>：next 关闭。
正文含 ```questions 的 message_full / note_full：next / modify 关闭。
node_name == "运行代码" 且 role == "assistant"：next 关闭，并走 deliverables_for_agent 内部分支（见 §7.1）。
show_actions：parser/text_extract.py 从 assistant 正文中解析 <show_right> / <show_left> / <show_down> 标签，剥离后写入 message_full.show_actions（{right, left, down}），并落库到 agent_run_message.show_actions。

confirm 拦截（handlers/task_lifecycle.py）：

活跃路由节点（名称含 调用智能体 或 路由节点，且 status 为 running / waiting_user）上的 confirm 会被拦截，不下发 done。
skills 占位期间及 auto_next && auto_run 自动续跑场景有额外拦截逻辑（主/子线各自维护状态机）。

这意味着 PE Agent 的按钮策略不完全等同于平台原始字段；前端应以 SSE 的 message_full 字段为准。

8. 业务 Flow 调用模式¶

对外导出见 app/integrations/ai_power/__init__.py：

导出	典型场景
`run_agent_stream_raw` + `AgentStreamAdapter`	用户 Agent 对话 SSE（`run_stream/service.py`）
`modify_llm_response_stream`	编辑 LLM 回复 SSE（`modify.py`）
`upload_file` / `download_file` / `replace_my_file`	聊天附件与交付物文件
`pause_task` / `force_close_user_ws`	断连收尾、暂停上游任务
`add_user`	注册时同步 AI Power 用户

未从 __init__.py 导出、由业务模块直接引用的 HTTP 能力：

模块	典型场景
`http/files.read_directory`	Agent Tab placeholder、用户学习计划/自述加载（`base_agent.py`）
`http/user_llm_keys`	用户大模型密钥 CRUD（`modules/user/service.py`）
`http/lightrag.fetch_open_graphs`	知识图谱（`modules/agent/service/user_agent/lightrag.py`）

主要业务入口：

modules/agent/service/user_agent/run_stream/：用户 Agent SSE 主链路。
modules/agent/service/user_agent/modify.py：编辑回复 SSE。
modules/agent/service/base_agent.py：Agent Tab 列表；通过 read_directory 加载入口介绍 placeholder、用户学习计划等。
modules/user/service.py：用户大模型密钥 CRUD 中转。
modules/agent/api/user_agent.py：GET /user-agent/lightrag/open_graphs 知识图谱。
app/tasks/agent.py：异步生成会话标题（agent.generate_task_name）。
app/scheduler/：定时任务执行，复用 run_stream_core。

9. 文件与扩展 HTTP 接口¶

实现位置：http/files.py、http/upload_helpers.py、http/user_llm_keys.py、http/lightrag.py。

9.1 文件接口¶

函数	平台接口	说明
`upload_file(user, file)`	`POST common/upload_file`	`multipart/form-data`，字段名由实现封装
`download_file(user, file_name, task_id=None)`	`GET common/download_file`	返回二进制
`replace_my_file(user, ...)`	`POST replace_my_file`	替换个人文件（笔记编辑等）
`read_directory(path, recursive=False)`	`GET common/read-directory`	读取目录内容；使用 App token + `get_lenient()`，失败返回 `{}`

read_directory 典型调用：

智能体入口介绍 — 按文件名 stem 索引各 Tab 的 placeholder 文案。
{phone}/学习文件夹 — 读取 学习计划.txt 作为 planning Tab placeholder。
{phone}/个人信息 — 读取 用户自述.txt，用于教学板块资料完整性提醒。

平台上传返回的 saved_path / file_size 会作为 WS dialog.data.files[] 的文件引用。对话只发附件时，content 可为空字符串。

9.2 用户大模型密钥¶

实现位置：http/user_llm_keys.py → modules/user/service.py → modules/user/api.py。

HTTP	集成函数	说明
`POST /user/llm-keys`	`create_user_llm_key`	新增密钥
`GET /user/llm-keys`	`list_user_llm_keys`	分页列表
`PUT /user/llm-keys/{key_id}`	`update_user_llm_key`	修改密钥
`DELETE /user/llm-keys/{key_id}`	`delete_user_llm_key`	删除密钥

均使用用户 token（build_user_headers），路径前缀由 AI_POWER_USER_LLM_KEYS_PATH 配置。

9.3 知识图谱（LightRAG）¶

实现位置：http/lightrag.py → modules/agent/service/user_agent/lightrag.py。

HTTP	集成函数	说明
`GET /user-agent/lightrag/open_graphs`	`fetch_open_graphs`	查询参数：`label`（默认 `*`）、`max_depth`（默认 3）、`max_nodes`（默认 1000）

与常规 AI Power 接口不同：响应为直接 JSON 体（无 code/data 包装），由 get_direct() 处理；根地址同样走 AI_POWER_API_URL。

10. 标准时序¶

10.1 SSE 新任务：无 `task_id`¶

1. run_stream_service 创建/加载运行记录
2. run_agent_stream_raw → pool.session → WS connect
3. enter_agent → task_id → associate_task_with_link
4. yield task_start
5. recv_stream 消费 enter 后推送
6. AgentStreamAdapter 转为 SSE
7. confirm / executed 后下发 done

10.2 SSE 续聊、确认、刷新¶

续聊：   session → 直接 ws.dialog(task_id, message/files)
确认：   session → ws.confirm(task_id)
刷新：   session → task_refresh(...) → 新 task_id → associate → recv_stream
编辑：   session → associate 原 task_id → modify_llm_response → recv_stream 至 refine

10.3 客户端断开¶

SSE 使用 guard_sse_stream_with_client_disconnect 与 link_id_holder。客户端断开时，utils/http_sync_disconnect.py 会调用：

force_close_user_ws(user_id, link_id=link_id)

如果能拿到 per-user 锁，会结束指定 link；拿不到锁时会软终止对应消费端，避免底层生成器未关闭导致 session 的 finally 无法及时执行。

11. 排错¶

现象 / 日志	常见原因	处理
`AI_POWER ws 无路由丢弃`	`enter_agent` / `task_refresh` 后未绑定 task；或 `active_task/call_agent` 映射缺失	确认调用了 `associate_task_with_link`，并检查 task_id 类型
`recv_stream 空闲超时`	上游未推送、平台任务卡住、超时配置过小	查看上游任务状态；必要时调大 timeout
`当前链路无消息`	link 队列暂时没有帧	若持续出现，检查 task 路由与平台是否仍在执行
`获取 AI_POWER WebSocket 会话槽超时`	同一用户 session attach/release 锁被占用，常见于 SSE 未正常收尾	检查断连处理与 `force_close_user_ws`
接收线程死亡 / pending error	上游 WS 断开或协议异常	依赖重连；多次失败后会向 link 消费端抛统一错误
HTTP `code != 200`	平台业务错误、token 过期、路径或 payload 不匹配	查看 `app.integrations.ai_power.http.client` 日志中的 payload
文件上传失败	上传仍带 JSON `Content-Type` 或字段不符	使用 `post_file()` / `upload_file()`，不要手写请求
前端按钮状态与平台字段不一致	PE Agent 在 SSE 层二次计算按钮	以前端收到的 `message_full` 字段为准
上游自动 user_message 未出现	上一条可见消息尚无 assistant 回复，被 `should_suppress_auto_user_message` 抑制	预期行为；待 assistant 回复后再收后续自动消息
LightRAG 接口失败	`AI_POWER_API_URL` 未配置或路径错误	检查 `AI_POWER_API_URL` 与 `AI_POWER_LIGHTRAG_OPEN_GRAPHS_PATH`
read_directory 返回空	目录不存在或 App token 无效	查看 `[read_directory]` 日志；失败时 `get_lenient` 静默返回 `{}`

常用 logger：

app.integrations.ai_power.auth.token_manager
app.integrations.ai_power.http.client
app.integrations.ai_power.agents.tasks
app.integrations.ai_power.ws
app.integrations.ai_power.ws_pool

12. 开发约定¶

13. 目录索引¶

app/integrations/ai_power/
├── __init__.py
├── auth/
│   ├── app_token.py
│   ├── token_manager.py
│   └── user.py
├── http/
│   ├── base_url.py
│   ├── client.py          # get/post/put/delete/get_lenient/get_direct/get_raw/post_file
│   ├── files.py           # upload/download/read_directory/replace_my_file
│   ├── upload_helpers.py
│   ├── user_facing.py
│   ├── user_llm_keys.py   # 用户大模型密钥 CRUD
│   └── lightrag.py        # 知识图谱 open_graphs
├── agents/
│   └── tasks.py          # enter_agent / task_refresh / modify / pause
├── parser/
│   ├── executing.py
│   ├── deliverables.py
│   ├── task_progress.py
│   └── text_extract.py   # show_actions / show_content / questions 解析
├── streaming/
│   ├── adapter.py
│   ├── adapter_helpers.py
│   ├── adapter_context.py
│   ├── raw.py
│   └── handlers/
│       ├── __init__.py
│       ├── stream.py
│       ├── executing.py
│       ├── task_lifecycle.py
│       ├── system.py
│       ├── user_message.py
│       └── del_message.py
├── ws/
├── ws_pool/
└── task_ids.py

PowerCore 集成（PE Agent 实现说明）¶

1. 集成边界¶

2. 配置¶

2.1 连接与路径¶

2.2 特殊业务 Agent（本地 category）¶

3. 鉴权¶

3.1 App Token¶

3.2 用户 Token 与用户注册¶

3.3 请求头约定¶

4. HTTP 客户端规范¶

5. 任务级 HTTP 能力¶

5.1 enter_agent¶

5.2 task_refresh¶

5.3 modify_llm_response¶

5.4 pause_task¶

6. WebSocket 架构¶

6.1 核心类¶

6.2 物理 WS 与 link¶

6.3 新任务标准顺序¶

6.4 出站消息¶

6.5 入站路由¶

6.6 超时与重连¶

7. SSE 流式链路¶

7.1 run_agent_stream_raw¶

7.2 上游 type 映射¶

7.3 executing 解析¶

7.4 展示与按钮规则¶

8. 业务 Flow 调用模式¶

9. 文件与扩展 HTTP 接口¶

9.1 文件接口¶

9.2 用户大模型密钥¶

9.3 知识图谱（LightRAG）¶

10. 标准时序¶

10.1 SSE 新任务：无 task_id¶

10.2 SSE 续聊、确认、刷新¶

10.3 客户端断开¶

11. 排错¶

12. 开发约定¶

13. 目录索引¶

5.1 `enter_agent`¶

5.2 `task_refresh`¶

5.3 `modify_llm_response`¶

5.4 `pause_task`¶

7.1 `run_agent_stream_raw`¶

7.3 `executing` 解析¶

10.1 SSE 新任务：无 `task_id`¶