youngkingdom/dify
1
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-05-03 17:08:03 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into feat/rag-2

Browse Source 
# Conflicts:
#	web/app/components/workflow/hooks/use-workflow.ts
This commit is contained in:
jyong
2025-07-31 10:30:28 +08:00
parent 1b6a925b34 1b2046da3f
commit a7a4c8228e
151 changed files with 2950 additions and 920 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
api/core/app/app_config/entities.py
View File

@ -159,6 +159,8 @@ SupportedComparisonOperator = Literal[
"is not",
"empty",
"not empty",
"in",
"not in",
# for number
"=",
"≠",

28
api/core/ops/aliyun_trace/aliyun_trace.py
View File

@ -10,6 +10,7 @@ from sqlalchemy.orm import Session, sessionmaker
from core.ops.aliyun_trace.data_exporter.traceclient import (
TraceClient,
convert_datetime_to_nanoseconds,
convert_string_to_id,
convert_to_span_id,
convert_to_trace_id,
generate_span_id,
@ -101,8 +102,9 @@ class AliyunDataTrace(BaseTraceInstance):
raise ValueError(f"Aliyun get run url failed: {str(e)}")
def workflow_trace(self, trace_info: WorkflowTraceInfo):
external_trace_id = trace_info.metadata.get("external_trace_id")
trace_id = external_trace_id or convert_to_trace_id(trace_info.workflow_run_id)
trace_id = convert_to_trace_id(trace_info.workflow_run_id)
if trace_info.trace_id:
trace_id = convert_string_to_id(trace_info.trace_id)
workflow_span_id = convert_to_span_id(trace_info.workflow_run_id, "workflow")
self.add_workflow_span(trace_id, workflow_span_id, trace_info)
@ -130,6 +132,9 @@ class AliyunDataTrace(BaseTraceInstance):
status = Status(StatusCode.ERROR, trace_info.error)
trace_id = convert_to_trace_id(message_id)
if trace_info.trace_id:
trace_id = convert_string_to_id(trace_info.trace_id)
message_span_id = convert_to_span_id(message_id, "message")
message_span = SpanData(
trace_id=trace_id,
@ -186,9 +191,13 @@ class AliyunDataTrace(BaseTraceInstance):
return
message_id = trace_info.message_id
trace_id = convert_to_trace_id(message_id)
if trace_info.trace_id:
trace_id = convert_string_to_id(trace_info.trace_id)
documents_data = extract_retrieval_documents(trace_info.documents)
dataset_retrieval_span = SpanData(
trace_id=convert_to_trace_id(message_id),
trace_id=trace_id,
parent_span_id=convert_to_span_id(message_id, "message"),
span_id=generate_span_id(),
name="dataset_retrieval",
@ -214,8 +223,12 @@ class AliyunDataTrace(BaseTraceInstance):
if trace_info.error:
status = Status(StatusCode.ERROR, trace_info.error)
trace_id = convert_to_trace_id(message_id)
if trace_info.trace_id:
trace_id = convert_string_to_id(trace_info.trace_id)
tool_span = SpanData(
trace_id=convert_to_trace_id(message_id),
trace_id=trace_id,
parent_span_id=convert_to_span_id(message_id, "message"),
span_id=generate_span_id(),
name=trace_info.tool_name,
@ -451,8 +464,13 @@ class AliyunDataTrace(BaseTraceInstance):
status: Status = Status(StatusCode.OK)
if trace_info.error:
status = Status(StatusCode.ERROR, trace_info.error)
trace_id = convert_to_trace_id(message_id)
if trace_info.trace_id:
trace_id = convert_string_to_id(trace_info.trace_id)
suggested_question_span = SpanData(
trace_id=convert_to_trace_id(message_id),
trace_id=trace_id,
parent_span_id=convert_to_span_id(message_id, "message"),
span_id=convert_to_span_id(message_id, "suggested_question"),
name="suggested_question",

12
api/core/ops/aliyun_trace/data_exporter/traceclient.py
View File

@ -181,15 +181,21 @@ def convert_to_trace_id(uuid_v4: Optional[str]) -> int:
raise ValueError(f"Invalid UUID input: {e}")
def convert_string_to_id(string: Optional[str]) -> int:
if not string:
return generate_span_id()
hash_bytes = hashlib.sha256(string.encode("utf-8")).digest()
id = int.from_bytes(hash_bytes[:8], byteorder="big", signed=False)
return id
def convert_to_span_id(uuid_v4: Optional[str], span_type: str) -> int:
try:
uuid_obj = uuid.UUID(uuid_v4)
except Exception as e:
raise ValueError(f"Invalid UUID input: {e}")
combined_key = f"{uuid_obj.hex}-{span_type}"
hash_bytes = hashlib.sha256(combined_key.encode("utf-8")).digest()
span_id = int.from_bytes(hash_bytes[:8], byteorder="big", signed=False)
return span_id
return convert_string_to_id(combined_key)
def convert_datetime_to_nanoseconds(start_time_a: Optional[datetime]) -> Optional[int]:

39
api/core/ops/arize_phoenix_trace/arize_phoenix_trace.py
View File

@ -4,6 +4,7 @@ import logging
import os
from datetime import datetime, timedelta
from typing import Any, Optional, Union, cast
from urllib.parse import urlparse
from openinference.semconv.trace import OpenInferenceSpanKindValues, SpanAttributes
from opentelemetry import trace
@ -40,8 +41,14 @@ def setup_tracer(arize_phoenix_config: ArizeConfig | PhoenixConfig) -> tuple[tra
try:
# Choose the appropriate exporter based on config type
exporter: Union[GrpcOTLPSpanExporter, HttpOTLPSpanExporter]
# Inspect the provided endpoint to determine its structure
parsed = urlparse(arize_phoenix_config.endpoint)
base_endpoint = f"{parsed.scheme}://{parsed.netloc}"
path = parsed.path.rstrip("/")
if isinstance(arize_phoenix_config, ArizeConfig):
arize_endpoint = f"{arize_phoenix_config.endpoint}/v1"
arize_endpoint = f"{base_endpoint}/v1"
arize_headers = {
"api_key": arize_phoenix_config.api_key or "",
"space_id": arize_phoenix_config.space_id or "",
@ -53,7 +60,7 @@ def setup_tracer(arize_phoenix_config: ArizeConfig | PhoenixConfig) -> tuple[tra
timeout=30,
)
else:
phoenix_endpoint = f"{arize_phoenix_config.endpoint}/v1/traces"
phoenix_endpoint = f"{base_endpoint}{path}/v1/traces"
phoenix_headers = {
"api_key": arize_phoenix_config.api_key or "",
"authorization": f"Bearer {arize_phoenix_config.api_key or ''}",
@ -91,16 +98,21 @@ def datetime_to_nanos(dt: Optional[datetime]) -> int:
return int(dt.timestamp() * 1_000_000_000)
def uuid_to_trace_id(string: Optional[str]) -> int:
"""Convert UUID string to a valid trace ID (16-byte integer)."""
def string_to_trace_id128(string: Optional[str]) -> int:
"""
Convert any input string into a stable 128-bit integer trace ID.
This uses SHA-256 hashing and takes the first 16 bytes (128 bits) of the digest.
It's suitable for generating consistent, unique identifiers from strings.
"""
if string is None:
string = ""
hash_object = hashlib.sha256(string.encode())
# Take the first 16 bytes (128 bits) of the hash
# Take the first 16 bytes (128 bits) of the hash digest
digest = hash_object.digest()[:16]
# Convert to integer (128 bits)
# Convert to a 128-bit integer
return int.from_bytes(digest, byteorder="big")
@ -153,8 +165,7 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
}
workflow_metadata.update(trace_info.metadata)
external_trace_id = trace_info.metadata.get("external_trace_id")
trace_id = external_trace_id or uuid_to_trace_id(trace_info.workflow_run_id)
trace_id = string_to_trace_id128(trace_info.trace_id or trace_info.workflow_run_id)
span_id = RandomIdGenerator().generate_span_id()
context = SpanContext(
trace_id=trace_id,
@ -310,7 +321,7 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
SpanAttributes.SESSION_ID: trace_info.message_data.conversation_id,
}
trace_id = uuid_to_trace_id(trace_info.message_id)
trace_id = string_to_trace_id128(trace_info.trace_id or trace_info.message_id)
message_span_id = RandomIdGenerator().generate_span_id()
span_context = SpanContext(
trace_id=trace_id,
@ -406,7 +417,7 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
}
metadata.update(trace_info.metadata)
trace_id = uuid_to_trace_id(trace_info.message_id)
trace_id = string_to_trace_id128(trace_info.message_id)
span_id = RandomIdGenerator().generate_span_id()
context = SpanContext(
trace_id=trace_id,
@ -468,7 +479,7 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
}
metadata.update(trace_info.metadata)
trace_id = uuid_to_trace_id(trace_info.message_id)
trace_id = string_to_trace_id128(trace_info.message_id)
span_id = RandomIdGenerator().generate_span_id()
context = SpanContext(
trace_id=trace_id,
@ -521,7 +532,7 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
}
metadata.update(trace_info.metadata)
trace_id = uuid_to_trace_id(trace_info.message_id)
trace_id = string_to_trace_id128(trace_info.message_id)
span_id = RandomIdGenerator().generate_span_id()
context = SpanContext(
trace_id=trace_id,
@ -568,7 +579,7 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
"tool_config": json.dumps(trace_info.tool_config, ensure_ascii=False),
}
trace_id = uuid_to_trace_id(trace_info.message_id)
trace_id = string_to_trace_id128(trace_info.message_id)
tool_span_id = RandomIdGenerator().generate_span_id()
logger.info("[Arize/Phoenix] Creating tool trace with trace_id: %s, span_id: %s", trace_id, tool_span_id)
@ -629,7 +640,7 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
}
metadata.update(trace_info.metadata)
trace_id = uuid_to_trace_id(trace_info.message_id)
trace_id = string_to_trace_id128(trace_info.message_id)
span_id = RandomIdGenerator().generate_span_id()
context = SpanContext(
trace_id=trace_id,

2
api/core/ops/entities/config_entity.py
View File

@ -87,7 +87,7 @@ class PhoenixConfig(BaseTracingConfig):
@field_validator("endpoint")
@classmethod
def endpoint_validator(cls, v, info: ValidationInfo):
return cls.validate_endpoint_url(v, "https://app.phoenix.arize.com")
return validate_url_with_path(v, "https://app.phoenix.arize.com")
class LangfuseConfig(BaseTracingConfig):

1
api/core/ops/entities/trace_entity.py
View File

@ -14,6 +14,7 @@ class BaseTraceInfo(BaseModel):
start_time: Optional[datetime] = None
end_time: Optional[datetime] = None
metadata: dict[str, Any]
trace_id: Optional[str] = None
@field_validator("inputs", "outputs")
@classmethod

19
api/core/ops/langfuse_trace/langfuse_trace.py
View File

@ -67,14 +67,13 @@ class LangFuseDataTrace(BaseTraceInstance):
self.generate_name_trace(trace_info)
def workflow_trace(self, trace_info: WorkflowTraceInfo):
external_trace_id = trace_info.metadata.get("external_trace_id")
trace_id = external_trace_id or trace_info.workflow_run_id
trace_id = trace_info.trace_id or trace_info.workflow_run_id
user_id = trace_info.metadata.get("user_id")
metadata = trace_info.metadata
metadata["workflow_app_log_id"] = trace_info.workflow_app_log_id
if trace_info.message_id:
trace_id = external_trace_id or trace_info.message_id
trace_id = trace_info.trace_id or trace_info.message_id
name = TraceTaskName.MESSAGE_TRACE.value
trace_data = LangfuseTrace(
id=trace_id,
@ -250,8 +249,10 @@ class LangFuseDataTrace(BaseTraceInstance):
user_id = end_user_data.session_id
metadata["user_id"] = user_id
trace_id = trace_info.trace_id or message_id
trace_data = LangfuseTrace(
id=message_id,
id=trace_id,
user_id=user_id,
name=TraceTaskName.MESSAGE_TRACE.value,
input={
@ -285,7 +286,7 @@ class LangFuseDataTrace(BaseTraceInstance):
langfuse_generation_data = LangfuseGeneration(
name="llm",
trace_id=message_id,
trace_id=trace_id,
start_time=trace_info.start_time,
end_time=trace_info.end_time,
model=message_data.model_id,
@ -311,7 +312,7 @@ class LangFuseDataTrace(BaseTraceInstance):
"preset_response": trace_info.preset_response,
"inputs": trace_info.inputs,
},
trace_id=trace_info.message_id,
trace_id=trace_info.trace_id or trace_info.message_id,
start_time=trace_info.start_time or trace_info.message_data.created_at,
end_time=trace_info.end_time or trace_info.message_data.created_at,
metadata=trace_info.metadata,
@ -334,7 +335,7 @@ class LangFuseDataTrace(BaseTraceInstance):
name=TraceTaskName.SUGGESTED_QUESTION_TRACE.value,
input=trace_info.inputs,
output=str(trace_info.suggested_question),
trace_id=trace_info.message_id,
trace_id=trace_info.trace_id or trace_info.message_id,
start_time=trace_info.start_time,
end_time=trace_info.end_time,
metadata=trace_info.metadata,
@ -352,7 +353,7 @@ class LangFuseDataTrace(BaseTraceInstance):
name=TraceTaskName.DATASET_RETRIEVAL_TRACE.value,
input=trace_info.inputs,
output={"documents": trace_info.documents},
trace_id=trace_info.message_id,
trace_id=trace_info.trace_id or trace_info.message_id,
start_time=trace_info.start_time or trace_info.message_data.created_at,
end_time=trace_info.end_time or trace_info.message_data.updated_at,
metadata=trace_info.metadata,
@ -365,7 +366,7 @@ class LangFuseDataTrace(BaseTraceInstance):
name=trace_info.tool_name,
input=trace_info.tool_inputs,
output=trace_info.tool_outputs,
trace_id=trace_info.message_id,
trace_id=trace_info.trace_id or trace_info.message_id,
start_time=trace_info.start_time,
end_time=trace_info.end_time,
metadata=trace_info.metadata,

17
api/core/ops/langsmith_trace/langsmith_trace.py
View File

@ -65,8 +65,7 @@ class LangSmithDataTrace(BaseTraceInstance):
self.generate_name_trace(trace_info)
def workflow_trace(self, trace_info: WorkflowTraceInfo):
external_trace_id = trace_info.metadata.get("external_trace_id")
trace_id = external_trace_id or trace_info.message_id or trace_info.workflow_run_id
trace_id = trace_info.trace_id or trace_info.message_id or trace_info.workflow_run_id
if trace_info.start_time is None:
trace_info.start_time = datetime.now()
message_dotted_order = (
@ -290,7 +289,7 @@ class LangSmithDataTrace(BaseTraceInstance):
reference_example_id=None,
input_attachments={},
output_attachments={},
trace_id=None,
trace_id=trace_info.trace_id,
dotted_order=None,
parent_run_id=None,
)
@ -319,7 +318,7 @@ class LangSmithDataTrace(BaseTraceInstance):
reference_example_id=None,
input_attachments={},
output_attachments={},
trace_id=None,
trace_id=trace_info.trace_id,
dotted_order=None,
id=str(uuid.uuid4()),
)
@ -351,7 +350,7 @@ class LangSmithDataTrace(BaseTraceInstance):
reference_example_id=None,
input_attachments={},
output_attachments={},
trace_id=None,
trace_id=trace_info.trace_id,
dotted_order=None,
error="",
file_list=[],
@ -381,7 +380,7 @@ class LangSmithDataTrace(BaseTraceInstance):
reference_example_id=None,
input_attachments={},
output_attachments={},
trace_id=None,
trace_id=trace_info.trace_id,
dotted_order=None,
error="",
file_list=[],
@ -410,7 +409,7 @@ class LangSmithDataTrace(BaseTraceInstance):
reference_example_id=None,
input_attachments={},
output_attachments={},
trace_id=None,
trace_id=trace_info.trace_id,
dotted_order=None,
error="",
file_list=[],
@ -440,7 +439,7 @@ class LangSmithDataTrace(BaseTraceInstance):
reference_example_id=None,
input_attachments={},
output_attachments={},
trace_id=None,
trace_id=trace_info.trace_id,
dotted_order=None,
error=trace_info.error or "",
)
@ -465,7 +464,7 @@ class LangSmithDataTrace(BaseTraceInstance):
reference_example_id=None,
input_attachments={},
output_attachments={},
trace_id=None,
trace_id=trace_info.trace_id,
dotted_order=None,
error="",
file_list=[],

19
api/core/ops/opik_trace/opik_trace.py
View File

@ -96,8 +96,7 @@ class OpikDataTrace(BaseTraceInstance):
self.generate_name_trace(trace_info)
def workflow_trace(self, trace_info: WorkflowTraceInfo):
external_trace_id = trace_info.metadata.get("external_trace_id")
dify_trace_id = external_trace_id or trace_info.workflow_run_id
dify_trace_id = trace_info.trace_id or trace_info.workflow_run_id
opik_trace_id = prepare_opik_uuid(trace_info.start_time, dify_trace_id)
workflow_metadata = wrap_metadata(
trace_info.metadata, message_id=trace_info.message_id, workflow_app_log_id=trace_info.workflow_app_log_id
@ -105,7 +104,7 @@ class OpikDataTrace(BaseTraceInstance):
root_span_id = None
if trace_info.message_id:
dify_trace_id = external_trace_id or trace_info.message_id
dify_trace_id = trace_info.trace_id or trace_info.message_id
opik_trace_id = prepare_opik_uuid(trace_info.start_time, dify_trace_id)
trace_data = {
@ -276,7 +275,7 @@ class OpikDataTrace(BaseTraceInstance):
return
metadata = trace_info.metadata
message_id = trace_info.message_id
dify_trace_id = trace_info.trace_id or trace_info.message_id
user_id = message_data.from_account_id
metadata["user_id"] = user_id
@ -291,7 +290,7 @@ class OpikDataTrace(BaseTraceInstance):
metadata["end_user_id"] = end_user_id
trace_data = {
"id": prepare_opik_uuid(trace_info.start_time, message_id),
"id": prepare_opik_uuid(trace_info.start_time, dify_trace_id),
"name": TraceTaskName.MESSAGE_TRACE.value,
"start_time": trace_info.start_time,
"end_time": trace_info.end_time,
@ -330,7 +329,7 @@ class OpikDataTrace(BaseTraceInstance):
start_time = trace_info.start_time or trace_info.message_data.created_at
span_data = {
"trace_id": prepare_opik_uuid(start_time, trace_info.message_id),
"trace_id": prepare_opik_uuid(start_time, trace_info.trace_id or trace_info.message_id),
"name": TraceTaskName.MODERATION_TRACE.value,
"type": "tool",
"start_time": start_time,
@ -356,7 +355,7 @@ class OpikDataTrace(BaseTraceInstance):
start_time = trace_info.start_time or message_data.created_at
span_data = {
"trace_id": prepare_opik_uuid(start_time, trace_info.message_id),
"trace_id": prepare_opik_uuid(start_time, trace_info.trace_id or trace_info.message_id),
"name": TraceTaskName.SUGGESTED_QUESTION_TRACE.value,
"type": "tool",
"start_time": start_time,
@ -376,7 +375,7 @@ class OpikDataTrace(BaseTraceInstance):
start_time = trace_info.start_time or trace_info.message_data.created_at
span_data = {
"trace_id": prepare_opik_uuid(start_time, trace_info.message_id),
"trace_id": prepare_opik_uuid(start_time, trace_info.trace_id or trace_info.message_id),
"name": TraceTaskName.DATASET_RETRIEVAL_TRACE.value,
"type": "tool",
"start_time": start_time,
@ -391,7 +390,7 @@ class OpikDataTrace(BaseTraceInstance):
def tool_trace(self, trace_info: ToolTraceInfo):
span_data = {
"trace_id": prepare_opik_uuid(trace_info.start_time, trace_info.message_id),
"trace_id": prepare_opik_uuid(trace_info.start_time, trace_info.trace_id or trace_info.message_id),
"name": trace_info.tool_name,
"type": "tool",
"start_time": trace_info.start_time,
@ -406,7 +405,7 @@ class OpikDataTrace(BaseTraceInstance):
def generate_name_trace(self, trace_info: GenerateNameTraceInfo):
trace_data = {
"id": prepare_opik_uuid(trace_info.start_time, trace_info.message_id),
"id": prepare_opik_uuid(trace_info.start_time, trace_info.trace_id or trace_info.message_id),
"name": TraceTaskName.GENERATE_NAME_TRACE.value,
"start_time": trace_info.start_time,
"end_time": trace_info.end_time,

16
api/core/ops/ops_trace_manager.py
View File

@ -422,8 +422,11 @@ class TraceTask:
self.timer = timer
self.file_base_url = os.getenv("FILES_URL", "http://127.0.0.1:5001")
self.app_id = None
self.trace_id = None
self.kwargs = kwargs
external_trace_id = kwargs.get("external_trace_id")
if external_trace_id:
self.trace_id = external_trace_id
def execute(self):
return self.preprocess()
@ -520,11 +523,8 @@ class TraceTask:
"app_id": workflow_run.app_id,
}
external_trace_id = self.kwargs.get("external_trace_id")
if external_trace_id:
metadata["external_trace_id"] = external_trace_id
workflow_trace_info = WorkflowTraceInfo(
trace_id=self.trace_id,
workflow_data=workflow_run.to_dict(),
conversation_id=conversation_id,
workflow_id=workflow_id,
@ -584,6 +584,7 @@ class TraceTask:
message_tokens = message_data.message_tokens
message_trace_info = MessageTraceInfo(
trace_id=self.trace_id,
message_id=message_id,
message_data=message_data.to_dict(),
conversation_model=conversation_mode,
@ -627,6 +628,7 @@ class TraceTask:
workflow_app_log_id = str(workflow_app_log_data.id) if workflow_app_log_data else None
moderation_trace_info = ModerationTraceInfo(
trace_id=self.trace_id,
message_id=workflow_app_log_id or message_id,
inputs=inputs,
message_data=message_data.to_dict(),
@ -667,6 +669,7 @@ class TraceTask:
workflow_app_log_id = str(workflow_app_log_data.id) if workflow_app_log_data else None
suggested_question_trace_info = SuggestedQuestionTraceInfo(
trace_id=self.trace_id,
message_id=workflow_app_log_id or message_id,
message_data=message_data.to_dict(),
inputs=message_data.message,
@ -708,6 +711,7 @@ class TraceTask:
}
dataset_retrieval_trace_info = DatasetRetrievalTraceInfo(
trace_id=self.trace_id,
message_id=message_id,
inputs=message_data.query or message_data.inputs,
documents=[doc.model_dump() for doc in documents] if documents else [],
@ -772,6 +776,7 @@ class TraceTask:
)
tool_trace_info = ToolTraceInfo(
trace_id=self.trace_id,
message_id=message_id,
message_data=message_data.to_dict(),
tool_name=tool_name,
@ -807,6 +812,7 @@ class TraceTask:
}
generate_name_trace_info = GenerateNameTraceInfo(
trace_id=self.trace_id,
conversation_id=conversation_id,
inputs=inputs,
outputs=generate_conversation_name,

35
api/core/ops/weave_trace/weave_trace.py
View File

@ -87,8 +87,7 @@ class WeaveDataTrace(BaseTraceInstance):
self.generate_name_trace(trace_info)
def workflow_trace(self, trace_info: WorkflowTraceInfo):
external_trace_id = trace_info.metadata.get("external_trace_id")
trace_id = external_trace_id or trace_info.message_id or trace_info.workflow_run_id
trace_id = trace_info.trace_id or trace_info.message_id or trace_info.workflow_run_id
if trace_info.start_time is None:
trace_info.start_time = datetime.now()
@ -245,8 +244,12 @@ class WeaveDataTrace(BaseTraceInstance):
attributes["start_time"] = trace_info.start_time
attributes["end_time"] = trace_info.end_time
attributes["tags"] = ["message", str(trace_info.conversation_mode)]
trace_id = trace_info.trace_id or message_id
attributes["trace_id"] = trace_id
message_run = WeaveTraceModel(
id=message_id,
id=trace_id,
op=str(TraceTaskName.MESSAGE_TRACE.value),
input_tokens=trace_info.message_tokens,
output_tokens=trace_info.answer_tokens,
@ -274,7 +277,7 @@ class WeaveDataTrace(BaseTraceInstance):
)
self.start_call(
llm_run,
parent_run_id=message_id,
parent_run_id=trace_id,
)
self.finish_call(llm_run)
self.finish_call(message_run)
@ -289,6 +292,9 @@ class WeaveDataTrace(BaseTraceInstance):
attributes["start_time"] = trace_info.start_time or trace_info.message_data.created_at
attributes["end_time"] = trace_info.end_time or trace_info.message_data.updated_at
trace_id = trace_info.trace_id or trace_info.message_id
attributes["trace_id"] = trace_id
moderation_run = WeaveTraceModel(
id=str(uuid.uuid4()),
op=str(TraceTaskName.MODERATION_TRACE.value),
@ -303,7 +309,7 @@ class WeaveDataTrace(BaseTraceInstance):
exception=getattr(trace_info, "error", None),
file_list=[],
)
self.start_call(moderation_run, parent_run_id=trace_info.message_id)
self.start_call(moderation_run, parent_run_id=trace_id)
self.finish_call(moderation_run)
def suggested_question_trace(self, trace_info: SuggestedQuestionTraceInfo):
@ -316,6 +322,9 @@ class WeaveDataTrace(BaseTraceInstance):
attributes["start_time"] = (trace_info.start_time or message_data.created_at,)
attributes["end_time"] = (trace_info.end_time or message_data.updated_at,)
trace_id = trace_info.trace_id or trace_info.message_id
attributes["trace_id"] = trace_id
suggested_question_run = WeaveTraceModel(
id=str(uuid.uuid4()),
op=str(TraceTaskName.SUGGESTED_QUESTION_TRACE.value),
@ -326,7 +335,7 @@ class WeaveDataTrace(BaseTraceInstance):
file_list=[],
)
self.start_call(suggested_question_run, parent_run_id=trace_info.message_id)
self.start_call(suggested_question_run, parent_run_id=trace_id)
self.finish_call(suggested_question_run)
def dataset_retrieval_trace(self, trace_info: DatasetRetrievalTraceInfo):
@ -338,6 +347,9 @@ class WeaveDataTrace(BaseTraceInstance):
attributes["start_time"] = (trace_info.start_time or trace_info.message_data.created_at,)
attributes["end_time"] = (trace_info.end_time or trace_info.message_data.updated_at,)
trace_id = trace_info.trace_id or trace_info.message_id
attributes["trace_id"] = trace_id
dataset_retrieval_run = WeaveTraceModel(
id=str(uuid.uuid4()),
op=str(TraceTaskName.DATASET_RETRIEVAL_TRACE.value),
@ -348,7 +360,7 @@ class WeaveDataTrace(BaseTraceInstance):
file_list=[],
)
self.start_call(dataset_retrieval_run, parent_run_id=trace_info.message_id)
self.start_call(dataset_retrieval_run, parent_run_id=trace_id)
self.finish_call(dataset_retrieval_run)
def tool_trace(self, trace_info: ToolTraceInfo):
@ -357,6 +369,11 @@ class WeaveDataTrace(BaseTraceInstance):
attributes["start_time"] = trace_info.start_time
attributes["end_time"] = trace_info.end_time
message_id = trace_info.message_id or getattr(trace_info, "conversation_id", None)
message_id = message_id or None
trace_id = trace_info.trace_id or message_id
attributes["trace_id"] = trace_id
tool_run = WeaveTraceModel(
id=str(uuid.uuid4()),
op=trace_info.tool_name,
@ -366,9 +383,7 @@ class WeaveDataTrace(BaseTraceInstance):
attributes=attributes,
exception=trace_info.error,
)
message_id = trace_info.message_id or getattr(trace_info, "conversation_id", None)
message_id = message_id or None
self.start_call(tool_run, parent_run_id=message_id)
self.start_call(tool_run, parent_run_id=trace_id)
self.finish_call(tool_run)
def generate_name_trace(self, trace_info: GenerateNameTraceInfo):

180
api/core/rag/datasource/vdb/elasticsearch/elasticsearch_vector.py
View File

@ -22,22 +22,50 @@ logger = logging.getLogger(__name__)
class ElasticSearchConfig(BaseModel):
host: str
port: int
username: str
password: str
# Regular Elasticsearch config
host: Optional[str] = None
port: Optional[int] = None
username: Optional[str] = None
password: Optional[str] = None
# Elastic Cloud specific config
cloud_url: Optional[str] = None # Cloud URL for Elasticsearch Cloud
api_key: Optional[str] = None
# Common config
use_cloud: bool = False
ca_certs: Optional[str] = None
verify_certs: bool = False
request_timeout: int = 100000
retry_on_timeout: bool = True
max_retries: int = 10000
@model_validator(mode="before")
@classmethod
def validate_config(cls, values: dict) -> dict:
if not values["host"]:
raise ValueError("config HOST is required")
if not values["port"]:
raise ValueError("config PORT is required")
if not values["username"]:
raise ValueError("config USERNAME is required")
if not values["password"]:
raise ValueError("config PASSWORD is required")
use_cloud = values.get("use_cloud", False)
cloud_url = values.get("cloud_url")
if use_cloud:
# Cloud configuration validation - requires cloud_url and api_key
if not cloud_url:
raise ValueError("cloud_url is required for Elastic Cloud")
api_key = values.get("api_key")
if not api_key:
raise ValueError("api_key is required for Elastic Cloud")
else:
# Regular Elasticsearch validation
if not values.get("host"):
raise ValueError("config HOST is required for regular Elasticsearch")
if not values.get("port"):
raise ValueError("config PORT is required for regular Elasticsearch")
if not values.get("username"):
raise ValueError("config USERNAME is required for regular Elasticsearch")
if not values.get("password"):
raise ValueError("config PASSWORD is required for regular Elasticsearch")
return values
@ -50,21 +78,69 @@ class ElasticSearchVector(BaseVector):
self._attributes = attributes
def _init_client(self, config: ElasticSearchConfig) -> Elasticsearch:
"""
Initialize Elasticsearch client for both regular Elasticsearch and Elastic Cloud.
"""
try:
parsed_url = urlparse(config.host)
if parsed_url.scheme in {"http", "https"}:
hosts = f"{config.host}:{config.port}"
# Check if using Elastic Cloud
client_config: dict[str, Any]
if config.use_cloud and config.cloud_url:
client_config = {
"request_timeout": config.request_timeout,
"retry_on_timeout": config.retry_on_timeout,
"max_retries": config.max_retries,
"verify_certs": config.verify_certs,
}
# Parse cloud URL and configure hosts
parsed_url = urlparse(config.cloud_url)
host = f"{parsed_url.scheme}://{parsed_url.hostname}"
if parsed_url.port:
host += f":{parsed_url.port}"
client_config["hosts"] = [host]
# API key authentication for cloud
client_config["api_key"] = config.api_key
# SSL settings
if config.ca_certs:
client_config["ca_certs"] = config.ca_certs
else:
hosts = f"http://{config.host}:{config.port}"
client = Elasticsearch(
hosts=hosts,
basic_auth=(config.username, config.password),
request_timeout=100000,
retry_on_timeout=True,
max_retries=10000,
)
except requests.exceptions.ConnectionError:
raise ConnectionError("Vector database connection error")
# Regular Elasticsearch configuration
parsed_url = urlparse(config.host or "")
if parsed_url.scheme in {"http", "https"}:
hosts = f"{config.host}:{config.port}"
use_https = parsed_url.scheme == "https"
else:
hosts = f"http://{config.host}:{config.port}"
use_https = False
client_config = {
"hosts": [hosts],
"basic_auth": (config.username, config.password),
"request_timeout": config.request_timeout,
"retry_on_timeout": config.retry_on_timeout,
"max_retries": config.max_retries,
}
# Only add SSL settings if using HTTPS
if use_https:
client_config["verify_certs"] = config.verify_certs
if config.ca_certs:
client_config["ca_certs"] = config.ca_certs
client = Elasticsearch(**client_config)
# Test connection
if not client.ping():
raise ConnectionError("Failed to connect to Elasticsearch")
except requests.exceptions.ConnectionError as e:
raise ConnectionError(f"Vector database connection error: {str(e)}")
except Exception as e:
raise ConnectionError(f"Elasticsearch client initialization failed: {str(e)}")
return client
@ -209,7 +285,11 @@ class ElasticSearchVector(BaseVector):
},
}
}
self._client.indices.create(index=self._collection_name, mappings=mappings)
logger.info("Created index %s with dimension %s", self._collection_name, dim)
else:
logger.info("Collection %s already exists.", self._collection_name)
redis_client.set(collection_exist_cache_key, 1, ex=3600)
@ -225,13 +305,51 @@ class ElasticSearchVectorFactory(AbstractVectorFactory):
dataset.index_struct = json.dumps(self.gen_index_struct_dict(VectorType.ELASTICSEARCH, collection_name))
config = current_app.config
# Check if ELASTICSEARCH_USE_CLOUD is explicitly set to false (boolean)
use_cloud_env = config.get("ELASTICSEARCH_USE_CLOUD", False)
if use_cloud_env is False:
# Use regular Elasticsearch with config values
config_dict = {
"use_cloud": False,
"host": config.get("ELASTICSEARCH_HOST", "elasticsearch"),
"port": config.get("ELASTICSEARCH_PORT", 9200),
"username": config.get("ELASTICSEARCH_USERNAME", "elastic"),
"password": config.get("ELASTICSEARCH_PASSWORD", "elastic"),
}
else:
# Check for cloud configuration
cloud_url = config.get("ELASTICSEARCH_CLOUD_URL")
if cloud_url:
config_dict = {
"use_cloud": True,
"cloud_url": cloud_url,
"api_key": config.get("ELASTICSEARCH_API_KEY"),
}
else:
# Fallback to regular Elasticsearch
config_dict = {
"use_cloud": False,
"host": config.get("ELASTICSEARCH_HOST", "localhost"),
"port": config.get("ELASTICSEARCH_PORT", 9200),
"username": config.get("ELASTICSEARCH_USERNAME", "elastic"),
"password": config.get("ELASTICSEARCH_PASSWORD", ""),
}
# Common configuration
config_dict.update(
{
"ca_certs": str(config.get("ELASTICSEARCH_CA_CERTS")) if config.get("ELASTICSEARCH_CA_CERTS") else None,
"verify_certs": bool(config.get("ELASTICSEARCH_VERIFY_CERTS", False)),
"request_timeout": int(config.get("ELASTICSEARCH_REQUEST_TIMEOUT", 100000)),
"retry_on_timeout": bool(config.get("ELASTICSEARCH_RETRY_ON_TIMEOUT", True)),
"max_retries": int(config.get("ELASTICSEARCH_MAX_RETRIES", 10000)),
}
)
return ElasticSearchVector(
index_name=collection_name,
config=ElasticSearchConfig(
host=config.get("ELASTICSEARCH_HOST", "localhost"),
port=config.get("ELASTICSEARCH_PORT", 9200),
username=config.get("ELASTICSEARCH_USERNAME", ""),
password=config.get("ELASTICSEARCH_PASSWORD", ""),
),
config=ElasticSearchConfig(**config_dict),
attributes=[],
)

2
api/core/rag/entities/metadata_entities.py
View File

@ -13,6 +13,8 @@ SupportedComparisonOperator = Literal[
"is not",
"empty",
"not empty",
"in",
"not in",
# for number
"=",
"≠",

5
api/core/rag/extractor/notion_extractor.py
View File

@ -1,5 +1,6 @@
import json
import logging
import operator
from typing import Any, Optional, cast
import requests
@ -132,13 +133,15 @@ class NotionExtractor(BaseExtractor):
data[property_name] = value
row_dict = {k: v for k, v in data.items() if v}
row_content = ""
for key, value in row_dict.items():
for key, value in sorted(row_dict.items(), key=operator.itemgetter(0)):
if isinstance(value, dict):
value_dict = {k: v for k, v in value.items() if v}
value_content = "".join(f"{k}:{v} " for k, v in value_dict.items())
row_content = row_content + f"{key}:{value_content}\n"
else:
row_content = row_content + f"{key}:{value}\n"
if "url" in result:
row_content = row_content + f"Row Page URL:{result.get('url', '')}\n"
database_content.append(row_content)
has_more = response_data.get("has_more", False)

2
api/core/tools/workflow_as_tool/tool.py
View File

@ -142,7 +142,7 @@ class WorkflowTool(Tool):
if not version:
workflow = (
db.session.query(Workflow)
.where(Workflow.app_id == app_id, Workflow.version != "draft")
.where(Workflow.app_id == app_id, Workflow.version != Workflow.VERSION_DRAFT)
.order_by(Workflow.created_at.desc())
.first()
)

4
api/core/workflow/nodes/http_request/executor.py
View File

@ -265,9 +265,9 @@ class Executor:
if not authorization.config.header:
authorization.config.header = "Authorization"
if self.auth.config.type == "bearer":
if self.auth.config.type == "bearer" and authorization.config.api_key:
headers[authorization.config.header] = f"Bearer {authorization.config.api_key}"
elif self.auth.config.type == "basic":
elif self.auth.config.type == "basic" and authorization.config.api_key:
credentials = authorization.config.api_key
if ":" in credentials:
encoded_credentials = base64.b64encode(credentials.encode("utf-8")).decode("utf-8")

2
api/core/workflow/nodes/knowledge_retrieval/entities.py
View File

@ -74,6 +74,8 @@ SupportedComparisonOperator = Literal[
"is not",
"empty",
"not empty",
"in",
"not in",
# for number
"=",
"≠",

22
api/core/workflow/nodes/knowledge_retrieval/knowledge_retrieval_node.py
View File

@ -602,6 +602,28 @@ class KnowledgeRetrievalNode(BaseNode):
**{key: metadata_name, key_value: f"%{value}"}
)
)
case "in":
if isinstance(value, str):
escaped_values = [v.strip().replace("'", "''") for v in str(value).split(",")]
escaped_value_str = ",".join(escaped_values)
else:
escaped_value_str = str(value)
filters.append(
(text(f"documents.doc_metadata ->> :{key} = any(string_to_array(:{key_value},','))")).params(
**{key: metadata_name, key_value: escaped_value_str}
)
)
case "not in":
if isinstance(value, str):
escaped_values = [v.strip().replace("'", "''") for v in str(value).split(",")]
escaped_value_str = ",".join(escaped_values)
else:
escaped_value_str = str(value)
filters.append(
(text(f"documents.doc_metadata ->> :{key} != all(string_to_array(:{key_value},','))")).params(
**{key: metadata_name, key_value: escaped_value_str}
)
)
case "=" | "is":
if isinstance(value, str):
filters.append(Document.doc_metadata[metadata_name] == f'"{value}"')