当前位置：首页 > news >正文

Langflow 评估与迭代技术深度分析

news 2025/9/1 10:31:15

Langflow 评估与迭代技术深度分析

1. 概述与设计理念

1.1 评估与迭代技术概述

Langflow 的评估与迭代技术是一个多层次、全方位的监控和优化系统，旨在提供：

实时执行监控：通过事件系统跟踪流程执行状态
性能评估：收集和分析执行时间、资源使用等指标
错误追踪：全面的异常捕获和错误分析机制
迭代优化：基于监控数据的自适应调整和持续改进
质量保证：多维度的评估指标和质量控制

1.2 核心设计理念

# 核心设计原则体现在图执行引擎中
class Graph:def __init__(self):# 执行监控self._runs = 0self._updates = 0self._start_time = datetime.now(timezone.utc)self._snapshots: list[dict[str, Any]] = []self._call_order: list[str] = []# 追踪服务self.tracing_service: TracingService | None = get_tracing_service()# 运行管理self.run_manager = RunnableVerticesManager()self.inactivated_vertices: set = set()self.activated_vertices: list[str] = []

设计理念：

可观测性优先：内置全面的监控和追踪机制
渐进式优化：支持增量改进和迭代优化
故障恢复：具备错误处理和状态恢复能力
性能导向：以性能指标驱动优化决策

2. 核心架构和执行监控模型

2.1 执行监控架构

2.2 执行状态模型

# 顶点状态管理
class VertexStates(str, Enum):ACTIVE = "ACTIVE"INACTIVE = "INACTIVE"ERROR = "ERROR"class Vertex:def __init__(self):self.built = Falseself.result: ResultData | None = Noneself.artifacts: dict[str, Any] = {}self.logs: dict[str, list[Log]] = {}self.state = VertexStates.ACTIVEasync def build(self, event_manager: EventManager = None):"""构建顶点并记录执行指标"""start_time = time.time()try:# 执行构建逻辑result = await self._build_implementation()# 记录成功指标self.built = Trueself.result = resultexcept Exception as e:# 记录错误指标self.state = VertexStates.ERRORawait self._log_error(e)raisefinally:# 记录执行时间execution_time = time.time() - start_timeawait self._record_metrics(execution_time)

3. 流程执行引擎分析

3.1 异步执行管理

# 图执行的核心流程监控
class Graph:async def process(self, *, fallback_to_env_vars: bool, start_component_id: str | None = None,event_manager: EventManager | None = None) -> Graph:"""处理图执行并进行全面监控"""# 初始化运行追踪await self.initialize_run()# 执行层级处理vertex_task_run_count: dict[str, int] = {}to_process = deque(first_layer)layer_index = 0while to_process:current_batch = list(to_process)to_process.clear()# 创建并发任务tasks = []for vertex_id in current_batch:vertex = self.get_vertex(vertex_id)task = asyncio.create_task(self.build_vertex(vertex_id=vertex_id,event_manager=event_manager,),name=f"{vertex.id} Run {vertex_task_run_count.get(vertex_id, 0)}")tasks.append(task)vertex_task_run_count[vertex_id] = vertex_task_run_count.get(vertex_id, 0) + 1logger.debug(f"Running layer {layer_index} with {len(tasks)} tasks")try:# 执行任务并收集结果next_runnable_vertices = await self._execute_tasks(tasks, lock=lock, has_webhook_component=has_webhook_component)except Exception:logger.exception(f"Error executing tasks in layer {layer_index}")raise# 准备下一层执行if next_runnable_vertices:to_process.extend(next_runnable_vertices)layer_index += 1return self

3.2 任务执行监控

async def _execute_tasks(self, tasks: list[asyncio.Task], lock: asyncio.Lock, *, has_webhook_component: bool = False) -> list[str]:"""执行任务并进行异常处理"""results = []completed_tasks = await asyncio.gather(*tasks, return_exceptions=True)vertices: list[Vertex] = []for i, result in enumerate(completed_tasks):task_name = tasks[i].get_name()vertex_id = tasks[i].get_name().split(" ")[0]if isinstance(result, Exception):logger.error(f"Task {task_name} failed with exception: {result}")# 记录异常到监控系统if has_webhook_component:await self._log_vertex_build_from_exception(vertex_id, result)# 取消剩余任务for t in tasks[i + 1:]:t.cancel()raise result# 记录成功执行if isinstance(result, VertexBuildResult):if self.flow_id is not None:await log_vertex_build(flow_id=self.flow_id,vertex_id=result.vertex.id,valid=result.valid,params=result.params,data=result.result_dict,artifacts=result.artifacts,)vertices.append(result.vertex)return results

4. 性能监控和指标收集

4.1 性能指标模型

# 顶点构建结果记录
class VertexBuildBase(SQLModel):timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))id: str = Field(nullable=False)data: dict | None = Field(default=None)artifacts: dict | None = Field(default=None)params: str | None = Field(default=None)valid: bool = Field(nullable=False)flow_id: UUID = Field()# 多租户支持job_id: UUID | None = Field(default=None)tenant_id: UUID | None = Field(default=None)organization_id: UUID | None = Field(default=None)business_domain_id: UUID | None = Field(default=None)@field_serializer("data")def serialize_data(self, data) -> dict:"""序列化数据并应用大小限制"""return serialize(data, max_length=get_max_text_length(), max_items=get_max_items_length())

4.2 事务日志记录

async def log_transaction(flow_id: str | UUID, source: Vertex, status, target: Vertex | None = None, error=None) -> None:"""异步记录事务日志用于性能分析"""try:if not get_settings_service().settings.transactions_storage_enabled:return# 清理敏感参数inputs = _vertex_to_primitive_dict(source)# 序列化结果数据if source.result:try:result_dict = source.result.model_dump()for key, value in result_dict.items():if isinstance(value, pd.DataFrame):result_dict[key] = value.to_dict()outputs = result_dictexcept Exception as e:logger.warning(f"Error serializing result: {e!s}")outputs = Noneelse:outputs = None# 创建事务记录transaction = TransactionBase(vertex_id=source.id,inputs=inputs,outputs=outputs,status=status,error=str(error) if error else None,flow_id=str(flow_id),timestamp=datetime.now(timezone.utc))# 异步存储await crud_log_transaction(transaction)except Exception as e:logger.error(f"Error logging transaction: {e}")

5. 事件系统和状态跟踪

5.1 事件管理器架构

class EventManager:def __init__(self, queue: asyncio.Queue, job_id: str | None = None):self.queue = queueself.events: dict[str, PartialEventCallback] = {}self.job_id = job_iddef register_event(self, name: str, event_type: str, callback: EventCallback | None = None) -> None:"""注册事件处理器"""if not name.startswith("on_"):msg = "Event name must start with 'on_'"raise ValueError(msg)if callback is None:callback_ = partial(self.send_event, event_type=event_type)else:callback_ = partial(callback, manager=self, event_type=event_type)self.events[name] = callback_def send_event(self, *, event_type: str, data: LoggableType):"""发送事件到监控系统"""try:# 创建标准化事件if isinstance(data, dict) and event_type in {"message", "error", "warning", "info", "token"}:data = create_event_by_type(event_type, **data)except TypeError as e:logger.debug(f"Error creating playground event: {e}")except Exception:raise# 序列化事件数据jsonable_data = jsonable_encoder(data)json_data = {"event": event_type, "data": jsonable_data}event_id = f"{event_type}-{uuid.uuid4()}"str_data = json.dumps(json_data) + "\n\n"# 更新作业状态from langflow.services.apa.resource.job.job_service import JobServicerun_until_complete_safe(JobService.update_status_by_event(self.job_id, event_type, data))# 加入事件队列self.queue.put_nowait((event_id, str_data.encode("utf-8"), time.time()))

5.2 默认事件管理器

def create_default_event_manager(queue):"""创建默认事件管理器"""manager = EventManager(queue)# 注册标准事件manager.register_event("on_token", "token")manager.register_event("on_vertices_sorted", "vertices_sorted")manager.register_event("on_error", "error")manager.register_event("on_end", "end")manager.register_event("on_message", "add_message")manager.register_event("on_remove_message", "remove_message")manager.register_event("on_end_vertex", "end_vertex")manager.register_event("on_build_start", "build_start")manager.register_event("on_build_end", "build_end")return manager

6. 日志系统和调试支持

6.1 结构化日志系统

class SizedLogBuffer:def __init__(self, max_readers: int = 20):"""日志缓冲区用于实时日志检索"""self.buffer: deque = deque()self._max_readers = max_readersself._wlock = Lock()self._rsemaphore = Semaphore(max_readers)self._max = 0def write(self, message: str) -> None:"""写入日志消息"""record = json.loads(message)log_entry = record["text"]epoch = int(record["record"]["time"]["timestamp"] * 1000)with self._wlock:if len(self.buffer) >= self.max:for _ in range(len(self.buffer) - self.max + 1):self.buffer.popleft()self.buffer.append((epoch, log_entry))def get_after_timestamp(self, timestamp: int, lines: int = 5) -> dict[int, str]:"""获取指定时间戳之后的日志"""rc = {}self._rsemaphore.acquire()try:with self._wlock:for ts, msg in self.buffer:if lines == 0:breakif ts >= timestamp and lines > 0:rc[ts] = msglines -= 1finally:self._rsemaphore.release()return rc

6.2 日志配置和管理

def configure(*, log_level: str | None = None,log_file: Path | None = None,disable: bool | None = False,log_env: str | None = None,log_format: str | None = None,async_file: bool = False) -> None:"""配置日志系统"""# 移除默认处理器logger.remove()logger.patch(patching)# 容器环境配置if log_env.lower() == "container" or log_env.lower() == "container_json":logger.add(sys.stdout, format="{message}", serialize=True)elif log_env.lower() == "container_csv":logger.add(sys.stdout, format="{time:YYYY-MM-DD HH:mm:ss.SSS} {level} {file} {line} {function} {message}")else:# 开发环境配置if log_format is None or not is_valid_log_format(log_format):log_format = DEFAULT_LOG_FORMAT# 美化输出支持log_stdout_pretty = os.getenv("LANGFLOW_PRETTY_LOGS", "false").lower() == "true"if log_stdout_pretty:logger.configure(handlers=[{"sink": RichHandler(rich_tracebacks=True, markup=True),"format": log_format,"level": log_level.upper(),}])else:logger.add(sys.stdout, level=log_level.upper(), format=log_format, backtrace=True, diagnose=True)# 文件日志配置if not log_file:cache_dir = Path(user_cache_dir("langflow"))log_file = cache_dir / "langflow.log"try:logger.add(sink=AsyncFileSink(log_file) if async_file else log_file,level=log_level.upper(),format=log_format,serialize=True,)except Exception:logger.exception("Error setting up log file")# 启用日志缓冲区if log_buffer.enabled():logger.add(sink=log_buffer.write, format="{time} {level} {message}", serialize=True)

7. 错误处理和异常分析

7.1 组件构建错误处理

async def _log_vertex_build_from_exception(self, vertex_id: str, result: Exception) -> None:"""记录顶点构建异常的详细信息"""if isinstance(result, ComponentBuildError):params = result.messagetb = result.formatted_tracebackelse:from langflow.api.utils import format_exception_messagetb = traceback.format_exc()logger.exception("Error building Component")params = format_exception_message(result)# 构建错误消息message = {"errorMessage": params, "stackTrace": tb}vertex = self.get_vertex(vertex_id)output_label = vertex.outputs[0]["name"] if vertex.outputs else "output"outputs = {output_label: OutputValue(message=message, type="error")}# 创建错误响应result_data_response = {"results": {},"outputs": outputs,"logs": {},"message": {},"artifacts": {},"timedelta": None,"duration": None,"used_frozen_result": False,}# 记录到监控系统await log_vertex_build(flow_id=self.flow_id or "",vertex_id=vertex_id or "errors",valid=False,params=params,data=result_data_response,artifacts={},)

7.2 异常拦截器

class InterceptHandler(logging.Handler):"""标准日志到 Loguru 的拦截器"""@overridedef emit(self, record) -> None:# 获取对应的 Loguru 级别try:level = logger.level(record.levelname).nameexcept ValueError:level = record.levelno# 查找调用者信息frame, depth = logging.currentframe(), 2while frame.f_code.co_filename == logging.__file__ and frame.f_back:frame = frame.f_backdepth += 1logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())

8. 迭代优化和自适应机制

8.1 图状态快照

class Graph:def _snapshot(self):"""创建图状态快照用于回滚和分析"""return {"_run_queue": self._run_queue.copy(),"_first_layer": self._first_layer.copy(),"vertices_layers": copy.deepcopy(self.vertices_layers),"vertices_to_run": copy.deepcopy(self.vertices_to_run),"run_manager": copy.deepcopy(self.run_manager.to_dict()),}def _record_snapshot(self, vertex_id: str | None = None) -> None:"""记录执行快照"""self._snapshots.append(self.get_snapshot())if vertex_id:self._call_order.append(vertex_id)def get_snapshot(self):"""获取当前状态快照"""return copy.deepcopy({"run_manager": self.run_manager.to_dict(),"run_queue": self._run_queue,"vertices_layers": self.vertices_layers,"first_layer": self.first_layer,"inactive_vertices": self.inactive_vertices,"activated_vertices": self.activated_vertices,})

8.2 自适应顶点管理

def activate_state_vertices(self, name: str, caller: str) -> None:"""激活与给定状态名称关联的顶点"""vertices_ids = set()new_predecessor_map = {}activated_vertices = []for vertex_id in self.is_state_vertices:caller_vertex = self.get_vertex(caller)vertex = self.get_vertex(vertex_id)if vertex_id == caller or vertex.display_name == caller_vertex.display_name:continuectx_key = vertex.raw_params.get("context_key")if isinstance(ctx_key, str) and name in ctx_key and vertex_id != caller:if isinstance(vertex, StateVertex):activated_vertices.append(vertex_id)vertices_ids.add(vertex_id)# 获取所有后继顶点successors = self.get_all_successors(vertex, flat=True)# 更新运行前驱映射successors_predecessors = set()for successor in successors:successors_predecessors.update(self.get_all_predecessors(successor))# 构建边集合edges_set = set()for _vertex in [vertex, *successors, *successors_predecessors]:edges_set.update(_vertex.edges)if _vertex.state == VertexStates.INACTIVE:_vertex.set_state("ACTIVE")vertices_ids.add(_vertex.id)# 重建前驱映射edges = list(edges_set)predecessor_map, _ = self.build_adjacency_maps(edges)new_predecessor_map.update(predecessor_map)# 更新运行状态vertices_ids.update(new_predecessor_map.keys())vertices_ids.update(v_id for value_list in new_predecessor_map.values() for v_id in value_list)self.activated_vertices = activated_verticesself.vertices_to_run.update(vertices_ids)self.run_manager.update_run_state(run_predecessors=new_predecessor_map,vertices_to_run=self.vertices_to_run,)

9. 评估指标和质量保证

9.1 分布式追踪系统

class TracingService(Service):"""分布式追踪服务"""async def start_tracers(self, run_id: UUID, run_name: str, user_id: str | None, session_id: str | None,project_name: str | None = None) -> None:"""启动图运行的追踪"""if self.deactivated:returntry:project_name = project_name or os.getenv("LANGCHAIN_PROJECT", "Langflow")trace_context = TraceContext(run_id, run_name, project_name, user_id, session_id)trace_context_var.set(trace_context)# 启动工作线程await self._start(trace_context)# 初始化各种追踪器self._initialize_langsmith_tracer(trace_context)self._initialize_langwatch_tracer(trace_context)self._initialize_langfuse_tracer(trace_context)self._initialize_arize_phoenix_tracer(trace_context)self._initialize_opik_tracer(trace_context)except Exception as e:logger.debug(f"Error initializing tracers: {e}")@asynccontextmanagerasync def trace_component(self, component: Component, trace_name: str,inputs: dict[str, Any], metadata: dict[str, Any] | None = None):"""追踪组件执行"""if self.deactivated:yield selfreturntrace_id = trace_nameif component._vertex:trace_id = component._vertex.idtrace_type = component.trace_typecomponent_trace_context = ComponentTraceContext(trace_id, trace_name, trace_type, component._vertex, inputs, metadata)component_context_var.set(component_trace_context)trace_context = trace_context_var.get()if trace_context is None:msg = "called trace_component but no trace context found"raise RuntimeError(msg)trace_context.all_inputs[trace_name] |= inputs or {}# 开始组件追踪await trace_context.traces_queue.put((self._start_component_traces, (component_trace_context, trace_context)))try:yield selfexcept Exception as e:# 记录异常追踪await trace_context.traces_queue.put((self._end_component_traces, (component_trace_context, trace_context, e)))raiseelse:# 正常结束追踪await trace_context.traces_queue.put((self._end_component_traces, (component_trace_context, trace_context, None)))

9.2 遥测数据收集

class TelemetryService(Service):"""遥测服务用于性能指标收集"""def __init__(self, settings_service: SettingsService):super().__init__()self.settings_service = settings_serviceself.base_url = settings_service.settings.telemetry_base_urlself.telemetry_queue: asyncio.Queue = asyncio.Queue()self.client = httpx.AsyncClient(timeout=10.0)self.running = Falseself._stopping = False# OpenTelemetry 集成self.ot = OpenTelemetry(prometheus_enabled=settings_service.settings.prometheus_enabled)# 隐私保护self.do_not_track = (os.getenv("DO_NOT_TRACK", "False").lower() == "true" or settings_service.settings.do_not_track)async def send_telemetry_data(self, payload: BaseModel, path: str | None = None) -> None:"""发送遥测数据"""if self.do_not_track:logger.debug("Telemetry tracking is disabled.")returnurl = f"{self.base_url}"if path:url = f"{url}/{path}"try:payload_dict = payload.model_dump(by_alias=True, exclude_none=True, exclude_unset=True)response = await self.client.get(url, params=payload_dict)if response.status_code != httpx.codes.OK:logger.error(f"Failed to send telemetry data: {response.status_code} {response.text}")else:logger.debug("Telemetry data sent successfully.")except Exception:logger.error("Error sending telemetry data")async def log_package_version(self) -> None:"""记录包版本信息"""python_version = ".".join(platform.python_version().split(".")[:2])version_info = get_version_info()if self.architecture is None:self.architecture = (await asyncio.to_thread(platform.architecture))[0]payload = VersionPayload(package=version_info["package"].lower(),version=version_info["version"],platform=platform.platform(),python=python_version,cache_type=self.settings_service.settings.cache_type,backend_only=self.settings_service.settings.backend_only,arch=self.architecture,auto_login=self.settings_service.auth_settings.AUTO_LOGIN,desktop=self._get_langflow_desktop(),)await self._queue_event((self.send_telemetry_data, payload, None))

10. 监控仪表板和可视化

10.1 实时日志流

@log_router.get("/logs-stream")
async def stream_logs(request: Request):"""HTTP/2 服务器发送事件 (SSE) 端点用于流式日志"""global log_bufferif log_buffer.enabled() is False:raise HTTPException(status_code=HTTPStatus.NOT_IMPLEMENTED,detail="Log retrieval is disabled",)return StreamingResponse(event_generator(request), media_type="text/event-stream")async def event_generator(request: Request):"""事件生成器用于实时日志流"""global log_bufferlast_read_item = Nonecurrent_not_sent = 0while not await request.is_disconnected():to_write: list[Any] = []with log_buffer.get_write_lock():if last_read_item is None:last_read_item = log_buffer.buffer[len(log_buffer.buffer) - 1]else:found_last = Falsefor item in log_buffer.buffer:if found_last:to_write.append(item)last_read_item = itemcontinueif item is last_read_item:found_last = Truecontinue# 处理缓冲区溢出情况if not found_last:for item in log_buffer.buffer:to_write.append(item)last_read_item = itemif to_write:for ts, msg in to_write:yield f"{json.dumps({ts: msg})}\n\n"else:current_not_sent += 1if current_not_sent == NUMBER_OF_NOT_SENT_BEFORE_KEEPALIVE:current_not_sent = 0yield "keepalive\n\n"await asyncio.sleep(1)

10.2 日志查询接口

@log_router.get("/logs")
async def logs(lines_before: Annotated[int, Query(description="时间戳之前的日志行数")] = 0,lines_after: Annotated[int, Query(description="时间戳之后的日志行数")] = 0,timestamp: Annotated[int, Query(description="开始获取日志的时间戳")] = 0,
):"""获取历史日志数据"""global log_bufferif log_buffer.enabled() is False:raise HTTPException(status_code=HTTPStatus.NOT_IMPLEMENTED,detail="Log retrieval is disabled",)if lines_after > 0 and lines_before > 0:raise HTTPException(status_code=HTTPStatus.BAD_REQUEST,detail="Cannot request logs before and after the timestamp",)if timestamp <= 0:if lines_after > 0:raise HTTPException(status_code=HTTPStatus.BAD_REQUEST,detail="Timestamp is required when requesting logs after the timestamp",)content = log_buffer.get_last_n(10) if lines_before <= 0 else log_buffer.get_last_n(lines_before)elif lines_before > 0:content = log_buffer.get_before_timestamp(timestamp=timestamp, lines=lines_before)elif lines_after > 0:content = log_buffer.get_after_timestamp(timestamp=timestamp, lines=lines_after)else:content = log_buffer.get_before_timestamp(timestamp=timestamp, lines=10)return JSONResponse(content=content)

11. 调试工具和开发支持

11.1 步进执行支持

class Graph:async def astep(self, inputs: InputValueRequest | None = None,files: list[str] | None = None,user_id: str | None = None,event_manager: EventManager | None = None):"""异步步进执行图中的下一个顶点"""if not self._prepared:msg = "Graph not prepared. Call prepare() first."raise ValueError(msg)if not self._run_queue:self._end_all_traces_async()return Finish()# 获取下一个要执行的顶点vertex_id = self.get_next_in_queue()chat_service = get_chat_service()# 构建顶点vertex_build_result = await self.build_vertex(vertex_id=vertex_id,user_id=user_id,inputs_dict=inputs.model_dump() if inputs else {},files=files,get_cache=chat_service.get_cache,set_cache=chat_service.set_cache,event_manager=event_manager,)# 获取下一批可运行顶点next_runnable_vertices = await self.get_next_runnable_vertices(self._lock, vertex=vertex_build_result.vertex, cache=False)if self.stop_vertex and self.stop_vertex in next_runnable_vertices:next_runnable_vertices = [self.stop_vertex]self.extend_run_queue(next_runnable_vertices)self.reset_inactivated_vertices()self.reset_activated_vertices()# 更新缓存await chat_service.set_cache(str(self.flow_id or self._run_id), self)self._record_snapshot(vertex_id)return vertex_build_resultdef step(self, inputs: InputValueRequest | None = None,files: list[str] | None = None,user_id: str | None = None):"""同步步进执行包装器"""return run_until_complete(self.astep(inputs, files, user_id))

11.2 调试信息收集

def get_artifact_type(value, build_result) -> str:"""确定构件类型用于调试显示"""result = ArtifactType.UNKNOWNmatch value:case Data():result = ArtifactType.RECORDcase str():result = ArtifactType.TEXTcase dict():result = ArtifactType.OBJECTcase list():result = ArtifactType.ARRAYcase Message():result = ArtifactType.MESSAGEif result == ArtifactType.UNKNOWN and (isinstance(build_result, Generator) or (isinstance(value, Message) and isinstance(value.text, Generator))):result = ArtifactType.STREAMreturn result.valuedef post_process_raw(raw, artifact_type: str):"""后处理原始数据用于调试显示"""if artifact_type == ArtifactType.STREAM.value:raw = ""return raw

12. 持续改进和反馈循环

12.1 性能优化机制

class Graph:def sort_by_avg_build_time(self, vertices_layers: list[list[str]]) -> list[list[str]]:"""根据平均构建时间对顶点进行排序"""def sort_layer_by_avg_build_time(vertices_ids: list[str]) -> list[str]:if len(vertices_ids) == 1:return vertices_ids# 按平均构建时间排序vertices_ids.sort(key=lambda vertex_id: self.get_vertex(vertex_id).avg_build_time)return vertices_idsreturn [sort_layer_by_avg_build_time(layer) for layer in vertices_layers]@staticmethoddef sort_interface_components_first(vertices_layers: list[list[str]]) -> list[list[str]]:"""优先排序接口组件"""def contains_interface_component(vertex):return any(component.value in vertex for component in InterfaceComponentTypes)return [sorted(inner_list,key=lambda vertex: not contains_interface_component(vertex),)for inner_list in vertices_layers]

12.2 自适应缓存策略

def _set_cache_to_vertices_in_cycle(self) -> None:"""为循环中的顶点设置缓存策略"""edges = self._get_edges_as_list_of_tuples()cycle_vertices = set(find_cycle_vertices(edges))for vertex in self.vertices:if vertex.id in cycle_vertices:# 禁用循环顶点的缓存以确保正确执行vertex.apply_on_outputs(lambda output_object: setattr(output_object, "cache", False))def _set_cache_if_listen_notify_components(self) -> None:"""如果存在 Listen/Notify 组件则禁用所有顶点的缓存"""has_listen_or_notify_component = any(vertex.id.split("-")[0] in {"Listen", "Notify"} for vertex in self.vertices)if has_listen_or_notify_component:for vertex in self.vertices:vertex.apply_on_outputs(lambda output_object: setattr(output_object, "cache", False))

13. 应用示例

13.1 基础监控示例

# 示例1: 基础流程监控
import asyncio
from langflow.graph.graph.base import Graph
from langflow.events.event_manager import create_default_event_managerasync def basic_monitoring_example():"""基础监控示例"""# 创建事件队列和管理器event_queue = asyncio.Queue()event_manager = create_default_event_manager(event_queue)# 创建图实例graph = Graph(flow_id="monitoring_example", flow_name="Basic Monitoring")# 注册自定义事件处理器def custom_error_handler(*, manager, event_type, data):print(f"Custom Error Handler: {event_type} - {data}")# 可以在这里实现自定义错误处理逻辑event_manager.register_event("on_custom_error", "custom_error", custom_error_handler)# 准备图执行graph.prepare()# 执行图并监控try:async for result in graph.async_start(inputs=[{"input": "Hello World"}],event_manager=event_manager):print(f"Step Result: {result}")# 检查事件队列while not event_queue.empty():event_id, event_data, timestamp = event_queue.get_nowait()print(f"Event: {event_id} at {timestamp}")except Exception as e:# 发送自定义错误事件event_manager.on_custom_error(data={"error": str(e), "context": "graph_execution"})print("Monitoring example completed")# 运行示例
# asyncio.run(basic_monitoring_example())

13.2 高级性能分析示例

# 示例2: 高级性能分析和优化
import time
from datetime import datetime, timezone
from langflow.services.tracing.service import TracingService
from langflow.services.telemetry.service import TelemetryServiceclass PerformanceAnalyzer:"""性能分析器"""def __init__(self):self.metrics = {}self.start_times = {}self.execution_history = []def start_timing(self, operation_id: str):"""开始计时"""self.start_times[operation_id] = time.time()def end_timing(self, operation_id: str):"""结束计时并记录"""if operation_id in self.start_times:duration = time.time() - self.start_times[operation_id]self.metrics[operation_id] = durationself.execution_history.append({"operation": operation_id,"duration": duration,"timestamp": datetime.now(timezone.utc)})del self.start_times[operation_id]return durationreturn Nonedef get_performance_report(self):"""生成性能报告"""if not self.execution_history:return "No performance data available"total_operations = len(self.execution_history)total_time = sum(item["duration"] for item in self.execution_history)avg_time = total_time / total_operations# 找出最慢的操作slowest = max(self.execution_history, key=lambda x: x["duration"])fastest = min(self.execution_history, key=lambda x: x["duration"])report = f"""
Performance Analysis Report
==========================
Total Operations: {total_operations}
Total Execution Time: {total_time:.4f}s
Average Time per Operation: {avg_time:.4f}sSlowest Operation: {slowest['operation']} ({slowest['duration']:.4f}s)
Fastest Operation: {fastest['operation']} ({fastest['duration']:.4f}s)Recent Operations:
"""# 显示最近的10个操作for item in self.execution_history[-10:]:report += f"  {item['operation']}: {item['duration']:.4f}s at {item['timestamp']}\n"return reportasync def advanced_performance_example():"""高级性能分析示例"""analyzer = PerformanceAnalyzer()# 模拟图执行过程class MockGraph:def __init__(self):self.vertices = ["input", "processor", "llm", "output"]async def execute_with_monitoring(self, analyzer: PerformanceAnalyzer):"""执行图并进行性能监控"""for vertex in self.vertices:analyzer.start_timing(f"vertex_{vertex}")# 模拟不同顶点的执行时间if vertex == "llm":await asyncio.sleep(0.5)  # LLM 调用较慢elif vertex == "processor":await asyncio.sleep(0.2)  # 处理器中等速度else:await asyncio.sleep(0.1)  # 输入输出较快duration = analyzer.end_timing(f"vertex_{vertex}")print(f"Vertex {vertex} completed in {duration:.4f}s")# 检查是否需要优化if duration > 0.3:print(f"⚠️  Warning: {vertex} is running slowly ({duration:.4f}s)")# 创建模拟图并执行mock_graph = MockGraph()print("Starting performance analysis...")analyzer.start_timing("total_execution")await mock_graph.execute_with_monitoring(analyzer)total_duration = analyzer.end_timing("total_execution")print(f"\nTotal execution completed in {total_duration:.4f}s")# 生成性能报告print(analyzer.get_performance_report())# 基于性能数据提供优化建议print("\nOptimization Recommendations:")for item in analyzer.execution_history:if item["duration"] > 0.3:print(f"  - Consider optimizing {item['operation']} (current: {item['duration']:.4f}s)")elif item["duration"] < 0.15:print(f"  - {item['operation']} is performing well ({item['duration']:.4f}s)")# 运行高级示例
# asyncio.run(advanced_performance_example())