971 lines
40 KiB
Python
971 lines
40 KiB
Python
"""发送聊天消息用例。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.ai_provider import ProviderManager, ProviderManagerFactory
|
||
from app.ai_provider.types.chat import ChatChunk
|
||
from app.application.agent import (
|
||
Agent,
|
||
AgentConfig,
|
||
AgentEvent,
|
||
AgentRunContext,
|
||
AgentRunRegistry,
|
||
ConfirmationRegistry,
|
||
ExtensionRegistry,
|
||
PythonSkillCatalog,
|
||
PythonSkillDefinition,
|
||
activate_python_skills,
|
||
compose_system_messages,
|
||
default_tools,
|
||
next_step_for_tool,
|
||
run_agent_turn,
|
||
step_event_payload,
|
||
task_from_deliberative_plan,
|
||
task_plan_payload,
|
||
)
|
||
from app.application.tool_runtime.runtime import ToolRuntime
|
||
from app.application.tool_runtime.view import RunScopedToolView
|
||
from app.core.config import settings
|
||
from app.core.errors import AppError, current_trace_id
|
||
from app.core.logging import logger
|
||
from app.core.time import utc_now
|
||
from app.db.session import AsyncSessionLocal
|
||
from app.models.chat import ChatMessage
|
||
from app.modules.agent import DEFAULT_CHAT_AGENT
|
||
from app.modules.chat.history import history_pairs
|
||
from app.modules.chat.policies import validate_text_message
|
||
from app.modules.goal_orchestrator import extract_session_goal, format_goal_directive
|
||
from app.modules.memory.context import build_memory_context
|
||
from app.modules.runtime.contract import RuntimeContract
|
||
from app.modules.runtime.events import EventStore
|
||
from app.modules.safety.confirmation import ConfirmationStore
|
||
from app.repositories import chat as chat_repo
|
||
from app.repositories import runtime as runtime_repo
|
||
from app.repositories.json_utils import dumps, loads_dict
|
||
from app.schemas.chat import SendMessageRequest, SendMessageResponse
|
||
from app.services import runtime as runtime_service
|
||
from app.services.audit import record_audit_event
|
||
from app.services.characters import default_model_hint, resolve_active_character
|
||
from app.services.chat import recent_chat_messages
|
||
from app.services.memory import retrieve_memories
|
||
from app.services.personality import build_personality_summary
|
||
from app.services.runtime import TERMINAL_RUN_STATUS
|
||
from app.services.safety import decide_chat_safety
|
||
|
||
# 后台流式任务引用：asyncio.create_task 仅保留弱引用，需强引用防止中途被 GC。
|
||
_running_stream_tasks: set[asyncio.Task[None]] = set()
|
||
|
||
# 流式增量分片阈值：每攒够 N 字符产出一条 delta 事件，配合 sse_poll_interval（0.01s）
|
||
# 让前端看到逐段平滑增量。值越小流式越细、事件数越多（run_max_events 默认 1000）。
|
||
_REASONING_FLUSH_CHARS = 8
|
||
_CONTENT_FLUSH_CHARS = 8
|
||
|
||
_provider_manager: ProviderManager | None = None
|
||
|
||
|
||
def _get_provider_manager() -> ProviderManager:
|
||
global _provider_manager
|
||
if _provider_manager is None:
|
||
_provider_manager = ProviderManagerFactory.create()
|
||
return _provider_manager
|
||
|
||
|
||
async def drain_pending_streams() -> None:
|
||
"""等待所有进行中的后台流式任务完成。
|
||
|
||
``execute`` 在创建 run 后即返回，LLM 流式在后台任务中执行；调用方在需要确保
|
||
运行终态时（测试断言、优雅停机）await 此函数。
|
||
"""
|
||
if _running_stream_tasks:
|
||
await asyncio.gather(*_running_stream_tasks, return_exceptions=True)
|
||
|
||
|
||
def _contract_system_messages(contract: RuntimeContract) -> list[str]:
|
||
messages = [*contract.system_instructions]
|
||
if contract.system_prompt:
|
||
messages.append(contract.system_prompt) # 卡的 system_prompt：道体增补层（§6）
|
||
if contract.session_goal:
|
||
messages.append(format_goal_directive(contract.session_goal))
|
||
if contract.active_goals:
|
||
messages.append(contract.active_goals) # 在 session_goal 之后注入
|
||
if contract.mission:
|
||
messages.append(f"核心使命：\n{contract.mission}") # 道体恒定层
|
||
if contract.lorebook_before:
|
||
messages.append(contract.lorebook_before)
|
||
if contract.personality_summary:
|
||
messages.append(f"当前人格：\n{contract.personality_summary}")
|
||
if contract.lorebook_after:
|
||
messages.append(contract.lorebook_after)
|
||
if contract.emotion_state:
|
||
messages.append(contract.emotion_state)
|
||
if contract.memory_context:
|
||
messages.append(contract.memory_context)
|
||
if contract.temporal_context:
|
||
messages.append(contract.temporal_context)
|
||
if contract.safety_guidelines:
|
||
messages.append("行为边界：\n" + contract.safety_guidelines) # 道体安全（先于司命动态约束）
|
||
if contract.safety_constraints:
|
||
messages.append(
|
||
"安全约束：\n" + "\n".join(f"- {item}" for item in contract.safety_constraints)
|
||
)
|
||
# 沙箱工作目录提示：避免模型在 shell 中使用 Linux 特有路径（如 /home/user、/tmp）
|
||
# 导致 macOS 上沙箱执行时 cd 失败。
|
||
messages.append(
|
||
"当前运行在隔离沙箱中，所有文件操作（read_file/write_file/shell）的工作目录"
|
||
"是沙箱隔离的 workspace。系统并非 Linux，/home/user、/root 等路径不存在。"
|
||
"不要 cd 到这些目录；直接在 workspace 中对文件进行读写或执行命令。"
|
||
)
|
||
return messages
|
||
|
||
|
||
# 首轮对话自动重命名：用默认模型生成简短标题，失败静默不影响对话主流程。
|
||
_TITLE_MAX_CHARS = 40
|
||
_TITLE_STRIP_CHARS = "\"'“”‘’「」『』"
|
||
|
||
|
||
|
||
|
||
async def _call_model(system_prompt: str, model_hint: str) -> str:
|
||
"""调用 LLM 返回纯文本。"""
|
||
manager = _get_provider_manager()
|
||
messages = manager.to_message_list([system_prompt], [('user', system_prompt)])
|
||
response = await manager.chat(messages, model_hint=model_hint)
|
||
return response.content
|
||
|
||
async def _generate_session_title(user_content: str, assistant_content: str) -> str:
|
||
"""用默认模型根据首轮对话内容生成简短会话标题。"""
|
||
manager = _get_provider_manager()
|
||
messages = manager.to_message_list(
|
||
[
|
||
"你会根据一段对话生成简短的中文会话标题。"
|
||
"要求：不超过 16 个字；概括对话主题；直接输出标题文字；"
|
||
"不要加引号、书名号、句号，不要解释。"
|
||
],
|
||
[("user", user_content), ("assistant", assistant_content)],
|
||
)
|
||
response = await manager.chat(messages, model_hint=settings.default_model_hint)
|
||
raw = response.content.strip().split("\n", 1)[0].strip()
|
||
return raw.strip(_TITLE_STRIP_CHARS)[:_TITLE_MAX_CHARS]
|
||
|
||
|
||
class SendChatMessageUseCase:
|
||
"""完成消息持久化与运行态记录；LLM 流式回复在后台任务中边产出边推 SSE 事件。"""
|
||
|
||
def __init__(
|
||
self,
|
||
session: AsyncSession,
|
||
*,
|
||
owner_scope: str = settings.default_owner_scope,
|
||
tool_runtime: ToolRuntime | None = None,
|
||
confirmation_registry: ConfirmationRegistry | None = None,
|
||
confirmation_store: ConfirmationStore | None = None,
|
||
event_store: EventStore | None = None,
|
||
agent_run_registry: AgentRunRegistry | None = None,
|
||
python_skill_catalog: PythonSkillCatalog | None = None,
|
||
) -> None:
|
||
self._session = session
|
||
self._owner_scope = owner_scope
|
||
self._tool_runtime = tool_runtime
|
||
self._confirmation_registry = confirmation_registry
|
||
self._confirmation_store = confirmation_store
|
||
self._event_store = event_store
|
||
self._agent_run_registry = agent_run_registry
|
||
self._python_skill_catalog = python_skill_catalog
|
||
|
||
async def execute(self, session_id: str, payload: SendMessageRequest) -> SendMessageResponse:
|
||
validate_text_message(modality=payload.modality, content=payload.content)
|
||
|
||
character = await resolve_active_character(self._session, owner_scope=self._owner_scope)
|
||
chat_session = await chat_repo.get_session(
|
||
self._session, session_id, owner_scope=self._owner_scope
|
||
)
|
||
if chat_session is None or chat_session.status == "deleted":
|
||
raise AppError("not_found", "会话不存在。")
|
||
|
||
if payload.prior_versions:
|
||
# 重答：复用原文用户消息，不新建 user 消息，避免刷新后出现重复的消息对。
|
||
recent = await chat_repo.list_recent_messages(
|
||
self._session,
|
||
chat_session.id,
|
||
limit=10,
|
||
)
|
||
existing_user = None
|
||
for msg in reversed(recent):
|
||
if msg.role == "user" and msg.deleted_at is None:
|
||
existing_user = msg
|
||
break
|
||
if existing_user is None:
|
||
raise AppError("not_found", "要重答的原始用户消息不存在。")
|
||
user_message = existing_user
|
||
else:
|
||
user_message = ChatMessage(
|
||
character_id=character.id,
|
||
session_id=chat_session.id,
|
||
role="user",
|
||
content=payload.content,
|
||
modality="text",
|
||
metadata_json=dumps({}),
|
||
sensitivity="low",
|
||
)
|
||
self._session.add(user_message)
|
||
await self._session.flush()
|
||
|
||
model_hint = payload.model_hint or default_model_hint(character)
|
||
session_goal = extract_session_goal(payload.content)
|
||
run = await runtime_service.create_run(
|
||
self._session,
|
||
character_id=character.id,
|
||
session_id=chat_session.id,
|
||
model_hint=model_hint,
|
||
goal=session_goal or "回复用户消息",
|
||
)
|
||
user_message.run_id = run.id
|
||
|
||
await runtime_service.append_event(
|
||
self._session,
|
||
run_id=run.id,
|
||
character_id=character.id,
|
||
event_type="run.started",
|
||
payload={"run_id": run.id, "session_id": chat_session.id},
|
||
)
|
||
# 先提交 run + run.started：HTTP 响应随后返回 run_id，前端据此建立 SSE；
|
||
# 后台流式在独立 session 中边产出边提交，前端 SSE（DB 轮询）即可实时读到增量。
|
||
await self._session.commit()
|
||
|
||
task = asyncio.create_task(
|
||
self._stream_assistant_response(
|
||
run_id=run.id,
|
||
character_id=character.id,
|
||
chat_session_id=chat_session.id,
|
||
payload=payload,
|
||
)
|
||
)
|
||
_running_stream_tasks.add(task)
|
||
task.add_done_callback(_running_stream_tasks.discard)
|
||
|
||
from app.core.config import settings as _settings
|
||
|
||
return SendMessageResponse(
|
||
message_id=user_message.id,
|
||
assistant_message_id=None,
|
||
run_id=run.id,
|
||
session_id=chat_session.id,
|
||
character_id=character.id,
|
||
status=run.status,
|
||
stream_url=f"{_settings.api_prefix}/runs/{run.id}/stream",
|
||
)
|
||
|
||
async def _stream_assistant_response(
|
||
self,
|
||
*,
|
||
run_id: str,
|
||
character_id: str,
|
||
chat_session_id: str,
|
||
payload: SendMessageRequest,
|
||
) -> None:
|
||
"""后台流式：独立 session 完成 LLM 调用与事件推送，异常定稿为 run.failed。"""
|
||
|
||
try:
|
||
async with AsyncSessionLocal() as session:
|
||
await self._produce_assistant_stream(
|
||
session,
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
chat_session_id=chat_session_id,
|
||
payload=payload,
|
||
)
|
||
except TimeoutError:
|
||
await self._fail_stream(
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
code="run_timeout",
|
||
message="运行超时。",
|
||
)
|
||
except Exception:
|
||
logger.exception("chat_stream_failed", run_id=run_id)
|
||
await self._fail_stream(
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
code="run_failed",
|
||
message="运行失败，请稍后重试。",
|
||
)
|
||
|
||
async def _produce_assistant_stream(
|
||
self,
|
||
session: AsyncSession,
|
||
*,
|
||
run_id: str,
|
||
character_id: str,
|
||
chat_session_id: str,
|
||
payload: SendMessageRequest,
|
||
) -> None:
|
||
character = await resolve_active_character(session, owner_scope=self._owner_scope)
|
||
model_hint = payload.model_hint or default_model_hint(character)
|
||
|
||
run = await runtime_repo.get_run(session, run_id)
|
||
if run is None or run.status in TERMINAL_RUN_STATUS:
|
||
return
|
||
session_goal = run.goal or ""
|
||
|
||
# 重答场景（prior_versions）：在构建上下文前软删除被替换的旧 assistant 消息，
|
||
# 保留原文用户消息不动——用户消息已在 execute() 中复用，不新建、不删除。
|
||
if payload.prior_versions:
|
||
latest = await chat_repo.list_recent_messages(
|
||
session,
|
||
chat_session_id,
|
||
limit=5,
|
||
)
|
||
# latest 按 created_at ASC 排列且仅含未删除消息，从末尾往前找最后一个 assistant
|
||
for i in range(len(latest) - 1, -1, -1):
|
||
if latest[i].role == "assistant" and latest[i].deleted_at is None:
|
||
latest[i].deleted_at = utc_now()
|
||
break
|
||
await session.commit()
|
||
|
||
safety = await decide_chat_safety(
|
||
session,
|
||
character_id=character_id,
|
||
run_id=run_id,
|
||
)
|
||
personality = await build_personality_summary(session, character)
|
||
memories = await retrieve_memories(
|
||
session,
|
||
character_id,
|
||
limit=settings.memory_injection_limit,
|
||
)
|
||
memory_context = build_memory_context(memories)
|
||
|
||
# 息壤推送：将检索到的记忆以结构化数据推送到前端 SSE 事件流
|
||
if memories.memories:
|
||
memory_items = []
|
||
for m in memories.memories:
|
||
memory_items.append(
|
||
{
|
||
"id": m.id,
|
||
"title": m.title,
|
||
"content": m.content,
|
||
"source": m.source,
|
||
"confidence": m.confidence,
|
||
"importance": m.importance,
|
||
"type": m.type,
|
||
}
|
||
)
|
||
await runtime_service.append_event(
|
||
session,
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
event_type="memory.retrieved",
|
||
payload={"memories": memory_items},
|
||
)
|
||
await session.commit()
|
||
|
||
recent_messages = await recent_chat_messages(session, chat_session_id)
|
||
trace_id = current_trace_id()
|
||
|
||
# 辰正 temporal_context（设计 §8.1）：对话入口算时态快照 + 发 time.* 事件到当前 run。
|
||
# 降级原则（§12）：任何失败 → temporal_context=None，不注入，不阻塞对话。
|
||
temporal_context: str | None = None
|
||
if self._event_store is not None:
|
||
try:
|
||
from app.modules.time_sense.modulation import ModulationInput
|
||
from app.services.time_sense import compute_live_snapshot
|
||
|
||
time_result = await compute_live_snapshot(
|
||
session,
|
||
self._event_store,
|
||
character_id,
|
||
ModulationInput(),
|
||
run_id=run_id,
|
||
owner_scope=self._owner_scope,
|
||
)
|
||
temporal_context = time_result.context_text or None
|
||
except Exception: # noqa: BLE001
|
||
logger.warning("temporal_context 计算失败，已跳过", run_id=run_id)
|
||
|
||
# 灵核情绪状态注入（设计 §11.5）：拉模式 compute_emotion_prompt。
|
||
# 降级原则（§12）：任何失败 → emotion_state=None，不注入，不阻塞对话。
|
||
emotion_state: str | None = None
|
||
try:
|
||
from app.services.personality_emotion import compute_emotion_prompt
|
||
|
||
emotion_state = await compute_emotion_prompt(session, character_id)
|
||
except Exception: # noqa: BLE001
|
||
logger.warning("emotion_state 计算失败，已跳过", run_id=run_id)
|
||
|
||
# 世谱（Lorebook）注入：扫对话历史匹配命中条目，注入 before/after（设计 §7）
|
||
# 降级原则：任何失败 → 不注入，不阻塞对话。
|
||
lorebook_before: str | None = None
|
||
lorebook_after: str | None = None
|
||
try:
|
||
from app.services.lorebook import build_lorebook_injection
|
||
|
||
scan_parts = [payload.content]
|
||
scan_parts.extend(
|
||
m.content for m in recent_messages[-8:] if m.role in ("user", "assistant")
|
||
)
|
||
lorebook_before, lorebook_after = await build_lorebook_injection(
|
||
session,
|
||
character_id,
|
||
self._owner_scope,
|
||
"\n".join(scan_parts),
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
logger.warning("世谱注入失败，已跳过", run_id=run_id)
|
||
|
||
profile_data = loads_dict(character.profile_json or "{}")
|
||
contract = RuntimeContract(
|
||
character_id=character_id,
|
||
session_id=chat_session_id,
|
||
run_id=run_id,
|
||
model_hint=model_hint,
|
||
agent=DEFAULT_CHAT_AGENT,
|
||
system_instructions=[],
|
||
system_prompt=profile_data.get("system_prompt") or None,
|
||
session_goal=session_goal or None,
|
||
personality_summary=personality.text,
|
||
mission=character.mission,
|
||
safety_guidelines=character.safety_guidelines,
|
||
lorebook_before=lorebook_before,
|
||
lorebook_after=lorebook_after,
|
||
emotion_state=emotion_state,
|
||
memory_context=memory_context,
|
||
temporal_context=temporal_context,
|
||
recent_messages=recent_messages,
|
||
safety_constraints=safety.constraints,
|
||
trace_id=trace_id,
|
||
)
|
||
# 道体完整性校验（§6）：确认恒定层未被动态注入在数据层覆盖
|
||
from app.modules.runtime.integrity import validate_daoti_integrity
|
||
|
||
validate_daoti_integrity(
|
||
contract,
|
||
mission=character.mission,
|
||
safety_guidelines=character.safety_guidelines,
|
||
)
|
||
|
||
# —— 活跃目标注入 ——
|
||
try:
|
||
from app.services.governance import load_active_goals_context
|
||
|
||
active_goals = await load_active_goals_context(character.id)
|
||
if active_goals:
|
||
contract.active_goals = active_goals
|
||
except Exception:
|
||
logger.warning("活跃目标加载失败，跳过注入")
|
||
|
||
await record_audit_event(
|
||
session,
|
||
action="model.request.started",
|
||
character_id=character_id,
|
||
run_id=run_id,
|
||
actor="system",
|
||
decision="auto",
|
||
input_summary=model_hint,
|
||
resource_type="run",
|
||
resource_id=run_id,
|
||
)
|
||
system_messages = compose_system_messages(
|
||
_contract_system_messages(contract),
|
||
contract.agent,
|
||
[],
|
||
)
|
||
|
||
# ── 百工技能执行引擎：匹配 → prompt 注入 → 工具绑定注册 ──
|
||
from app.application.skill_executor import SkillExecutor
|
||
|
||
tool_view = RunScopedToolView(self._tool_runtime) if self._tool_runtime is not None else None
|
||
skill_exec = SkillExecutor(
|
||
tool_view=tool_view,
|
||
event_store=self._event_store,
|
||
)
|
||
try:
|
||
await skill_exec.match_and_activate(
|
||
payload.content,
|
||
system_messages,
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
skill_refs=contract.agent.skill_refs if contract.agent else (),
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
logger.warning("skill_executor 匹配失败，已跳过")
|
||
|
||
agent_task = None
|
||
completed_plan_step_ids: set[str] = set()
|
||
|
||
# ── 洛书 Deliberative 规划 ──
|
||
if settings.enable_deliberative_planning:
|
||
from app.application.planning import (
|
||
build_planning_prompt,
|
||
format_plan_for_prompt,
|
||
parse_plan,
|
||
should_plan,
|
||
)
|
||
|
||
if should_plan(payload.content):
|
||
try:
|
||
planner_prompt = build_planning_prompt(
|
||
payload.content,
|
||
memory_context=memory_context[:300] if memory_context else '',
|
||
)
|
||
planner_raw = await _call_model(planner_prompt, model_hint)
|
||
plan = parse_plan(planner_raw)
|
||
if plan:
|
||
agent_task = task_from_deliberative_plan(
|
||
plan,
|
||
run_id=run_id,
|
||
agent_name=contract.agent.name if contract.agent else "默认 Agent",
|
||
objective=payload.content,
|
||
)
|
||
await runtime_service.append_event(
|
||
session,
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
event_type="plan.created",
|
||
source="planner",
|
||
payload=task_plan_payload(agent_task),
|
||
)
|
||
await session.commit()
|
||
plan_text = format_plan_for_prompt(plan)
|
||
system_messages.append(
|
||
"【当前计划】\n请按以下计划逐步执行。"
|
||
"每一步完成后标记完成，再进入下一步。"
|
||
"全部完成后回复总结。\n\n" + plan_text
|
||
)
|
||
logger.info(f"洛书生成 deliberative plan: {plan.title} ({len(plan.steps)} 步)")
|
||
except Exception:
|
||
logger.warning('deliberative 规划生成失败，回退 ReAct')
|
||
|
||
python_skills: tuple[PythonSkillDefinition, ...] = ()
|
||
if self._python_skill_catalog is not None:
|
||
python_skills = self._python_skill_catalog.match(
|
||
payload.content,
|
||
skill_refs=contract.agent.skill_refs if contract.agent else (),
|
||
metadata={
|
||
"run_id": run_id,
|
||
"session_id": chat_session_id,
|
||
"character_id": character_id,
|
||
"model_hint": model_hint,
|
||
},
|
||
)
|
||
for skill in python_skills:
|
||
if skill.instructions:
|
||
system_messages.append(f"【Python Skill：{skill.name}】\n{skill.instructions}")
|
||
|
||
manager = _get_provider_manager()
|
||
messages = manager.to_message_list(
|
||
system_messages,
|
||
history_pairs(recent_messages),
|
||
)
|
||
|
||
reasoning_buf = ""
|
||
content_buf = ""
|
||
full_content = ""
|
||
had_reasoning = False
|
||
reasoning_finalized = False
|
||
last_model = model_hint
|
||
|
||
async def flush_reasoning(*, force: bool = False) -> None:
|
||
nonlocal reasoning_buf
|
||
while reasoning_buf and (len(reasoning_buf) >= _REASONING_FLUSH_CHARS or force):
|
||
cut = (
|
||
_REASONING_FLUSH_CHARS
|
||
if len(reasoning_buf) >= _REASONING_FLUSH_CHARS
|
||
else len(reasoning_buf)
|
||
)
|
||
delta, reasoning_buf = reasoning_buf[:cut], reasoning_buf[cut:]
|
||
await runtime_service.append_event(
|
||
session,
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
event_type="reasoning.delta",
|
||
source="model",
|
||
payload={"delta": delta},
|
||
)
|
||
await session.commit()
|
||
|
||
async def flush_content(*, force: bool = False) -> None:
|
||
nonlocal content_buf
|
||
while content_buf and (len(content_buf) >= _CONTENT_FLUSH_CHARS or force):
|
||
cut = (
|
||
_CONTENT_FLUSH_CHARS
|
||
if len(content_buf) >= _CONTENT_FLUSH_CHARS
|
||
else len(content_buf)
|
||
)
|
||
delta, content_buf = content_buf[:cut], content_buf[cut:]
|
||
await runtime_service.append_event(
|
||
session,
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
event_type="message.delta",
|
||
source="model",
|
||
payload={"delta": delta},
|
||
)
|
||
await session.commit()
|
||
|
||
async def finalize_reasoning() -> None:
|
||
"""收尾思考阶段：冲刷残余 reasoning 并推送 reasoning.completed。
|
||
|
||
在第一段正文出现前调用，确保思考摘要完整先于正文；流结束兜底再调一次。
|
||
多轮 Agent 场景下，每轮模型输出的 content 出现前都会触发，已被 reasoning_finalized
|
||
标记短路——即仅第一段 reasoning 完整 flush 后发 completed，后续轮次的 reasoning
|
||
仍经 on_chunk 实时 flush，但不再发 completed（避免事件流断裂）。
|
||
|
||
思考模式关闭时（payload.thinking=False）不处理任何 reasoning 事件，
|
||
防止模型仍产出 reasoning_content 导致前端 ThoughtBlock 误展示。
|
||
"""
|
||
nonlocal reasoning_finalized
|
||
if reasoning_finalized or not payload.thinking:
|
||
return
|
||
reasoning_finalized = True
|
||
await flush_reasoning(force=True)
|
||
if had_reasoning:
|
||
await runtime_service.append_event(
|
||
session,
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
event_type="reasoning.completed",
|
||
source="model",
|
||
payload={},
|
||
)
|
||
await session.commit()
|
||
|
||
async def on_chunk(chunk: ChatChunk) -> None:
|
||
"""把 Agent 流式 chunk 喂给现有 buf/flush/finalize 逻辑，事件行为与单轮流式等价。"""
|
||
nonlocal reasoning_buf, content_buf, full_content, had_reasoning
|
||
# 思考模式关闭时不处理 reasoning chunk，避免前端 ThoughtBlock 错误展示
|
||
if payload.thinking and chunk.delta_reasoning:
|
||
had_reasoning = True
|
||
reasoning_buf += chunk.delta_reasoning
|
||
await flush_reasoning()
|
||
if chunk.delta_content:
|
||
# 正文开始前先收尾思考，保证思考摘要完整先于正文
|
||
await finalize_reasoning()
|
||
content_buf += chunk.delta_content
|
||
full_content += chunk.delta_content
|
||
await flush_content()
|
||
|
||
# 收集工具调用信息，最终存入消息 metadata 供前端持久化展示
|
||
tool_calls: list[dict[str, object]] = []
|
||
_started_tool_calls: dict[str, dict[str, object]] = {}
|
||
_plan_steps_by_tool_call = {}
|
||
|
||
async def _reset_turn_state() -> None:
|
||
"""每轮 Agent 开始前重置 reasoning 标记（多轮场景支持）。
|
||
|
||
Agent loop 首轮产出 reasoning → content → tool_calls，次轮可能再产出 reasoning。
|
||
若不重置，reasoning_finalized=True 导致 reasoning.completed 无法再发出。
|
||
"""
|
||
nonlocal had_reasoning, reasoning_finalized
|
||
had_reasoning = False
|
||
reasoning_finalized = False
|
||
|
||
async def on_tool_event(event_type: str, payload_dict: dict[str, object]) -> None:
|
||
"""工具生命周期事件：作为 agent 事件落库 + 收集 tool_calls 入 metadata。"""
|
||
await runtime_service.append_event(
|
||
session,
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
event_type=event_type,
|
||
source="agent",
|
||
payload=payload_dict,
|
||
)
|
||
await session.commit()
|
||
|
||
# 收集工具调用信息供前端持久化展示
|
||
if event_type == "tool.started":
|
||
tc_id = str(payload_dict.get("tool_call_id", ""))
|
||
tool_name = str(payload_dict.get("tool_name", ""))
|
||
if agent_task is not None:
|
||
step = next_step_for_tool(agent_task, tool_name, completed_plan_step_ids)
|
||
if step is not None:
|
||
_plan_steps_by_tool_call[tc_id] = step
|
||
await runtime_service.append_event(
|
||
session,
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
event_type="plan.step.started",
|
||
source="planner",
|
||
payload=step_event_payload(agent_task, step, status="running"),
|
||
)
|
||
await session.commit()
|
||
_started_tool_calls[tc_id] = {
|
||
"tool_name": str(payload_dict.get("tool_name", "")),
|
||
"tool_call_id": tc_id,
|
||
"args": str(payload_dict.get("args", "")),
|
||
"status": "running",
|
||
}
|
||
elif event_type in ("tool.completed", "tool.failed"):
|
||
tc_id = str(payload_dict.get("tool_call_id", ""))
|
||
entry = _started_tool_calls.pop(tc_id, None)
|
||
step = _plan_steps_by_tool_call.pop(tc_id, None)
|
||
if step is not None and agent_task is not None:
|
||
tool_status = str(payload_dict.get("status", "error"))
|
||
step_status = "completed" if tool_status == "success" else "failed"
|
||
if step_status == "completed":
|
||
completed_plan_step_ids.add(step.id)
|
||
await runtime_service.append_event(
|
||
session,
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
event_type=f"plan.step.{step_status}",
|
||
source="planner",
|
||
payload=step_event_payload(
|
||
agent_task,
|
||
step,
|
||
status=step_status,
|
||
result_summary=str(payload_dict.get("result", ""))[:500] or None,
|
||
),
|
||
)
|
||
await session.commit()
|
||
if entry is not None:
|
||
entry["status"] = str(payload_dict.get("status", "error"))
|
||
entry["result"] = str(payload_dict.get("result", ""))
|
||
tool_calls.append(entry)
|
||
|
||
extensions = ExtensionRegistry()
|
||
python_skill_activation = await activate_python_skills(extensions, python_skills)
|
||
if python_skills:
|
||
await runtime_service.append_event(
|
||
session,
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
event_type="skill.activated",
|
||
source="agent",
|
||
payload={"skills": [skill.name for skill in python_skills], "kind": "python"},
|
||
)
|
||
await session.commit()
|
||
|
||
async def persist_agent_event(event: AgentEvent, _run_context: AgentRunContext) -> None:
|
||
await runtime_service.append_event(
|
||
session,
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
event_type=event.type,
|
||
source="agent",
|
||
payload=event.payload,
|
||
)
|
||
await session.commit()
|
||
|
||
extensions.add_event_listener(persist_agent_event)
|
||
on_confirm_timeout = None
|
||
if self._confirmation_store is not None:
|
||
confirmation_store = self._confirmation_store
|
||
|
||
async def on_confirm_timeout(confirmation_id: str) -> None:
|
||
await confirmation_store.mark_timed_out(confirmation_id)
|
||
|
||
agent = Agent(
|
||
AgentConfig(
|
||
tools=tuple(
|
||
default_tools(
|
||
tool_view.specs_by_name(),
|
||
allowed_tools=contract.agent.tool_policy.allowed_tools
|
||
if contract.agent
|
||
else (),
|
||
)
|
||
if tool_view is not None
|
||
else ()
|
||
),
|
||
tool_runtime=tool_view,
|
||
model_hint=(contract.agent.model_hint or model_hint)
|
||
if contract.agent
|
||
else model_hint,
|
||
thinking_enabled=payload.thinking,
|
||
reasoning_effort=payload.reasoning_effort,
|
||
max_iterations=contract.agent.execution_policy.max_iterations
|
||
if contract.agent
|
||
else settings.agent_max_iterations,
|
||
confirmation_registry=self._confirmation_registry,
|
||
confirmation_timeout=settings.confirmation_timeout_seconds,
|
||
on_turn=_reset_turn_state,
|
||
on_chunk=on_chunk,
|
||
on_tool_event=on_tool_event,
|
||
on_confirm_timeout=on_confirm_timeout,
|
||
),
|
||
messages=messages,
|
||
run_context=AgentRunContext(
|
||
run_id=run_id,
|
||
session_id=chat_session_id,
|
||
character_id=character_id,
|
||
trace_id=trace_id,
|
||
),
|
||
extensions=extensions,
|
||
# 保留模块级符号作为注入点，现有测试与外部集成无需重写。
|
||
runner=run_agent_turn,
|
||
)
|
||
if self._agent_run_registry is not None:
|
||
await self._agent_run_registry.register(run_id, agent)
|
||
try:
|
||
run = await runtime_repo.get_run(session, run_id)
|
||
if run is None or run.status in TERMINAL_RUN_STATUS:
|
||
return
|
||
async with asyncio.timeout(settings.run_timeout_seconds):
|
||
agent_result = await agent.continue_run()
|
||
finally:
|
||
if self._agent_run_registry is not None:
|
||
await self._agent_run_registry.unregister(run_id, agent)
|
||
await python_skill_activation.close()
|
||
|
||
# 流结束兜底：模型可能只输出思考无正文，或残余 buffer 未达阈值。
|
||
await finalize_reasoning()
|
||
await flush_content(force=True)
|
||
|
||
# 取消可能发生在模型返回与持久化之间；终态 Run 不再写助手消息或覆盖状态。
|
||
run = await runtime_repo.get_run(session, run_id)
|
||
if run is None or run.status in TERMINAL_RUN_STATUS:
|
||
return
|
||
|
||
# 以 agent_result.content 作为权威最终回复正文持久化（应与 full_content 一致；
|
||
# 不一致时以 agent 循环产出为准，因后者覆盖多轮模型最终输出）。
|
||
final_content = agent_result.content or full_content
|
||
|
||
version_ts = utc_now().isoformat()
|
||
# 构造版本历史：prior_versions 前置 + 当前回复作为最新版本
|
||
prior = payload.prior_versions or []
|
||
all_versions = [*prior, {"content": final_content, "created_at": version_ts}]
|
||
active_idx = len(all_versions) - 1 # 始终最新
|
||
|
||
assistant = ChatMessage(
|
||
character_id=character_id,
|
||
session_id=chat_session_id,
|
||
role="assistant",
|
||
content=final_content,
|
||
modality="text",
|
||
run_id=run_id,
|
||
metadata_json=dumps(
|
||
{
|
||
"model_hint": model_hint,
|
||
"model": last_model,
|
||
"tool_calls": tool_calls,
|
||
}
|
||
),
|
||
sensitivity="low",
|
||
versions_json=dumps(all_versions),
|
||
active_version_idx=active_idx,
|
||
)
|
||
session.add(assistant)
|
||
await session.flush()
|
||
await record_audit_event(
|
||
session,
|
||
action="model.response.completed",
|
||
character_id=character_id,
|
||
run_id=run_id,
|
||
actor="model",
|
||
decision="auto",
|
||
output_summary="模型响应已完成。",
|
||
resource_type="message",
|
||
resource_id=assistant.id,
|
||
)
|
||
await runtime_service.append_event(
|
||
session,
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
event_type="message.completed",
|
||
source="model",
|
||
payload={"message_id": assistant.id, "content": final_content},
|
||
)
|
||
run = await runtime_repo.get_run(session, run_id)
|
||
if run is not None and run.status not in TERMINAL_RUN_STATUS:
|
||
await runtime_service.complete_run(session, run)
|
||
await runtime_service.append_event(
|
||
session,
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
event_type="run.completed",
|
||
payload={"run_id": run_id},
|
||
)
|
||
await session.commit()
|
||
|
||
# 首轮对话完成后，用默认模型自动生成会话标题（失败不影响主流程）
|
||
await self._maybe_rename_session(
|
||
session,
|
||
chat_session_id=chat_session_id,
|
||
user_content=payload.content,
|
||
assistant_content=final_content,
|
||
)
|
||
|
||
# 每轮对话后检查大周天触发条件（失败不影响主流程）
|
||
try:
|
||
from app.application.big_cycle import maybe_trigger_big_cycle
|
||
|
||
await maybe_trigger_big_cycle(character_id, trigger="session_turn")
|
||
except Exception: # noqa: BLE001
|
||
logger.warning("大周天触发检查失败，已跳过")
|
||
|
||
# 小周天每轮记忆候选 + 返照反思（后台任务，失败不影响主流程）
|
||
try:
|
||
from app.application.reflection import start_reflection_after_turn
|
||
|
||
start_reflection_after_turn(
|
||
character_id=character_id,
|
||
session_id=chat_session_id,
|
||
user_content=payload.content,
|
||
assistant_content=final_content,
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
logger.warning("小周天反思触发失败，已跳过")
|
||
|
||
# —— 会话目标自动持久化 ——
|
||
try:
|
||
from app.db.session import AsyncSessionLocal as GoalSession
|
||
from app.services.goal_persist import auto_persist_session_goal
|
||
|
||
async with GoalSession() as goal_session:
|
||
await auto_persist_session_goal(goal_session, character.id, payload.content)
|
||
await goal_session.commit()
|
||
except Exception:
|
||
logger.warning("会话目标自动持久化失败，跳过")
|
||
|
||
async def _maybe_rename_session(
|
||
self,
|
||
session: AsyncSession,
|
||
*,
|
||
chat_session_id: str,
|
||
user_content: str,
|
||
assistant_content: str,
|
||
) -> None:
|
||
"""首轮对话完成后用 LLM 生成会话标题；已有标题或非首轮则跳过，失败静默。"""
|
||
try:
|
||
existing = await chat_repo.get_session(session, chat_session_id)
|
||
if existing is None or existing.title:
|
||
return
|
||
assistant_count = await chat_repo.count_messages_by_role(
|
||
session, chat_session_id, role="assistant"
|
||
)
|
||
if assistant_count != 1:
|
||
return
|
||
title = await _generate_session_title(user_content, assistant_content)
|
||
if not title:
|
||
return
|
||
await chat_repo.update_session_title(session, chat_session_id, title)
|
||
await session.commit()
|
||
logger.info("session_auto_renamed", session_id=chat_session_id, title=title)
|
||
except Exception:
|
||
logger.warning("session_title_generate_failed", session_id=chat_session_id)
|
||
|
||
async def _fail_stream(
|
||
self,
|
||
*,
|
||
run_id: str,
|
||
character_id: str,
|
||
code: str,
|
||
message: str,
|
||
) -> None:
|
||
async with AsyncSessionLocal() as session:
|
||
run = await runtime_repo.get_run(session, run_id)
|
||
if run is None or run.status in TERMINAL_RUN_STATUS:
|
||
return
|
||
await runtime_service.fail_run(session, run, code=code, message=message)
|
||
await runtime_service.append_event(
|
||
session,
|
||
run_id=run_id,
|
||
character_id=character_id,
|
||
event_type="run.failed",
|
||
payload={"code": code, "message": message},
|
||
)
|
||
await session.commit()
|