chore: add interaction tracing and remove personal fields from logs

This commit is contained in:
2026-03-13 10:23:47 +01:00
parent 9c97702daa
commit 9bd629cb59
3 changed files with 72 additions and 12 deletions

View File

@@ -219,6 +219,13 @@ async def _handle_home_request(
request_id = frame.get("request_id") or str(uuid4()) request_id = frame.get("request_id") or str(uuid4())
message: str = frame.get("message", "") message: str = frame.get("message", "")
session_id: str = frame.get("session_id") or str(uuid4()) session_id: str = frame.get("session_id") or str(uuid4())
logger.info(
"device_ws: home_request_start user=%s req=%s session=%s msg=%s",
user_id,
request_id,
session_id,
message[:200],
)
# ── Memory: enrich context before LLM call ──────────────────────── # ── Memory: enrich context before LLM call ────────────────────────
async with async_session() as db: async with async_session() as db:
@@ -256,6 +263,13 @@ async def _handle_home_request(
await memory.store_episode( await memory.store_episode(
user_id, session_id, message, "".join(response_chunks), trace_id=request_id user_id, session_id, message, "".join(response_chunks), trace_id=request_id
) )
logger.info(
"device_ws: home_request_end user=%s req=%s session=%s response_chars=%d",
user_id,
request_id,
session_id,
len("".join(response_chunks)),
)
async def _handle_floating_request( async def _handle_floating_request(
@@ -268,6 +282,14 @@ async def _handle_floating_request(
message: str = frame.get("message", "") message: str = frame.get("message", "")
session_id: str = frame.get("session_id") or str(uuid4()) session_id: str = frame.get("session_id") or str(uuid4())
scope: dict = frame.get("scope", {}) scope: dict = frame.get("scope", {})
logger.info(
"device_ws: floating_request_start user=%s req=%s session=%s scope=%s msg=%s",
user_id,
request_id,
session_id,
json.dumps(scope, ensure_ascii=True)[:200],
message[:200],
)
# ── Memory: enrich context before LLM call ──────────────────────── # ── Memory: enrich context before LLM call ────────────────────────
async with async_session() as db: async with async_session() as db:
@@ -304,6 +326,13 @@ async def _handle_floating_request(
await memory.store_episode( await memory.store_episode(
user_id, session_id, message, "".join(response_chunks), trace_id=request_id user_id, session_id, message, "".join(response_chunks), trace_id=request_id
) )
logger.info(
"device_ws: floating_request_end user=%s req=%s session=%s response_chars=%d",
user_id,
request_id,
session_id,
len("".join(response_chunks)),
)
# ── Heartbeat ───────────────────────────────────────────────────────── # ── Heartbeat ─────────────────────────────────────────────────────────

View File

@@ -320,6 +320,7 @@ async def _run_single_agent(
), ),
] ]
tool_calls_count = 0
collected: list[dict[str, Any]] = [] collected: list[dict[str, Any]] = []
set_tool_result_collector(collected) set_tool_result_collector(collected)
try: try:
@@ -328,10 +329,19 @@ async def _run_single_agent(
messages.append(response) messages.append(response)
if not response.tool_calls: if not response.tool_calls:
return _as_text(response.content) final_text = _as_text(response.content)
logger.info(
"deep_agent: run_single_agent_end trace=%s user=%s tool_calls=%d response_chars=%d",
trace_id or "-",
user_id,
tool_calls_count,
len(final_text),
)
return final_text
tool_map = {tool_def.name: tool_def for tool_def in tools} tool_map = {tool_def.name: tool_def for tool_def in tools}
for call in response.tool_calls: for call in response.tool_calls:
tool_calls_count += 1
call_id = str(call.get("id", "")) call_id = str(call.get("id", ""))
call_name = str(call.get("name", "")) call_name = str(call.get("name", ""))
call_args = call.get("args", {}) call_args = call.get("args", {})
@@ -358,7 +368,15 @@ async def _run_single_agent(
messages.append(ToolMessage(content=str(tool_output), tool_call_id=call["id"])) messages.append(ToolMessage(content=str(tool_output), tool_call_id=call["id"]))
final = await llm.ainvoke(messages) final = await llm.ainvoke(messages)
return _as_text(final.content) final_text = _as_text(final.content)
logger.info(
"deep_agent: run_single_agent_end trace=%s user=%s tool_calls=%d response_chars=%d fallback=1",
trace_id or "-",
user_id,
tool_calls_count,
len(final_text),
)
return final_text
finally: finally:
clear_tool_result_collector() clear_tool_result_collector()
@@ -387,6 +405,8 @@ async def _run_single_agent_stream(
), ),
] ]
tool_calls_count = 0
streamed_chars = 0
collected: list[dict[str, Any]] = [] collected: list[dict[str, Any]] = []
set_tool_result_collector(collected) set_tool_result_collector(collected)
try: try:
@@ -398,11 +418,20 @@ async def _run_single_agent_stream(
async for chunk in llm.astream(messages): async for chunk in llm.astream(messages):
token = _as_text(getattr(chunk, "content", "")) token = _as_text(getattr(chunk, "content", ""))
if token: if token:
streamed_chars += len(token)
yield "token", token yield "token", token
logger.info(
"deep_agent: run_single_agent_stream_end trace=%s user=%s tool_calls=%d response_chars=%d",
trace_id or "-",
user_id,
tool_calls_count,
streamed_chars,
)
return return
tool_map = {tool_def.name: tool_def for tool_def in tools} tool_map = {tool_def.name: tool_def for tool_def in tools}
for call in response.tool_calls: for call in response.tool_calls:
tool_calls_count += 1
call_id = str(call.get("id", "")) call_id = str(call.get("id", ""))
call_name = str(call.get("name", "")) call_name = str(call.get("name", ""))
call_args = call.get("args", {}) call_args = call.get("args", {})
@@ -431,7 +460,15 @@ async def _run_single_agent_stream(
async for chunk in llm.astream(messages): async for chunk in llm.astream(messages):
token = _as_text(getattr(chunk, "content", "")) token = _as_text(getattr(chunk, "content", ""))
if token: if token:
streamed_chars += len(token)
yield "token", token yield "token", token
logger.info(
"deep_agent: run_single_agent_stream_end trace=%s user=%s tool_calls=%d response_chars=%d fallback=1",
trace_id or "-",
user_id,
tool_calls_count,
streamed_chars,
)
finally: finally:
clear_tool_result_collector() clear_tool_result_collector()

View File

@@ -70,10 +70,9 @@ class MemoryMiddleware:
user_dbg = await self._get_user_debug(user_id) user_dbg = await self._get_user_debug(user_id)
logger.info( logger.info(
"memory: enrich_context trace=%s user=%s email=%s tier=%s core=%d associative=%d episodic=%d proactive=%d", "memory: enrich_context trace=%s user=%s tier=%s core=%d associative=%d episodic=%d proactive=%d",
trace_id or "-", trace_id or "-",
user_id, user_id,
user_dbg.get("email") or "-",
user_dbg.get("tier") or "-", user_dbg.get("tier") or "-",
len(core), len(core),
len(associative), len(associative),
@@ -119,10 +118,9 @@ class MemoryMiddleware:
await self._db.commit() await self._db.commit()
user_dbg = await self._get_user_debug(user_id) user_dbg = await self._get_user_debug(user_id)
logger.info( logger.info(
"memory: store_episode trace=%s user=%s email=%s tier=%s session=%s", "memory: store_episode trace=%s user=%s tier=%s session=%s",
trace_id or "-", trace_id or "-",
user_id, user_id,
user_dbg.get("email") or "-",
user_dbg.get("tier") or "-", user_dbg.get("tier") or "-",
session_id, session_id,
) )
@@ -158,10 +156,9 @@ class MemoryMiddleware:
await self._db.commit() await self._db.commit()
user_dbg = await self._get_user_debug(user_id) user_dbg = await self._get_user_debug(user_id)
logger.info( logger.info(
"memory: update_core trace=%s user=%s email=%s tier=%s key=%s", "memory: update_core trace=%s user=%s tier=%s key=%s",
trace_id or "-", trace_id or "-",
user_id, user_id,
user_dbg.get("email") or "-",
user_dbg.get("tier") or "-", user_dbg.get("tier") or "-",
key, key,
) )
@@ -343,12 +340,9 @@ class MemoryMiddleware:
result = await self._db.execute(select(User).where(User.id == user_id)) result = await self._db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none() user = result.scalar_one_or_none()
if user is None: if user is None:
return {"email": None, "tier": None, "name": None, "surname": None} return {"tier": None}
return { return {
"email": user.email,
"tier": user.tier, "tier": user.tier,
"name": user.name,
"surname": user.surname,
} }
async def _load_core(self, user_id: str, fernet: Fernet) -> dict[str, str]: async def _load_core(self, user_id: str, fernet: Fernet) -> dict[str, str]: