kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat: 集成 Langfuse 以增强 Yuxi 中的可观察性

Browse Source 
- 更新了 Dockerfile，在同步过程中包含 Langfuse 集成。
- 在文档中添加了新的部分，详细介绍 Langfuse 集成的目的、配置和在 Yuxi 中的使用方法。
- 更新了 Vitepress 配置，包含了指向新 Langfuse 集成文档的链接。
- 增强了路线图，反映了对 Langfuse 自托管支持的添加，并明确了集成方法。
This commit is contained in:
Wenjie Zhang 2026-03-31 09:58:16 +08:00
parent 81a07570d1
commit b4401d0339
15 changed files with 3072 additions and 2258 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

69
.env.template
View File

@ -1,40 +1,6 @@
MODEL_DIR=./models
SAVE_DIR=./saves
# Sandbox (deerFlow-style provisioner)
SANDBOX_PROVIDER=provisioner
SANDBOX_PROVISIONER_URL=http://sandbox-provisioner:8002
SANDBOX_VIRTUAL_PATH_PREFIX=/home/gem/user-data
SANDBOX_EXEC_TIMEOUT_SECONDS=180
SANDBOX_MAX_OUTPUT_BYTES=262144
SANDBOX_KEEPALIVE_INTERVAL_SECONDS=30
SANDBOX_IDLE_TIMEOUT_SECONDS=120
SANDBOX_IDLE_CHECK_INTERVAL_SECONDS=10
# sandbox-provisioner backend: memory | docker | kubernetes
# `local` 仍兼容，但只是 `docker` 的历史别名，不再推荐继续配置
SANDBOX_PROVISIONER_BACKEND=docker
# sandbox-provisioner 通用配置
# SANDBOX_IMAGE=enterprise-public-cn-beijing.cr.volces.com/vefaas-public/all-in-one-sandbox:latest
# SANDBOX_CONTAINER_PORT=8080
# SANDBOX_HEALTH_TIMEOUT_SECONDS=300
# MEMORY_SANDBOX_URL_TEMPLATE=http://agent-sandbox:8000
# SANDBOX_HTTP_PROXY=http://host.docker.internal:7897
# SANDBOX_HTTPS_PROXY=http://host.docker.internal:7897
# Docker backend 专用 (used when SANDBOX_PROVISIONER_BACKEND=docker/local)
# SANDBOX_DOCKER_NETWORK=yuxi-know_app-network
# SANDBOX_DOCKER_THREADS_HOST_PATH=
# SANDBOX_DOCKER_SANDBOX_PREFIX=yuxi-sandbox
# SANDBOX_DOCKER_SANDBOX_HOST=host.docker.internal
# Kubernetes backend 专用 (used when SANDBOX_PROVISIONER_BACKEND=kubernetes)
# SANDBOX_K8S_NAMESPACE=yuxi-know
# SANDBOX_NODE_HOST=host.docker.internal
# KUBECONFIG_PATH=/root/.kube/config
# THREAD_PVC=yuxi-thread
# SKILLS_PVC=yuxi-skills # 当前代码会读取，但 Pod 挂载实际仍只使用 THREAD_PVC
# region model_provider
SILICONFLOW_API_KEY= # 推荐使用硅基流动免费服务 https://cloud.siliconflow.cn/i/Eo5yTHGJ
TAVILY_API_KEY= # 获取搜索服务的 api key 请访问 https://app.tavily.com/
@ -69,3 +35,38 @@ TAVILY_API_KEY= # 获取搜索服务的 api key 请访问 https://app.tavily.co
# LightRag llm 并发限制
# MAX_ASYNC=5
# EMBEDDING_FUNC_MAX_ASYNC=8
# Sandbox (deerFlow-style provisioner)
# SANDBOX_PROVIDER=provisioner
# SANDBOX_PROVISIONER_URL=http://sandbox-provisioner:8002
# SANDBOX_VIRTUAL_PATH_PREFIX=/home/gem/user-data
# SANDBOX_EXEC_TIMEOUT_SECONDS=180
# SANDBOX_MAX_OUTPUT_BYTES=262144
# SANDBOX_KEEPALIVE_INTERVAL_SECONDS=30
# SANDBOX_IDLE_TIMEOUT_SECONDS=120
# SANDBOX_IDLE_CHECK_INTERVAL_SECONDS=10
# # sandbox-provisioner backend: memory | docker | kubernetes
# # `local` 仍兼容，但只是 `docker` 的历史别名，不再推荐继续配置
# SANDBOX_PROVISIONER_BACKEND=docker
# sandbox-provisioner 通用配置
# SANDBOX_IMAGE=enterprise-public-cn-beijing.cr.volces.com/vefaas-public/all-in-one-sandbox:latest
# SANDBOX_CONTAINER_PORT=8080
# SANDBOX_HEALTH_TIMEOUT_SECONDS=300
# MEMORY_SANDBOX_URL_TEMPLATE=http://agent-sandbox:8000
# SANDBOX_HTTP_PROXY=http://host.docker.internal:7897
# SANDBOX_HTTPS_PROXY=http://host.docker.internal:7897
# Docker backend 专用 (used when SANDBOX_PROVISIONER_BACKEND=docker/local)
# SANDBOX_DOCKER_NETWORK=yuxi-know_app-network
# SANDBOX_DOCKER_THREADS_HOST_PATH=
# SANDBOX_DOCKER_SANDBOX_PREFIX=yuxi-sandbox
# SANDBOX_DOCKER_SANDBOX_HOST=host.docker.internal
# Kubernetes backend 专用 (used when SANDBOX_PROVISIONER_BACKEND=kubernetes)
# SANDBOX_K8S_NAMESPACE=yuxi-know
# SANDBOX_NODE_HOST=host.docker.internal
# KUBECONFIG_PATH=/root/.kube/config
# THREAD_PVC=yuxi-thread
# SKILLS_PVC=yuxi-skills # 当前代码会读取，但 Pod 挂载实际仍只使用 THREAD_PVC

3
AGENTS.md
View File

@ -50,8 +50,9 @@ docker compose exec api uv run python test/your_script.py # 放在 test 文件
- 如果需要新建说明文档（仅开发者可见，非必要不创建），则保存在 `docs/vibe` 文件夹下面
- 代码更新后要检查文档部分是否有需要更新的地方，文档的目录定义在 `docs/.vitepress/config.mts` 中
- 如果新增面向用户的正式文档，除了补正文档内容外，还需要同步更新 `docs/.vitepress/config.mts` 的导航；Langfuse 集成说明归档在 `docs/agents` 分组下维护，并同步更新 `docs/develop-guides/roadmap.md`
## 提交规范
1. 参考 [Conventional Commits](https://www.conventionalcommits.org/en/v1.0.0/) 规范编写提交信息。
2. 使用中文提交信息，标题简洁明了，描述具体改动内容和原因。
2. 使用中文提交信息，标题简洁明了，描述具体改动内容和原因。

16
backend/package/yuxi/agents/base.py
View File

@ -80,6 +80,14 @@ class BaseAgent:
"recursion_limit": 300,
}
# langfuse metadata and callbacks integration
if callbacks := kwargs.get("callbacks"):
input_config["callbacks"] = list(callbacks)
if metadata := kwargs.get("metadata"):
input_config["metadata"] = dict(metadata)
if tags := kwargs.get("tags"):
input_config["tags"] = list(tags)
async for msg, metadata in graph.astream(
{"messages": messages},
stream_mode="messages",
@ -100,6 +108,14 @@ class BaseAgent:
"recursion_limit": 100,
}
# langfuse metadata and callbacks integration
if callbacks := kwargs.get("callbacks"):
input_config["callbacks"] = list(callbacks)
if metadata := kwargs.get("metadata"):
input_config["metadata"] = dict(metadata)
if tags := kwargs.get("tags"):
input_config["tags"] = list(tags)
msg = await graph.ainvoke(
{"messages": messages},
context=context,

154
backend/package/yuxi/services/chat_service.py
View File

@ -14,6 +14,12 @@ from yuxi.agents.state import AgentStatePayload
from yuxi.plugins.guard import content_guard
from yuxi.repositories.agent_config_repository import AgentConfigRepository
from yuxi.repositories.conversation_repository import ConversationRepository
from yuxi.services.langfuse_service import (
LangfuseRunContext,
build_run_context,
flush_langfuse,
get_trace_info,
)
from yuxi.storage.postgres.manager import pg_manager
from yuxi.storage.postgres.models_business import User
from yuxi.utils.logging_config import logger
@ -71,6 +77,30 @@ async def _get_langgraph_messages(agent_instance, config_dict):
return state.values.get("messages", [])
def _build_langfuse_run_context(
*,
current_user,
thread_id: str,
agent_id: str,
request_id: str,
operation: str,
agent_config_id: int | None = None,
message_type: str | None = None,
) -> LangfuseRunContext:
return build_run_context(
user_id=str(current_user.id),
thread_id=thread_id,
agent_id=agent_id,
request_id=request_id,
operation=operation,
agent_config_id=agent_config_id,
message_type=message_type,
username=getattr(current_user, "username", None),
login_user_id=getattr(current_user, "user_id", None),
department_id=getattr(current_user, "department_id", None),
)
def extract_agent_state(values: dict) -> AgentStatePayload:
"""从 LangGraph state 中提取 agent 状态"""
if not isinstance(values, dict):
@ -97,16 +127,24 @@ async def _get_existing_message_ids(conv_repo: ConversationRepository, thread_id
}
async def _save_ai_message(conv_repo: ConversationRepository, thread_id: str, msg_dict: dict) -> None:
async def _save_ai_message(
conv_repo: ConversationRepository,
thread_id: str,
msg_dict: dict,
trace_info: dict[str, Any] | None = None,
) -> None:
content = msg_dict.get("content", "")
tool_calls_data = msg_dict.get("tool_calls", [])
extra_metadata = dict(msg_dict)
if trace_info:
extra_metadata.update(trace_info)
ai_msg = await conv_repo.add_message_by_thread_id(
thread_id=thread_id,
role="assistant",
content=content,
message_type="text",
extra_metadata=msg_dict,
extra_metadata=extra_metadata,
)
if ai_msg and tool_calls_data:
@ -145,6 +183,7 @@ async def save_partial_message(
full_msg=None,
error_message: str | None = None,
error_type: str = "interrupted",
trace_info: dict[str, Any] | None = None,
):
try:
extra_metadata = {
@ -159,6 +198,9 @@ async def save_partial_message(
else:
content = ""
if trace_info:
extra_metadata.update(trace_info)
return await conv_repo.add_message_by_thread_id(
thread_id=thread_id,
role="assistant",
@ -178,6 +220,7 @@ async def save_messages_from_langgraph_state(
thread_id: str,
conv_repo: ConversationRepository,
config_dict: dict,
trace_info: dict[str, Any] | None = None,
) -> None:
try:
messages = await _get_langgraph_messages(agent_instance, config_dict)
@ -194,7 +237,7 @@ async def save_messages_from_langgraph_state(
continue
if msg_type == "ai":
await _save_ai_message(conv_repo, thread_id, msg_dict)
await _save_ai_message(conv_repo, thread_id, msg_dict, trace_info=trace_info)
elif msg_type == "tool":
await _save_tool_message(conv_repo, msg_dict)
@ -483,6 +526,16 @@ async def agent_chat(
logger.warning(f"No thread_id provided, generated new thread_id: {thread_id}")
input_context = agent_config | {"user_id": user_id, "thread_id": thread_id}
langfuse_run = _build_langfuse_run_context(
current_user=current_user,
thread_id=thread_id,
agent_id=agent_id,
request_id=meta["request_id"],
operation="agent_chat_sync",
agent_config_id=agent_config_id,
message_type=message_type,
)
trace_info: dict[str, Any] = {}
try:
conv_repo = ConversationRepository(db)
@ -507,8 +560,15 @@ async def agent_chat(
logger.error(f"Error saving user message: {e}")
langgraph_config = {"configurable": {"thread_id": thread_id, "user_id": user_id}}
invoke_result = await agent.invoke_messages(messages, input_context=input_context)
invoke_result = await agent.invoke_messages(
messages,
input_context=input_context,
callbacks=langfuse_run.callbacks,
metadata=langfuse_run.metadata,
tags=langfuse_run.tags,
)
full_msg = _extract_ai_message(invoke_result.get("messages") if isinstance(invoke_result, dict) else None)
trace_info = get_trace_info(langfuse_run)
if full_msg is None:
try:
@ -521,7 +581,13 @@ async def agent_chat(
full_content = full_msg.content if full_msg else ""
if conf.enable_content_guard and await content_guard.check(full_content):
await save_partial_message(conv_repo, thread_id, full_msg, "content_guard_blocked")
await save_partial_message(
conv_repo,
thread_id,
full_msg,
"content_guard_blocked",
trace_info=trace_info,
)
return {
"status": "interrupted",
"message": "检测到敏感内容，已中断输出",
@ -541,6 +607,7 @@ async def agent_chat(
thread_id=thread_id,
conv_repo=conv_repo,
config_dict=langgraph_config,
trace_info=trace_info,
)
return {
@ -560,6 +627,8 @@ async def agent_chat(
"error_message": str(e),
"request_id": meta.get("request_id"),
}
finally:
flush_langfuse()
async def stream_agent_chat(
@ -657,8 +726,18 @@ async def stream_agent_chat(
logger.warning(f"No thread_id provided, generated new thread_id: {thread_id}")
input_context = agent_config | {"user_id": user_id, "thread_id": thread_id}
langfuse_run = _build_langfuse_run_context(
current_user=current_user,
thread_id=thread_id,
agent_id=agent_id,
request_id=meta["request_id"],
operation="agent_chat_stream",
agent_config_id=agent_config_id,
message_type=message_type,
)
full_msg = None
accumulated_content: list[str] = []
trace_info: dict[str, Any] = {}
try:
conv_repo = ConversationRepository(db)
@ -690,14 +769,27 @@ async def stream_agent_chat(
full_msg = None
accumulated_content = []
async for msg, metadata in agent.stream_messages(messages, input_context=input_context):
async for msg, metadata in agent.stream_messages(
messages,
input_context=input_context,
callbacks=langfuse_run.callbacks,
metadata=langfuse_run.metadata,
tags=langfuse_run.tags,
):
if isinstance(msg, AIMessageChunk):
accumulated_content.append(msg.content)
trace_info = get_trace_info(langfuse_run)
content_for_check = "".join(accumulated_content[-10:])
if conf.enable_content_guard and await content_guard.check_with_keywords(content_for_check):
full_msg = AIMessage(content="".join(accumulated_content))
await save_partial_message(conv_repo, thread_id, full_msg, "content_guard_blocked")
await save_partial_message(
conv_repo,
thread_id,
full_msg,
"content_guard_blocked",
trace_info=trace_info,
)
meta["time_cost"] = asyncio.get_event_loop().time() - start_time
yield make_chunk(status="interrupted", message="检测到敏感内容，已中断输出", meta=meta)
return
@ -705,6 +797,7 @@ async def stream_agent_chat(
yield make_chunk(content=msg.content, msg=msg.model_dump(), metadata=metadata, status="loading")
else:
msg_dict = msg.model_dump()
trace_info = get_trace_info(langfuse_run)
yield make_chunk(msg=msg_dict, metadata=metadata, status="loading")
try:
@ -718,9 +811,16 @@ async def stream_agent_chat(
logger.error(f"Error processing tool message: {e}")
full_msg = _ensure_full_msg(full_msg, accumulated_content)
trace_info = get_trace_info(langfuse_run)
if conf.enable_content_guard and hasattr(full_msg, "content") and await content_guard.check(full_msg.content):
await save_partial_message(conv_repo, thread_id, full_msg, "content_guard_blocked")
await save_partial_message(
conv_repo,
thread_id,
full_msg,
"content_guard_blocked",
trace_info=trace_info,
)
meta["time_cost"] = asyncio.get_event_loop().time() - start_time
yield make_chunk(status="interrupted", message="检测到敏感内容，已中断输出", meta=meta)
return
@ -745,6 +845,7 @@ async def stream_agent_chat(
thread_id=thread_id,
conv_repo=conv_repo,
config_dict=langgraph_config,
trace_info=trace_info,
)
yield make_chunk(status="finished", meta=meta)
@ -764,6 +865,7 @@ async def stream_agent_chat(
full_msg=full_msg,
error_message="对话已中断" if not full_msg else None,
error_type="interrupted",
trace_info=trace_info,
)
cleanup_task = asyncio.create_task(save_cleanup())
@ -792,9 +894,12 @@ async def stream_agent_chat(
full_msg=full_msg,
error_message=error_msg,
error_type=error_type,
trace_info=trace_info,
)
yield make_chunk(status="error", error_type=error_type, error_message=error_msg, meta=meta)
finally:
flush_langfuse()
async def stream_agent_resume(
@ -844,16 +949,32 @@ async def stream_agent_resume(
context.update(agent_config or {})
context.update({"user_id": user_id, "thread_id": thread_id})
graph = await agent.get_graph(context=context)
langfuse_run = _build_langfuse_run_context(
current_user=current_user,
thread_id=thread_id,
agent_id=agent_id,
request_id=meta.get("request_id") or str(uuid.uuid4()),
operation="agent_chat_resume",
agent_config_id=agent_config_id,
message_type="resume",
)
trace_info: dict[str, Any] = {}
stream_source = graph.astream(
resume_command,
context=context,
config={"configurable": {"thread_id": thread_id, "user_id": user_id}},
config={
"configurable": {"thread_id": thread_id, "user_id": user_id},
"callbacks": langfuse_run.callbacks,
"metadata": langfuse_run.metadata,
"tags": langfuse_run.tags,
},
stream_mode="messages",
)
try:
async for msg, metadata in stream_source:
trace_info = get_trace_info(langfuse_run)
msg_dict = msg.model_dump()
if "id" not in msg_dict:
msg_dict["id"] = str(uuid.uuid4())
@ -875,6 +996,7 @@ async def stream_agent_resume(
thread_id=thread_id,
conv_repo=conv_repo,
config_dict=langgraph_config,
trace_info=trace_info,
)
yield make_resume_chunk(status="finished", meta=meta)
@ -885,7 +1007,11 @@ async def stream_agent_resume(
async with pg_manager.get_async_session_context() as new_db:
new_conv_repo = ConversationRepository(new_db)
await save_partial_message(
new_conv_repo, thread_id, error_message="对话恢复已中断", error_type="resume_interrupted"
new_conv_repo,
thread_id,
error_message="对话恢复已中断",
error_type="resume_interrupted",
trace_info=trace_info,
)
yield make_resume_chunk(status="interrupted", message="对话恢复已中断", meta=meta)
@ -896,10 +1022,16 @@ async def stream_agent_resume(
async with pg_manager.get_async_session_context() as new_db:
new_conv_repo = ConversationRepository(new_db)
await save_partial_message(
new_conv_repo, thread_id, error_message=f"Error during resume: {e}", error_type="resume_error"
new_conv_repo,
thread_id,
error_message=f"Error during resume: {e}",
error_type="resume_error",
trace_info=trace_info,
)
yield make_resume_chunk(message=f"Error during resume: {e}", status="error")
finally:
flush_langfuse()
async def get_agent_state_view(

210
backend/package/yuxi/services/langfuse_service.py Normal file
View File

@ -0,0 +1,210 @@
from __future__ import annotations
import asyncio
import os
from dataclasses import dataclass, field
from functools import lru_cache
from typing import Any
from yuxi.utils.logging_config import logger
try:
from langfuse import Langfuse
from langfuse.langchain import CallbackHandler
except Exception: # pragma: no cover - optional dependency during local test collection
Langfuse = None # type: ignore[assignment]
CallbackHandler = None # type: ignore[assignment]
_FALSE_VALUES = {"0", "false", "no", "off"}
@dataclass(slots=True)
class LangfuseRunContext:
callbacks: list[Any] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
tags: list[str] = field(default_factory=list)
trace_id: str | None = None
def is_langfuse_enabled() -> bool:
enabled_raw = (os.getenv("LANGFUSE_ENABLED") or "true").strip().lower()
if enabled_raw in _FALSE_VALUES:
return False
if Langfuse is None or CallbackHandler is None:
return False
return bool(os.getenv("LANGFUSE_PUBLIC_KEY") and os.getenv("LANGFUSE_SECRET_KEY"))
@lru_cache(maxsize=1)
def get_langfuse_client() -> Langfuse | None:
if not is_langfuse_enabled():
return None
if Langfuse is None:
return None
kwargs: dict[str, Any] = {
"public_key": os.getenv("LANGFUSE_PUBLIC_KEY"),
"secret_key": os.getenv("LANGFUSE_SECRET_KEY"),
}
host = os.getenv("LANGFUSE_BASE_URL")
if host:
kwargs["host"] = host
try:
return Langfuse(**kwargs)
except Exception as exc:
logger.warning(f"初始化 Langfuse 客户端失败，将跳过 tracing: {exc}")
return None
def build_trace_metadata(
*,
user_id: str,
thread_id: str,
agent_id: str,
request_id: str,
operation: str,
agent_config_id: int | None = None,
message_type: str | None = None,
username: str | None = None,
login_user_id: str | None = None,
department_id: int | str | None = None,
) -> dict[str, Any]:
metadata: dict[str, Any] = {
"langfuse_user_id": user_id,
"langfuse_session_id": thread_id,
"request_id": request_id,
"thread_id": thread_id,
"agent_id": agent_id,
"operation": operation,
"source": "yuxi",
"feature": "chat",
}
if agent_config_id is not None:
metadata["agent_config_id"] = str(agent_config_id)
if message_type:
metadata["message_type"] = message_type
if username:
metadata["username"] = username
if login_user_id:
metadata["login_user_id"] = login_user_id
if department_id is not None:
metadata["department_id"] = str(department_id)
return metadata
def build_trace_tags(*, agent_id: str, operation: str, message_type: str | None = None) -> list[str]:
tags = ["yuxi", "chat", operation, f"agent:{agent_id}"]
if message_type:
tags.append(f"message_type:{message_type}")
return tags
def build_run_context(
*,
user_id: str,
thread_id: str,
agent_id: str,
request_id: str,
operation: str,
agent_config_id: int | None = None,
message_type: str | None = None,
username: str | None = None,
login_user_id: str | None = None,
department_id: int | str | None = None,
) -> LangfuseRunContext:
metadata = build_trace_metadata(
user_id=user_id,
thread_id=thread_id,
agent_id=agent_id,
request_id=request_id,
operation=operation,
agent_config_id=agent_config_id,
message_type=message_type,
username=username,
login_user_id=login_user_id,
department_id=department_id,
)
tags = build_trace_tags(agent_id=agent_id, operation=operation, message_type=message_type)
client = get_langfuse_client()
if client is None or CallbackHandler is None:
return LangfuseRunContext(metadata=metadata, tags=tags)
trace_id = client.create_trace_id(seed=request_id)
handler = CallbackHandler(trace_context={"trace_id": trace_id})
return LangfuseRunContext(callbacks=[handler], metadata=metadata, tags=tags, trace_id=trace_id)
def get_trace_info(run_context: LangfuseRunContext | None) -> dict[str, Any]:
if run_context is None:
return {}
metadata = run_context.metadata or {}
trace_id = run_context.trace_id
if run_context.callbacks:
last_trace_id = getattr(run_context.callbacks[0], "last_trace_id", None)
if last_trace_id:
trace_id = last_trace_id
if not trace_id:
return {}
trace_info = {
"langfuse_trace_id": trace_id,
"langfuse_user_id": metadata.get("langfuse_user_id"),
"langfuse_session_id": metadata.get("langfuse_session_id"),
}
# Do not fetch trace_url on the request critical path. Langfuse resolves the
# project id via a remote API call, which can add noticeable latency when the
# base URL is slow or unreachable. If a trace URL is still needed, fetch it
# later via get_trace_url_async() and patch message metadata asynchronously.
return trace_info
async def get_trace_url_async(
run_context: LangfuseRunContext | None,
*,
timeout: float = 5.0,
) -> str | None:
if run_context is None:
return None
trace_id = run_context.trace_id
if run_context.callbacks:
last_trace_id = getattr(run_context.callbacks[0], "last_trace_id", None)
if last_trace_id:
trace_id = last_trace_id
if not trace_id:
return None
client = get_langfuse_client()
if client is None:
return None
try:
return await asyncio.wait_for(
asyncio.to_thread(client.get_trace_url, trace_id=trace_id),
timeout=timeout,
)
except Exception:
return None
def flush_langfuse() -> None:
client = get_langfuse_client()
if client is None:
return
try:
client.flush()
except Exception as exc:
logger.warning(f"刷新 Langfuse 事件失败: {exc}")

1
backend/pyproject.toml
View File

@ -20,6 +20,7 @@ dependencies = [
"langchain-openai>=1.0.2",
"langchain-tavily>=0.2.13",
"langchain-text-splitters>=1.0",
"langfuse>=4.0.0",
"langgraph>=1.0.1",
"langgraph-checkpoint-sqlite>=3.0",
"langgraph-checkpoint-postgres>=2.0.0",

81
backend/test/unit/services/test_base_agent_langfuse_config.py Normal file
View File

@ -0,0 +1,81 @@
from __future__ import annotations
from types import SimpleNamespace
import pytest
from yuxi.agents.base import BaseAgent
class _FakeGraph:
def __init__(self):
self.last_stream_config = None
self.last_invoke_config = None
async def astream(self, payload, *, stream_mode, context, config):
self.last_stream_config = config
yield SimpleNamespace(model_dump=lambda: {"type": "ai"}), {"node": "llm"}
async def ainvoke(self, payload, *, context, config):
self.last_invoke_config = config
return {"messages": []}
class _TestAgent(BaseAgent):
name = "test_agent"
description = "test"
async def get_graph(self, **kwargs):
if getattr(self, "_graph", None) is None:
self._graph = _FakeGraph()
return self._graph
_TestAgent.__module__ = "yuxi.agents.tests.fake"
@pytest.mark.asyncio
async def test_base_agent_stream_messages_passes_callbacks_metadata_and_tags():
agent = _TestAgent()
items = []
async for item in agent.stream_messages(
["hello"],
input_context={"user_id": "user-1", "thread_id": "thread-1"},
callbacks=["handler-1"],
metadata={"langfuse_user_id": "user-1"},
tags=["yuxi"],
):
items.append(item)
graph = await agent.get_graph()
assert len(items) == 1
assert graph.last_stream_config == {
"configurable": {"thread_id": "thread-1", "user_id": "user-1"},
"recursion_limit": 300,
"callbacks": ["handler-1"],
"metadata": {"langfuse_user_id": "user-1"},
"tags": ["yuxi"],
}
@pytest.mark.asyncio
async def test_base_agent_invoke_messages_passes_callbacks_metadata_and_tags():
agent = _TestAgent()
await agent.invoke_messages(
["hello"],
input_context={"user_id": "user-1", "thread_id": "thread-1"},
callbacks=["handler-1"],
metadata={"langfuse_user_id": "user-1"},
tags=["yuxi"],
)
graph = await agent.get_graph()
assert graph.last_invoke_config == {
"configurable": {"thread_id": "thread-1", "user_id": "user-1"},
"recursion_limit": 100,
"callbacks": ["handler-1"],
"metadata": {"langfuse_user_id": "user-1"},
"tags": ["yuxi"],
}

152
backend/test/unit/services/test_chat_service_langfuse_stream.py Normal file
View File

@ -0,0 +1,152 @@
from __future__ import annotations
import json
from types import SimpleNamespace
import pytest
from langchain.messages import AIMessageChunk, HumanMessage
from yuxi.services import chat_service as svc
class _FakeConvRepo:
def __init__(self, _db):
self.saved_messages: list[dict] = []
self.bound_agent_configs: list[tuple[str, int]] = []
self.conversations: dict[str, SimpleNamespace] = {}
async def add_message_by_thread_id(
self,
*,
thread_id: str,
role: str,
content: str,
message_type: str = "text",
extra_metadata: dict | None = None,
image_content: str | None = None,
):
self.saved_messages.append(
{
"thread_id": thread_id,
"role": role,
"content": content,
"message_type": message_type,
"extra_metadata": extra_metadata,
"image_content": image_content,
}
)
return SimpleNamespace(id=1)
async def get_conversation_by_thread_id(self, thread_id: str):
return self.conversations.get(thread_id)
async def create_conversation(self, *, user_id: str, agent_id: str, thread_id: str):
conversation = SimpleNamespace(
user_id=user_id,
agent_id=agent_id,
thread_id=thread_id,
extra_metadata={},
)
self.conversations[thread_id] = conversation
return conversation
async def bind_agent_config(self, thread_id: str, agent_config_id: int):
conversation = self.conversations.setdefault(
thread_id,
SimpleNamespace(user_id="user-1", agent_id="test-agent", thread_id=thread_id, extra_metadata={}),
)
conversation.extra_metadata["agent_config_id"] = agent_config_id
self.bound_agent_configs.append((thread_id, agent_config_id))
@pytest.mark.asyncio
async def test_stream_agent_chat_passes_langfuse_callbacks_and_persists_trace_info(monkeypatch: pytest.MonkeyPatch):
calls: dict[str, object] = {}
class FakeAgent:
async def stream_messages(self, messages, input_context=None, **kwargs):
calls["stream_messages"] = messages
calls["stream_input_context"] = input_context
calls["stream_kwargs"] = kwargs
yield AIMessageChunk(content="hello"), {"node": "llm"}
async def get_graph(self):
class FakeGraph:
async def aget_state(self, config):
return SimpleNamespace(values={"messages": [], "files": {}, "artifacts": []})
return FakeGraph()
async def fake_get_agent_config_by_id(db, user, agent_config_id):
return SimpleNamespace(agent_id="test-agent", config_json={"context": {"temperature": 0.1}})
async def fake_save_messages_from_langgraph_state(*, agent_instance, thread_id, conv_repo, config_dict, trace_info):
calls["saved_state"] = {
"thread_id": thread_id,
"config_dict": config_dict,
"trace_info": trace_info,
}
async def fake_guard_check(_content):
return False
async def fake_guard_check_with_keywords(_content):
return False
async def fake_interrupts(agent, langgraph_config, make_chunk, meta, thread_id):
if False:
yield None
return
monkeypatch.setattr(svc.agent_manager, "get_agent", lambda agent_id: FakeAgent())
monkeypatch.setattr(svc, "get_agent_config_by_id", fake_get_agent_config_by_id)
monkeypatch.setattr(svc, "ConversationRepository", _FakeConvRepo)
monkeypatch.setattr(svc, "save_messages_from_langgraph_state", fake_save_messages_from_langgraph_state)
monkeypatch.setattr(svc.content_guard, "check", fake_guard_check)
monkeypatch.setattr(svc.content_guard, "check_with_keywords", fake_guard_check_with_keywords)
monkeypatch.setattr(svc, "check_and_handle_interrupts", fake_interrupts)
monkeypatch.setattr(
svc,
"_build_langfuse_run_context",
lambda **kwargs: SimpleNamespace(
callbacks=["handler-1"],
metadata={"langfuse_user_id": kwargs["current_user"].id, "langfuse_session_id": kwargs["thread_id"]},
tags=["yuxi", "chat"],
trace_id="trace-seeded",
),
)
monkeypatch.setattr(
svc,
"get_trace_info",
lambda _run_context: {
"langfuse_trace_id": "trace-runtime",
"langfuse_session_id": "thread-1",
},
)
monkeypatch.setattr(svc, "flush_langfuse", lambda: calls.setdefault("flushed", True))
chunks = []
async for chunk in svc.stream_agent_chat(
query="hello",
agent_config_id=123,
thread_id="thread-1",
meta={"request_id": "req-1"},
image_content=None,
current_user=SimpleNamespace(id="user-1", department_id="dept-1"),
db=object(),
):
chunks.append(json.loads(chunk.decode("utf-8")))
assert calls["stream_input_context"] == {"temperature": 0.1, "user_id": "user-1", "thread_id": "thread-1"}
assert calls["stream_kwargs"] == {
"callbacks": ["handler-1"],
"metadata": {"langfuse_user_id": "user-1", "langfuse_session_id": "thread-1"},
"tags": ["yuxi", "chat"],
}
assert calls["saved_state"]["trace_info"] == {
"langfuse_trace_id": "trace-runtime",
"langfuse_session_id": "thread-1",
}
assert chunks[-1]["status"] == "finished"
assert calls["flushed"] is True
assert isinstance(calls["stream_messages"][0], HumanMessage)

40
backend/test/unit/services/test_chat_service_sync.py
View File

@ -71,6 +71,7 @@ async def test_agent_chat_uses_invoke_messages_and_persists_langgraph_state(monk
async def invoke_messages(self, messages, input_context=None, **kwargs):
calls["invoke_messages"] = messages
calls["invoke_input_context"] = input_context
calls["invoke_kwargs"] = kwargs
return {"messages": [messages[0], AIMessage(content="Hi from invoke")]}
async def stream_messages(self, messages, input_context=None, **kwargs):
@ -84,17 +85,34 @@ async def test_agent_chat_uses_invoke_messages_and_persists_langgraph_state(monk
assert agent_config_id == 123
return SimpleNamespace(agent_id="test-agent", config_json={"context": {"temperature": 0.1}})
async def fake_save_messages_from_langgraph_state(*, agent_instance, thread_id, conv_repo, config_dict):
async def fake_save_messages_from_langgraph_state(*, agent_instance, thread_id, conv_repo, config_dict, trace_info):
calls["saved_state"] = {
"agent_instance": agent_instance,
"thread_id": thread_id,
"conv_repo": conv_repo,
"config_dict": config_dict,
"trace_info": trace_info,
}
async def fake_guard_check(_content):
return False
def fake_build_langfuse_run_context(**kwargs):
calls["langfuse_kwargs"] = kwargs
return SimpleNamespace(
callbacks=["handler-1"],
metadata={"langfuse_user_id": kwargs["current_user"].id, "langfuse_session_id": kwargs["thread_id"]},
tags=["yuxi", "chat"],
trace_id="trace-seeded",
)
def fake_get_trace_info(_run_context):
return {"langfuse_trace_id": "trace-runtime", "langfuse_session_id": "thread-1"}
monkeypatch.setattr(svc, "_build_langfuse_run_context", fake_build_langfuse_run_context)
monkeypatch.setattr(svc, "get_trace_info", fake_get_trace_info)
monkeypatch.setattr(svc, "flush_langfuse", lambda: calls.setdefault("flushed", True))
monkeypatch.setattr(svc.agent_manager, "get_agent", lambda agent_id: FakeAgent())
monkeypatch.setattr(svc, "get_agent_config_by_id", fake_get_agent_config_by_id)
monkeypatch.setattr(svc, "ConversationRepository", _FakeConvRepo)
@ -123,9 +141,19 @@ async def test_agent_chat_uses_invoke_messages_and_persists_langgraph_state(monk
assert isinstance(invoke_messages[0], HumanMessage)
assert invoke_messages[0].content == "hello"
assert calls["invoke_input_context"] == {"temperature": 0.1, "user_id": "user-1", "thread_id": "thread-1"}
assert calls["invoke_kwargs"] == {
"callbacks": ["handler-1"],
"metadata": {"langfuse_user_id": "user-1", "langfuse_session_id": "thread-1"},
"tags": ["yuxi", "chat"],
}
assert calls["saved_state"]["thread_id"] == "thread-1"
assert calls["saved_state"]["config_dict"] == {"configurable": {"thread_id": "thread-1", "user_id": "user-1"}}
assert calls["saved_state"]["trace_info"] == {
"langfuse_trace_id": "trace-runtime",
"langfuse_session_id": "thread-1",
}
assert calls["saved_state"]["conv_repo"].bound_agent_configs == [("thread-1", 123)]
assert calls["flushed"] is True
@pytest.mark.asyncio
@ -152,12 +180,20 @@ async def test_agent_chat_sync_returns_finished_even_when_state_has_interrupt(mo
async def fake_get_agent_config_by_id(db, user, agent_config_id):
return SimpleNamespace(agent_id="test-agent", config_json={"context": {}})
async def fake_save_messages_from_langgraph_state(*, agent_instance, thread_id, conv_repo, config_dict):
async def fake_save_messages_from_langgraph_state(*, agent_instance, thread_id, conv_repo, config_dict, trace_info):
return None
async def fake_guard_check(_content):
return False
monkeypatch.setattr(
svc,
"_build_langfuse_run_context",
lambda **kwargs: SimpleNamespace(callbacks=[], metadata={}, tags=[], trace_id=None),
)
monkeypatch.setattr(svc, "get_trace_info", lambda _run_context: {})
monkeypatch.setattr(svc, "flush_langfuse", lambda: None)
monkeypatch.setattr(svc.agent_manager, "get_agent", lambda agent_id: FakeAgent())
monkeypatch.setattr(svc, "get_agent_config_by_id", fake_get_agent_config_by_id)
monkeypatch.setattr(svc, "ConversationRepository", _FakeConvRepo)

110
backend/test/unit/services/test_langfuse_service.py Normal file
View File

@ -0,0 +1,110 @@
from __future__ import annotations
from yuxi.services import langfuse_service as svc
class _FakeLangfuseClient:
def __init__(self, **kwargs):
self.kwargs = kwargs
def create_trace_id(self, *, seed: str | None = None) -> str:
return f"trace-{seed}"
def get_trace_url(self, *, trace_id: str | None = None) -> str | None:
if trace_id is None:
return None
return f"https://langfuse.local/trace/{trace_id}"
def flush(self) -> None:
return None
class _FakeCallbackHandler:
def __init__(self, *, public_key=None, trace_context=None):
self.public_key = public_key
self.trace_context = trace_context
self.last_trace_id = None
def test_build_run_context_includes_trace_metadata(monkeypatch):
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-test")
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-test")
monkeypatch.setenv("LANGFUSE_BASE_URL", "https://cloud.langfuse.example")
monkeypatch.delenv("LANGFUSE_ENABLED", raising=False)
monkeypatch.setattr(svc, "Langfuse", _FakeLangfuseClient)
monkeypatch.setattr(svc, "CallbackHandler", _FakeCallbackHandler)
svc.get_langfuse_client.cache_clear()
run_context = svc.build_run_context(
user_id="user-1",
thread_id="thread-1",
agent_id="agent-a",
request_id="req-1",
operation="agent_chat_stream",
agent_config_id=42,
message_type="text",
username="alice",
login_user_id="alice-login",
department_id=7,
)
assert run_context.trace_id == "trace-req-1"
assert len(run_context.callbacks) == 1
assert run_context.callbacks[0].trace_context == {"trace_id": "trace-req-1"}
assert run_context.metadata["langfuse_user_id"] == "user-1"
assert run_context.metadata["langfuse_session_id"] == "thread-1"
assert run_context.metadata["agent_config_id"] == "42"
assert run_context.metadata["department_id"] == "7"
assert run_context.tags == [
"yuxi",
"chat",
"agent_chat_stream",
"agent:agent-a",
"message_type:text",
]
def test_get_trace_info_prefers_handler_last_trace_id(monkeypatch):
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-test")
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-test")
monkeypatch.setattr(svc, "Langfuse", _FakeLangfuseClient)
monkeypatch.setattr(svc, "CallbackHandler", _FakeCallbackHandler)
svc.get_langfuse_client.cache_clear()
run_context = svc.build_run_context(
user_id="user-1",
thread_id="thread-1",
agent_id="agent-a",
request_id="req-1",
operation="agent_chat_stream",
)
run_context.callbacks[0].last_trace_id = "trace-runtime"
trace_info = svc.get_trace_info(run_context)
assert trace_info == {
"langfuse_trace_id": "trace-runtime",
"langfuse_user_id": "user-1",
"langfuse_session_id": "thread-1",
}
async def test_get_trace_url_async_returns_trace_url(monkeypatch):
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-test")
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-test")
monkeypatch.setattr(svc, "Langfuse", _FakeLangfuseClient)
monkeypatch.setattr(svc, "CallbackHandler", _FakeCallbackHandler)
svc.get_langfuse_client.cache_clear()
run_context = svc.build_run_context(
user_id="user-1",
thread_id="thread-1",
agent_id="agent-a",
request_id="req-1",
operation="agent_chat_stream",
)
run_context.callbacks[0].last_trace_id = "trace-runtime"
trace_url = await svc.get_trace_url_async(run_context)
assert trace_url == "https://langfuse.local/trace/trace-runtime"

4449
backend/uv.lock
View File

File diff suppressed because it is too large Load Diff

2
docker/api.Dockerfile
View File

@ -44,7 +44,7 @@ COPY ../backend/uv.lock /app/uv.lock
# 如果网络还是不好，可以在后面添加 --index-url https://pypi.tuna.tsinghua.edu.cn/simple
RUN --mount=type=cache,target=/root/.cache/uv \
uv sync --no-dev --frozen
uv sync --group test --no-dev --frozen
# 激活虚拟环境并添加到PATH
ENV PATH="/app/.venv/bin:$PATH"

1
docs/.vitepress/config.mts
View File

@ -41,6 +41,7 @@ export default defineConfig({
text: '智能体开发',
items: [
{ text: '智能体配置', link: '/agents/agents-config' },
{ text: 'Langfuse 集成', link: '/agents/langfuse-integration' },
{ text: '工具系统', link: '/agents/tools-system' },
{ text: '中间件', link: '/agents/middleware' },
{ text: '沙盒架构与设计', link: '/agents/sandbox-architecture' },

39
docs/agents/langfuse-integration.md Normal file
View File

@ -0,0 +1,39 @@
# Langfuse 集成
## 为什么 Yuxi 需要 Langfuse
Langfuse 是一套面向大模型应用的可观测性平台，适合用来观察一次智能体执行过程中到底发生了什么。在 Yuxi 里，一轮用户消息通常不会只对应一次简单的模型调用，它往往会伴随 LangGraph 图执行、工具调用、知识库检索以及多轮中间状态切换。仅靠普通后端日志，虽然也能定位问题，但往往需要在多个文件和多个服务日志之间来回跳转，阅读成本高，而且很难从用户、线程和智能体三个维度统一查看。Langfuse 的价值就在于，它把这些原本分散的执行细节收拢到同一条 trace 里，让你能够从一次对话出发，回看模型输入输出、工具链路、耗时和错误位置。
在 Yuxi 当前的实现中，Langfuse 主要承担的是智能体执行观测层，而不是业务主流程的一部分。换句话说，它不会替代模型服务，也不会替代聊天接口本身，而是帮助你在智能体已经能够工作的前提下，看清楚它是如何工作的。对于调试复杂 Agent、排查工具调用失败、评估多轮会话质量以及分析不同智能体的耗时与成本来说，这类观测能力非常关键。尤其是在一个线程里连续发生多轮交互时，Langfuse 可以帮助你把“这轮请求是谁发起的、落在哪个 thread、触发了哪个 agent、调用了哪些模型和工具”这些信息统一串起来。
## 在 Yuxi 中能做什么
Yuxi 对 Langfuse 的映射方式比较直接。一个 Yuxi 用户会映射为 Langfuse 中的 `user_id`，一个对话线程会映射为 `session_id`，而每次用户输入触发的一轮智能体执行会形成一条独立的 trace。这样做的好处是，既能按单轮请求排查问题，也能在同一个线程维度下连续查看多轮会话。对于需要长期分析使用质量、成本和延迟的场景，这种映射方式能够兼顾可读性和后续统计需求。
当 Langfuse 与 Yuxi 连通之后，它最直接的作用是帮助你看清一轮智能体请求内部发生了哪些步骤。你可以看到这一轮调用关联的是哪个用户、哪个线程、哪个智能体，也可以进一步观察模型调用、工具调用和整体耗时表现。对于日常调试来说，它让“问题到底出在模型、工具、配置还是图流程”这件事变得更容易判断。对于长期运行的系统来说，它也为后续做延迟分析、成本分析和用户反馈分析提供了统一的观察入口。
## 如何配置
如果你准备启用 Langfuse，首先需要在 Langfuse Cloud 中创建项目并获取访问凭证。当前版本推荐优先使用云端模式，因为接入成本最低，也更适合先把 tracing 跑通。你需要在运行 Yuxi 的环境中配置 `LANGFUSE_PUBLIC_KEY`、`LANGFUSE_SECRET_KEY` 和 `LANGFUSE_BASE_URL`。其中前两个字段用于鉴权，`LANGFUSE_BASE_URL` 用于指定 Langfuse 服务地址；如果你使用官方云服务，通常可以直接填写 `https://cloud.langfuse.com`。在大多数部署场景下，只要把这些变量写入 `.env` 并通过 Docker Compose 传给 `api` 服务即可生效。
从当前实现来看，Langfuse 只有在 key 配置完整时才会被启用。如果没有配置 `LANGFUSE_PUBLIC_KEY` 或 `LANGFUSE_SECRET_KEY`，Yuxi 会自动退化为“不启用 tracing”的状态，正常聊天功能不会因此中断。这意味着 Langfuse 是一个可选增强项，而不是系统启动的前置依赖。对于希望先验证主流程、后续再逐步补全观测能力的部署者来说，这种行为比较友好，因为它降低了接入门槛，也减少了配置错误对主业务的影响。
## 配置后系统会如何工作
理解 Langfuse 的另一个关键点在于，它并不等于“所有调试信息都会立即显示在界面里”。当前 Yuxi 的设计重点是先把 trace id 与执行上下文稳定关联起来，再把这些信息作为后续调试和分析的基础。也正因为如此，系统会优先保证聊天主链路的稳定性，而不是为了获取额外的可点击 URL 去同步等待 Langfuse 的远程接口。换句话说，Langfuse 在 Yuxi 里首先是观测数据的来源，其次才是一个方便跳转查看的外部页面入口。
这也意味着，Langfuse 接入的目标并不是改变用户聊天体验，而是在不破坏主流程稳定性的前提下，为系统补上一层可观测性。只要配置正确，用户的对话仍然按照原有方式执行，只是在后台额外留下可追踪的执行记录。对于运维和开发来说，这类“尽量不影响主流程”的接入方式更适合在现有系统中逐步落地。
## 如何查看是否生效
启用完成后，你最常见的查看方式是在 Langfuse 控制台中按项目查看 traces。进入项目后，可以按照用户、线程、Agent 或时间范围来筛选，定位到某一轮具体的请求。打开单条 trace 之后，你通常可以看到这一轮智能体执行的整体耗时、模型调用、工具调用以及相关 metadata。对于排查问题来说，这比直接翻阅后端日志更高效，因为你不需要自己手动拼装上下文。对于性能分析来说，也可以更直观地看出某个智能体是否在某类请求上耗时异常，或者某个工具是否经常成为慢点。
如果你是系统管理员或开发者，希望快速确认 Yuxi 是否已经成功把 tracing 打到 Langfuse，最简单的方法不是先看代码，而是先发起一条真实对话，再到 Langfuse 控制台中按最近时间排序查看是否出现新的 trace。如果配置正确，你应该可以看到对应线程下新增的一轮执行记录；如果没有看到，则优先检查 `.env` 中的三个关键变量是否正确传入 `api` 容器，以及容器内依赖是否已经包含 Langfuse SDK。对于基于 Docker Compose 的开发环境，这一步尤其重要，因为仅修改 `pyproject.toml` 并不会自动把新依赖装进已经运行中的镜像，通常还需要重新构建或更新容器。
## 当前建议的接入方式
目前 Yuxi 推荐的接入顺序是先完成 tracing，再逐步扩展到反馈分析或更完整的运营面板。这样做的原因很简单：只有在 trace 关联已经稳定、用户和线程维度映射已经一致的前提下，后续的评分、质量分析和使用统计才会真正可靠。也正因如此，当前文档重点介绍的是 Langfuse 的定位、接入方式和查看路径，而不是一次性覆盖所有更复杂的高级功能。对于大多数项目来说，先把“能看清每轮智能体执行发生了什么”这件事做好，已经能显著改善调试和运维体验。
## 关于后续的 self-host 计划
当前版本优先支持的是 Langfuse Cloud 接入路径。对于有私有化部署、数据合规或内网隔离需求的团队，后续 roadmap 中已经预留了 self-host 模式的支持规划。也就是说，如果你现在先使用云端模式完成接入，未来并不会锁死在这一路径上，后续仍然可以根据部署需求迁移到自托管版本。对大多数团队而言，这是一个更稳妥的推进方式：先用最低成本验证价值，再根据实际使用情况决定是否进入 self-host 阶段。

3
docs/develop-guides/roadmap.md
View File

@ -10,6 +10,7 @@
### 看板
- 集成 LangFuse (观望) 添加用户日志与用户反馈模块，可以在 AgentView 中查看信息
- Langfuse 增加 self-host 模式支持，补齐私有化部署与配置说明
- 部分场景应该使用默认模型作为默认值而不是空值
- 检索测试中，添加问答
- 集成 Memory，基于 deepagents 的文件后端实现
@ -49,6 +50,8 @@
- 新增知识库 PDF、图片的预览功能
- 重构后端测试目录结构：按 `unit / integration / e2e` 分层迁移现有测试，拆分全局 `conftest.py`，统一测试入口为 `uv run --group test pytest`，并新增独立测试规范文档 `docs/vibe/testing-guidelines.md`
- 新增工具元数据 `config_guide` 字段：后端工具列表接口现在可返回“给人看的配置说明”，前端工具详情页会展示该说明，用于提示工具使用前需要配置的环境变量或入口；首批为 MySQL 工具和 `Qwen-Image` 补充了配置指引
- 补充 Langfuse 集成方案文档：明确采用“云端优先、先 tracing 后 feedback”的接入路径，并约定 Yuxi 的 `user/thread` 到 Langfuse `user_id/session_id` 的映射关系
- 新增面向用户的 Langfuse 集成文档：在“智能体开发”分组中说明 Langfuse 的定位、能力、配置方式与查看路径，并与当前 `LANGFUSE_BASE_URL` 配置保持一致
<!-- 添加到这里 -->