xiaohei/socaichat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add call_id (run_id) to tool events, add on_chain_start debug logging

Browse Source 
- tool_start/tool_end/tool_error SSE payloads now include call_id field
  derived from LangGraph run_id for reliable tool event correlation
- tool_start_ts dict keyed by call_id instead of tool_name to handle
  concurrent calls to the same tool
- Added on_chain_start debug logging to observe chain names and metadata
  (no SSE emission yet, observation only)

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
gongzhiyong
2026-04-09 17:01:27 +08:00
co-authored by Claude Sonnet 4.6
parent 056cf6be9d
commit f03efbfe0c
1 changed files with 18 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/chat.py
+18 -3
View File
@@ -99,13 +99,14 @@ async def _stream_response(request: ChatRequest) -> AsyncIterator[bytes]:
yield _sse({"type": "token", "content": chunk.content})
elif kind == "on_tool_start":
call_id = event.get("run_id", str(uuid.uuid4()))
tool_name = event.get("name", "unknown")
tool_input = event.get("data", {}).get("input", {})
ts = int(time.time() * 1000)
tool_start_ts[tool_name] = ts
has_tool_activity = True
tool_start_ts[call_id] = ts
yield _sse({
"type": "tool_start",
"call_id": call_id,
"tool": tool_name,
"title": _TOOL_TITLES.get(tool_name, tool_name),
"input_summary": _summarize_input(tool_name, tool_input),
@@ -113,14 +114,16 @@ async def _stream_response(request: ChatRequest) -> AsyncIterator[bytes]:
})
elif kind == "on_tool_end":
call_id = event.get("run_id", "")
tool_name = event.get("name", "unknown")
output = event.get("data", {}).get("output", "")
output_str = output if isinstance(output, str) else str(output)
ts = int(time.time() * 1000)
duration_ms = ts - tool_start_ts.pop(tool_name, ts)
duration_ms = ts - tool_start_ts.pop(call_id, ts)
if _is_tool_error(output_str):
yield _sse({
"type": "tool_error",
"call_id": call_id,
"tool": tool_name,
"title": _TOOL_TITLES.get(tool_name, tool_name),
"error_summary": _extract_error_summary(output_str),
@@ -130,6 +133,7 @@ async def _stream_response(request: ChatRequest) -> AsyncIterator[bytes]:
else:
yield _sse({
"type": "tool_end",
"call_id": call_id,
"tool": tool_name,
"title": _TOOL_TITLES.get(tool_name, tool_name),
"output_summary": _summarize_output(tool_name, output_str),
@@ -138,6 +142,17 @@ async def _stream_response(request: ChatRequest) -> AsyncIterator[bytes]:
"ts": ts,
})
elif kind == "on_chain_start":
chain_name = event.get("name", "")
metadata = event.get("metadata", {})
tags = event.get("tags", [])
logger.debug(
"on_chain_start: name=%s, langgraph_node=%s, tags=%s",
chain_name,
metadata.get("langgraph_node", ""),
tags,
)
except Exception as exc:
logger.error("SSE stream error for conversation %s: %s", request.conversation_id, exc, exc_info=True)