Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion apps/application/models/application_chat.py
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,10 @@ def get_ai_message(self):
return to_ai_message_list(self.messages)

def get_node_details_runtime_node_id(self, runtime_node_id):
return self.details.get(runtime_node_id, None)
for node_details in self.details:
if node_details.get("node_id") == runtime_node_id:
return node_details
return None

class Meta:
db_table = "application_chat_record"
Expand Down
46 changes: 27 additions & 19 deletions apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,18 +18,20 @@
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
from rest_framework import serializers

from application.workflow.message.aggregator import AggregationManager
from application.flow.tools import get_tools, mcp_response_generator
from application.models import Application, ApplicationAccessToken, ApplicationApiKey
from application.workflow.common import WorkflowType
from application.workflow.i_node import INode
from application.workflow.message.struct.content import NodeInfo, Position
from application.workflow.message.struct.content import NodeInfo, Position, Content
from application.workflow.message.struct.reasoning_content import ReasoningContent
from application.workflow.message.struct.text_content import TextContent
from application.workflow.message.struct.tool_content import ToolContent
from application.workflow.status import Status
from application.workflow.tools import Reasoning
from common.exception.app_exception import AppApiException
from common.utils.common import guess_image_format
from common.utils.messages_util import to_ai_message_list, to_human_message_list
from common.utils.rsa_util import rsa_long_decrypt
from common.utils.shared_resource_auth import filter_authorized_ids
from common.utils.tool_code import ToolExecutor
Expand Down Expand Up @@ -96,7 +98,8 @@ def _get_node_message(chat_record, runtime_node_id):
node_details = chat_record.get_node_details_runtime_node_id(runtime_node_id)
if node_details is None:
return []
return [HumanMessage(node_details.get("question")), AIMessage(node_details.get("answer"))]
return [*to_human_message_list(node_details.get("question")), *to_ai_message_list(node_details.get("messages"))]
return [HumanMessage(node_details.get("question")), AIMessage(node_details.get("messages"))]


def _get_workflow_message(chat_record):
Expand Down Expand Up @@ -193,6 +196,12 @@ class AIChatNode(INode):
supported_workflow_type_list = [WorkflowType.APPLICATION, WorkflowType.KNOWLEDGE, WorkflowType.TOOL]
type = "ai-chat-node"

def write(self, message: Content):
super().write(message)
if not self.data.get("messages"):
self.data["messages"] = []
self.data["messages"].append(message)

def execute(self):
workflow_params = self.get_workflow_parameters()
node_params = self.get_parameters()
Expand Down Expand Up @@ -254,7 +263,7 @@ def execute(self):
"reasoning_content_end": "</think>",
"reasoning_content_start": "<think>",
}
self.write_context("model_setting", model_setting)
self.data["model_setting"] = model_setting

chat_model = get_model_instance_by_model_workspace_id(model_id, workspace_id, **(model_params_setting or {}))

Expand All @@ -263,12 +272,12 @@ def execute(self):
"history_message",
[{"content": message.content, "role": message.type} for message in (history_message or [])],
)

question = self._generate_prompt_question(prompt, chat_model, vision, image_list, video_list)
self.write_context("question", question.content)
question_str = self.workflow_manage.generate_prompt(prompt)
question = self._generate_prompt_question(question_str, chat_model, vision, image_list, video_list)
self.data["question"] = {"content": question_str, "image_list": image_list, "video_list": video_list}

system = self.workflow_manage.generate_prompt(system)
self.write_context("system", system)
self.data["system"] = system

message_list = [*history_message, question]

Expand Down Expand Up @@ -321,7 +330,7 @@ def execute(self):
r, chat_model, message_list_with_system, question.content, is_result, text_content_id
)

def _generate_prompt_question(self, prompt, model, vision, image_list, video_list):
def _generate_prompt_question(self, question_str, model, vision, image_list, video_list):
images = []
videos = []
if vision:
Expand All @@ -334,9 +343,7 @@ def _generate_prompt_question(self, prompt, model, vision, image_list, video_lis
if video_list:
video = self.workflow_manage.get_reference_field(video_list[0], video_list[1:])
videos = _process_videos(video, model)
return HumanMessage(
content=[*videos, *images, {"type": "text", "text": self.workflow_manage.generate_prompt(prompt)}]
)
return HumanMessage(content=[*videos, *images, {"type": "text", "text": question_str}])

def _stream_response(self, response, chat_model, message_list, question, reasoning_content_id, text_content_id):
node_info = NodeInfo(self.get_node_id(), self.get_node_name(), Status.RUNNING)
Expand Down Expand Up @@ -436,11 +443,10 @@ def _invoke_response(self, response, chat_model, message_list, question, is_resu
def _write_final_context(self, chat_model, message_list, question, answer, reasoning_content):
message_tokens = chat_model.get_num_tokens_from_messages(message_list)
answer_tokens = chat_model.get_num_tokens(answer)
self.write_context("message_tokens", message_tokens)
self.write_context("answer_tokens", answer_tokens)
self.data["message_tokens"] = message_tokens
self.data["answer_tokens"] = answer_tokens
self.write_context("answer", answer)
self.write_context("question", question)
self.write_context("reasoning_content", reasoning_content)
self.data["reasoning_content"] = reasoning_content

def _handle_mcp(
self,
Expand Down Expand Up @@ -498,7 +504,6 @@ def _handle_mcp(

tools = get_tools(source_type, chat_id, tool_ids, workspace_id)
if tool_ids and len(tool_ids) > 0:
self.write_context("tool_ids", tool_ids)
custom_tools_map = {
str(t.id): t for t in QuerySet(Tool).filter(id__in=tool_ids, tool_type=ToolType.CUSTOM, is_active=True)
}
Expand All @@ -516,7 +521,6 @@ def _handle_mcp(
mcp_servers_config[str(tool.id)] = tool_config

if application_ids and len(application_ids) > 0:
self.write_context("application_ids", application_ids)
apps_map = {str(a.id): a for a in QuerySet(Application).filter(id__in=application_ids, is_publish=True)}
app_keys_map = {
str(ak.application_id): ak
Expand Down Expand Up @@ -550,7 +554,6 @@ def _handle_mcp(
mcp_servers_config[app.name] = app_config

if skill_tool_ids and len(skill_tool_ids) > 0:
self.write_context("skill_tool_ids", skill_tool_ids)
skill_file_items = []
skill_tools_map = {str(t.id): t for t in QuerySet(Tool).filter(id__in=skill_tool_ids, is_active=True)}
for tool_id in skill_tool_ids:
Expand Down Expand Up @@ -632,14 +635,19 @@ def _get_reference_content(self, fields):

def get_details(self, index: int = 0, position: dict = None, old_details: dict = None, **kwargs):
details = super().get_details(index, position, old_details, **kwargs)
aggregation = AggregationManager()
for m in self.data.get("messages"):
aggregation.aggregate(m)
messages = aggregation.get_contents()
details.update(
{
"question": self.get_context("question"),
"question": self.data.get("question"),
"answer": self.get_context("answer"),
"reasoning_content": self.get_context("reasoning_content"),
"message_tokens": self.get_context("message_tokens"),
"answer_tokens": self.get_context("answer_tokens"),
"history_message": self.get_context("history_message"),
"messages": messages,
}
)
return details
8 changes: 7 additions & 1 deletion apps/chat/serializers/chat.py
Original file line number Diff line number Diff line change
Expand Up @@ -329,19 +329,25 @@ def on_complete(wf_manage, error):
)
messages = aggregation.get_contents()
old_details = None
chat_record = None
if chat_record_id is not None:
chat_record = QuerySet(ChatRecord).filter(id=chat_record_id).first()
if chat_record:
old_details = chat_record.details
if position and chat_record.messages:
messages = [*chat_record.messages, *messages]
details = wf_manage.get_details(position=position, old_details=old_details)
self.update_chat_record(chat_user_id, chat_record_id_str, wf_manage.context, messages, details)
ChatCountSerializer(data={"chat_id": chat_id}).update_chat()
# 表单续跑时 message_dict.content 为空;保留原记录里的用户问题,避免 WORKFLOW 历史丢问题
question = chat_record.question if (chat_record and chat_record.question) else message_dict
ChatHistory(chat_id).append(
ChatRecord(
id=chat_record_id_str,
chat_id=chat_id,
question=message_dict,
question=question,
messages=messages,
details=details,
create_time=timezone.now(),
)
)
Expand Down
2 changes: 2 additions & 0 deletions apps/chat/serializers/chat_history.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ def _to_map(r):
"chat_id": str(r.chat_id),
"question": r.question,
"messages": r.messages,
"details": r.details,
"create_time": r.create_time,
}

Expand All @@ -50,6 +51,7 @@ def _from_map(d):
chat_id=d.get("chat_id"),
question=d.get("question"),
messages=d.get("messages"),
details=d.get("details"),
create_time=d.get("create_time"),
)

Expand Down
Loading