wx44wx/ai-video-fullstack
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: persist photos and enrich conversation history

Browse Source
This commit is contained in:
Xin Wang
2026-08-05 16:14:03 +08:00
parent 545536bdf3
commit 317a0600bc
19 changed files with 1146 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/services/brains/prompt_brain.py
+54
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import asyncio
from collections.abc import Awaitable
from time import monotonic
from typing import Any
from uuid import uuid4
@@ -503,12 +504,45 @@ class PromptBrain(BaseBrain):
await params.result_callback(result)
async def call_tool(params: FunctionCallParams) -> None:
invocation_id = f"tool_{uuid4().hex[:20]}"
started_at = monotonic()
await self._emit_trace(
"tool_started",
invocationId=invocation_id,
toolId=tool.id,
toolName=tool.name,
functionName=tool.function_name,
toolType=tool.type,
)
try:
result = await self._tools.execute(tool, dict(params.arguments or {}))
if result["updated_variables"]:
self._refresh_prompt()
await self._emit_trace(
"tool_completed" if result.get("status") == "ok" else "tool_failed",
invocationId=invocation_id,
toolId=tool.id,
toolName=tool.name,
functionName=tool.function_name,
toolType=tool.type,
status=str(result.get("status") or "unknown"),
durationMs=max(0, round((monotonic() - started_at) * 1000)),
updatedVariables=list(result.get("updated_variables") or []),
resultKeys=sorted(str(name) for name in result if name != "data"),
)
await return_result(params, result)
except (ToolExecutionError, ValueError) as exc:
await self._emit_trace(
"tool_failed",
invocationId=invocation_id,
toolId=tool.id,
toolName=tool.name,
functionName=tool.function_name,
toolType=tool.type,
status="error",
durationMs=max(0, round((monotonic() - started_at) * 1000)),
error=str(exc)[:2048],
)
await return_result(
params,
{"status": "error", "message": f"工具调用失败: {exc}"},
@@ -522,6 +556,26 @@ class PromptBrain(BaseBrain):
)
return schema, call_tool
async def _emit_trace(self, event: str, **details: Any) -> None:
runtime = self._runtime
if runtime is None:
return
try:
await runtime.queue_frame(
OutputTransportMessageUrgentFrame(
message={
"type": "workflow-event",
"eventId": f"wfe_{uuid4().hex[:20]}",
"event": event,
"timestamp": time_now_iso8601(),
"sessionId": runtime.session_id,
**details,
}
)
)
except Exception as exc: # noqa: BLE001 - trace must not alter execution
logger.warning(f"发送 Prompt 工具轨迹失败，不影响当前调用: {exc}")
def _make_end_call_tool(self, tool, runtime: BrainRuntime):
config = (tool.definition or {}).get("config") or {}
message_type = str(config.get("message_type") or "none")
backend/services/brains/workflow_brain.py
+38
View File
@@ -6,7 +6,9 @@ import asyncio
from collections.abc import Awaitable
from copy import deepcopy
from dataclasses import dataclass, replace
from time import monotonic
from typing import Any
from uuid import uuid4
from loguru import logger
from models import AssistantConfig, RuntimeTool
@@ -764,10 +766,46 @@ class WorkflowBrain(BaseBrain):
async def handler(args, _flow_manager):
transition_id = self._state.transition_id
invocation_id = f"tool_{uuid4().hex[:20]}"
started_at = monotonic()
await self._emit_trace(
"tool_started",
nodeId=node_id,
invocationId=invocation_id,
toolId=tool.id,
toolName=tool.name,
functionName=tool.function_name,
toolType=tool.type,
)
try:
result = await self._tools.execute(tool, dict(args or {}))
except ToolExecutionError as exc:
await self._emit_trace(
"tool_failed",
nodeId=node_id,
invocationId=invocation_id,
toolId=tool.id,
toolName=tool.name,
functionName=tool.function_name,
toolType=tool.type,
status="error",
durationMs=max(0, round((monotonic() - started_at) * 1000)),
error=str(exc)[:2048],
)
return {"status": "error", "message": str(exc)}
await self._emit_trace(
"tool_completed" if result.get("status") == "ok" else "tool_failed",
nodeId=node_id,
invocationId=invocation_id,
toolId=tool.id,
toolName=tool.name,
functionName=tool.function_name,
toolType=tool.type,
status=str(result.get("status") or "unknown"),
durationMs=max(0, round((monotonic() - started_at) * 1000)),
updatedVariables=list(result.get("updated_variables") or []),
resultKeys=sorted(str(name) for name in result if name != "data"),
)
if (
self._state.current_node_id != node_id
or self._state.transition_id != transition_id
backend/services/client_tools.py
+58 -3
View File
@@ -7,6 +7,7 @@ from collections import deque
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any, Literal, Protocol
from uuid import uuid4
from loguru import logger
from pipecat.frames.frames import (
@@ -17,6 +18,7 @@ from pipecat.frames.frames import (
StopFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.time import time_now_iso8601
class ClientToolError(RuntimeError):
@@ -29,6 +31,7 @@ ClientToolResponseWaitMode = Literal["timeout", "session"]
@dataclass(frozen=True)
class _PendingClientToolCall:
future: asyncio.Future[dict[str, Any]]
function_name: str
interrupt_on_result: bool = False
@@ -80,8 +83,6 @@ class ClientToolBroker(FrameProcessor):
response_wait_mode: ClientToolResponseWaitMode = "timeout",
interrupt_on_result: bool = False,
) -> dict[str, Any]:
from uuid import uuid4
if self._closed_message is not None:
raise ClientToolError(self._closed_message)
@@ -99,9 +100,21 @@ class ClientToolBroker(FrameProcessor):
OutputTransportMessageUrgentFrame(message=message)
)
except Exception as exc:
await self._emit_trace(
"client_tool_failed",
toolCallId=tool_call_id,
functionName=function_name,
error="客户端工具调用发送失败",
)
raise ClientToolError(
f"客户端工具调用发送失败: {function_name}"
) from exc
await self._emit_trace(
"client_tool_completed",
toolCallId=tool_call_id,
functionName=function_name,
status="dispatched",
)
return {
"status": "ok",
"data": {"dispatched": True},
@@ -111,6 +124,7 @@ class ClientToolBroker(FrameProcessor):
future: asyncio.Future[dict[str, Any]] = loop.create_future()
self._pending[tool_call_id] = _PendingClientToolCall(
future=future,
function_name=function_name,
interrupt_on_result=interrupt_on_result,
)
try:
@@ -121,6 +135,12 @@ class ClientToolBroker(FrameProcessor):
return await future
return await asyncio.wait_for(future, timeout=timeout_seconds)
except TimeoutError as exc:
await self._emit_trace(
"client_tool_failed",
toolCallId=tool_call_id,
functionName=function_name,
error="客户端工具调用超时",
)
raise ClientToolError(f"客户端工具调用超时: {function_name}") from exc
except ClientToolError:
raise
@@ -163,10 +183,28 @@ class ClientToolBroker(FrameProcessor):
future = pending.future
status = str(message.get("status") or "error")
result_data = message.get("data")
user_action = (
str(result_data.get("action") or "")
if isinstance(result_data, dict)
else ""
)
await self._emit_trace(
"client_tool_completed" if status == "ok" else "client_tool_failed",
toolCallId=tool_call_id,
functionName=pending.function_name,
status=status,
**({"userAction": user_action} if user_action else {}),
**(
{"error": str(message.get("message") or "客户端工具执行失败")}
if status != "ok"
else {}
),
)
if status == "ok":
response = {
"status": "ok",
"data": message.get("data"),
"data": result_data,
}
if pending.interrupt_on_result:
# Match text input exactly: broadcasting only starts the
@@ -206,6 +244,23 @@ class ClientToolBroker(FrameProcessor):
}
)
async def _emit_trace(self, event: str, **details: Any) -> None:
"""Publish sanitized client interaction metadata for conversation history."""
try:
await self.push_frame(
OutputTransportMessageUrgentFrame(
message={
"type": "workflow-event",
"eventId": f"wfe_{uuid4().hex[:20]}",
"event": event,
"timestamp": time_now_iso8601(),
**details,
}
)
)
except Exception as exc: # noqa: BLE001 - tracing must not alter the tool call
logger.warning(f"发送客户端工具轨迹失败，不影响工具调用: {exc}")
def on_interruption_processed(self) -> None:
"""Release one result after the aggregator acknowledges interruption."""
while self._deferred_results:
backend/services/conversation_history.py
+89 -1
View File
@@ -9,9 +9,10 @@ from copy import deepcopy
from datetime import UTC, datetime
from uuid import uuid4
from db.models import ConversationMessage, ConversationSession
from db.models import ConversationArtifact, ConversationMessage, ConversationSession
from db.session import SessionLocal
from loguru import logger
from services.object_storage import delete_object, put_object, storage_uri
MAX_WORKFLOW_TRACE_EVENTS = 1000
@@ -36,6 +37,7 @@ class ConversationRecorder:
self._trace_sequence = 0
self._lock = asyncio.Lock()
self._seen_events: set[str] = set()
self._pending_artifacts: set[asyncio.Task[None]] = set()
@classmethod
async def start(
@@ -168,6 +170,89 @@ class ConversationRecorder:
except Exception as exc:
logger.error(f"保存对话文本失败，不影响本次通话: {exc}")
def record_image_later(
self,
data: bytes,
*,
input_id: str,
timestamp: object,
mime_type: str = "image/jpeg",
) -> None:
"""Persist a captured frame without delaying the realtime model turn."""
task = asyncio.create_task(
self._record_image(
data,
input_id=input_id,
timestamp=timestamp,
mime_type=mime_type,
),
name=f"conversation-image:{self.session_id}:{input_id}",
)
self._pending_artifacts.add(task)
task.add_done_callback(self._pending_artifacts.discard)
async def _record_image(
self,
data: bytes,
*,
input_id: str,
timestamp: object,
mime_type: str,
) -> None:
message_id = f"msg_{uuid4().hex[:20]}"
artifact_id = f"artifact_{uuid4().hex[:20]}"
extension = ".jpg" if mime_type == "image/jpeg" else ".bin"
key = f"conversations/{self.session_id}/{artifact_id}{extension}"
try:
await asyncio.to_thread(put_object, key, data, mime_type)
except Exception as exc:
logger.error(f"保存会话图片到对象存储失败，不影响本次通话: {exc}")
return
async with self._lock:
next_sequence = self._sequence + 1
try:
async with SessionLocal() as db:
db.add(
ConversationMessage(
id=message_id,
session_id=self.session_id,
sequence=next_sequence,
role="user",
content_type="image",
content="",
occurred_at=_parse_timestamp(timestamp),
extra={
"input_id": input_id,
"source": "camera_capture",
},
)
)
db.add(
ConversationArtifact(
id=artifact_id,
session_id=self.session_id,
message_id=message_id,
kind="image",
storage_uri=storage_uri(key),
mime_type=mime_type,
size_bytes=len(data),
extra={"input_id": input_id},
)
)
conversation = await db.get(ConversationSession, self.session_id)
if conversation:
conversation.message_count = next_sequence
await db.commit()
self._sequence = next_sequence
except Exception as exc:
logger.error(f"保存会话图片索引失败，不影响本次通话: {exc}")
try:
await asyncio.to_thread(delete_object, key)
except Exception:
logger.warning(f"清理未关联的会话图片失败: {key}")
async def finish(self, *, status: str = "completed") -> None:
"""Finish the session even if the owning pipeline is being cancelled."""
finish_task = asyncio.create_task(
@@ -184,6 +269,9 @@ class ConversationRecorder:
raise
async def _finish(self, *, status: str) -> None:
pending_artifacts = list(self._pending_artifacts)
if pending_artifacts:
await asyncio.gather(*pending_artifacts, return_exceptions=True)
async with self._lock:
try:
async with SessionLocal() as db:
backend/services/knowledge.py
+4 -38
View File
@@ -6,9 +6,6 @@ from pathlib import Path
import re
import uuid
import boto3
from botocore.config import Config
from botocore.exceptions import ClientError
from docx import Document as DocxDocument
from openai import AsyncOpenAI
from pypdf import PdfReader
@@ -17,38 +14,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
import settings
from db.models import KnowledgeBase, KnowledgeChunk, KnowledgeDocument, ModelResource
def _s3_client():
return boto3.client(
"s3",
endpoint_url=settings.S3_ENDPOINT_URL,
aws_access_key_id=settings.S3_ACCESS_KEY,
aws_secret_access_key=settings.S3_SECRET_KEY,
region_name=settings.S3_REGION,
config=Config(signature_version="s3v4", s3={"addressing_style": "path"}),
)
def _ensure_bucket_and_put(key: str, data: bytes, mime_type: str) -> None:
client = _s3_client()
try:
client.head_bucket(Bucket=settings.S3_BUCKET)
except ClientError as exc:
status = exc.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
if status != 404:
raise
client.create_bucket(Bucket=settings.S3_BUCKET)
client.put_object(Bucket=settings.S3_BUCKET, Key=key, Body=data, ContentType=mime_type)
def _delete_object(key: str) -> None:
_s3_client().delete_object(Bucket=settings.S3_BUCKET, Key=key)
def _get_object(key: str) -> bytes:
response = _s3_client().get_object(Bucket=settings.S3_BUCKET, Key=key)
return response["Body"].read()
from services.object_storage import delete_object, get_object, put_object
def extract_text(filename: str, data: bytes) -> str:
@@ -122,7 +88,7 @@ async def create_document(
safe_name = Path(name).name
extension = ".txt" if source_type == "text" else Path(safe_name).suffix
storage_key = f"knowledge/{kb.id}/{document_id}/source{extension}"
await asyncio.to_thread(_ensure_bucket_and_put, storage_key, raw_data, mime_type)
await asyncio.to_thread(put_object, storage_key, raw_data, mime_type)
document = KnowledgeDocument(
id=document_id,
@@ -158,7 +124,7 @@ async def process_document(document_id: str) -> None:
await session.commit()
try:
data = await asyncio.to_thread(_get_object, document.storage_key)
data = await asyncio.to_thread(get_object, document.storage_key)
text = (
data.decode("utf-8", errors="replace")
if document.source_type == "text"
@@ -245,4 +211,4 @@ async def search(
async def delete_storage_object(document: KnowledgeDocument) -> None:
if document.storage_key:
await asyncio.to_thread(_delete_object, document.storage_key)
await asyncio.to_thread(delete_object, document.storage_key)
backend/services/object_storage.py
+60
View File
@@ -0,0 +1,60 @@
"""Small S3-compatible object-storage adapter shared by backend features."""
from __future__ import annotations
import boto3
from botocore.config import Config
from botocore.exceptions import ClientError
import settings
def _client():
return boto3.client(
"s3",
endpoint_url=settings.S3_ENDPOINT_URL,
aws_access_key_id=settings.S3_ACCESS_KEY,
aws_secret_access_key=settings.S3_SECRET_KEY,
region_name=settings.S3_REGION,
config=Config(signature_version="s3v4", s3={"addressing_style": "path"}),
)
def put_object(key: str, data: bytes, mime_type: str) -> None:
client = _client()
try:
client.head_bucket(Bucket=settings.S3_BUCKET)
except ClientError as exc:
status = exc.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
if status != 404:
raise
client.create_bucket(Bucket=settings.S3_BUCKET)
client.put_object(
Bucket=settings.S3_BUCKET,
Key=key,
Body=data,
ContentType=mime_type,
)
def get_object(key: str) -> bytes:
response = _client().get_object(Bucket=settings.S3_BUCKET, Key=key)
return response["Body"].read()
def delete_object(key: str) -> None:
_client().delete_object(Bucket=settings.S3_BUCKET, Key=key)
def storage_uri(key: str) -> str:
return f"s3://{settings.S3_BUCKET}/{key.lstrip('/')}"
def key_from_storage_uri(uri: str) -> str:
prefix = f"s3://{settings.S3_BUCKET}/"
if not uri.startswith(prefix):
raise ValueError("不支持的会话附件存储地址")
key = uri[len(prefix) :].strip("/")
if not key:
raise ValueError("会话附件存储地址缺少对象键")
return key
backend/services/pipecat/pipeline.py
+10
View File
@@ -36,6 +36,7 @@ from services.vision import (
config_with_main_llm_as_vision,
config_with_vision_resource,
image_data_uri,
image_jpeg_bytes,
)
from services.workflow_engine import WorkflowEngine
@@ -58,6 +59,7 @@ from pipecat.processors.aggregators.llm_response_universal import (
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.llm_service import FunctionCallParams
from pipecat.utils.time import time_now_iso8601
from services.pipecat.turn_config import (
ConfigurableLLMUserAggregator,
create_user_turn_strategies,
@@ -763,6 +765,14 @@ async def run_pipeline(
except asyncio.TimeoutError as exc:
raise ValueError("等待摄像头视频帧超时") from exc
if recorder:
image_bytes = await asyncio.to_thread(image_jpeg_bytes, image_frame)
recorder.record_image_later(
image_bytes,
input_id=value.input_id,
timestamp=time_now_iso8601(),
)
if native_vision:
input_frame = await asyncio.to_thread(
_multimodal_user_input_frame,
backend/services/pipecat/processors.py
+4 -3
View File
@@ -61,9 +61,10 @@ class UserInput:
@property
def transcript_text(self) -> str:
if not self.has_camera_frame:
return self.text
return f"{self.text}\n已发送一张图片".strip()
# Images are represented as structured media records. A synthetic
# sentence would be shown as chat text and persisted as if the user had
# said it, while adding no visual information for the model.
return self.text
class UserInputError(ValueError):
backend/services/vision.py
+8 -3
View File
@@ -30,8 +30,8 @@ def _require(value: str, label: str) -> str:
raise ValueError(f"缺少模型资源配置: {label}")
def image_data_uri(frame: UserImageRawFrame) -> str:
"""Encode one Pipecat camera frame for a vision chat-completion request."""
def image_jpeg_bytes(frame: UserImageRawFrame) -> bytes:
"""Encode one Pipecat camera frame as a storage- and model-ready JPEG."""
if not frame.format:
raise ValueError("摄像头图片帧缺少 format,无法编码给视觉模型")
buffer = BytesIO()
@@ -40,7 +40,12 @@ def image_data_uri(frame: UserImageRawFrame) -> str:
format="JPEG",
quality=85,
)
encoded = base64.b64encode(buffer.getvalue()).decode("utf-8")
return buffer.getvalue()
def image_data_uri(frame: UserImageRawFrame) -> str:
"""Encode one Pipecat camera frame for a vision chat-completion request."""
encoded = base64.b64encode(image_jpeg_bytes(frame)).decode("utf-8")
return f"data:image/jpeg;base64,{encoded}"