From 08b7d0e2b8ace174b914bcd4b04688dab931d6c6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9C=B1=E6=BD=AE?= Date: Mon, 1 Dec 2025 21:52:14 +0800 Subject: [PATCH] add cot --- prompt/guideline_prompt.md | 189 +++++++++++-------------------------- routes/chat.py | 122 +++++------------------- utils/fastapi_utils.py | 57 +++-------- 3 files changed, 93 insertions(+), 275 deletions(-) diff --git a/prompt/guideline_prompt.md b/prompt/guideline_prompt.md index 9381337..2567ebe 100644 --- a/prompt/guideline_prompt.md +++ b/prompt/guideline_prompt.md @@ -1,143 +1,66 @@ -GENERAL INSTRUCTIONS ------------------ -In our system, the behavior of a conversational AI agent is guided by "guidelines". -Each guideline is composed of two parts: -- "condition": This is a natural-language condition that specifies when a guideline should apply -- "action": This is a natural-language instruction that should be followed by the agent +# 思考指南 -Task Description ----------------- -Your task is to identify which guidelines from the provided list apply to the most recent state of an interaction between yourself (an AI agent) and a user. -Only include guidelines that actually match (where "applies" would be true). Skip guidelines that don't match to improve efficiency. +请基于以下信息进行深度思考，制定清晰的行动计划。使用链式思考（Chain of Thought）方法，逐步分析并做出决策。 +## 思考框架 -Examples of Guideline Match Evaluations: +### 1. 理解阶段 (Understanding) +- **目标分析**: 用户的核心需求是什么？ +- **上下文总结**: 从聊天记录中提取关键信息 +- **约束识别**: 识别限制条件、规则和要求 + +### 2. 分析阶段 (Analysis) +- **问题拆解**: 将复杂问题分解为可管理的小任务 +- **依赖关系**: 分析任务间的先后依赖 +- **风险评估**: 识别潜在问题和解决方案 + +### 3. 决策阶段 (Decision) +- **方案对比**: 评估不同方法的优缺点 +- **资源评估**: 确定所需的工具、信息和步骤 +- **优先级排序**: 确定任务执行顺序 + +### 4. 计划阶段 (Planning) +- **步骤制定**: 详细的执行步骤 +- **检查点设置**: 验证每个步骤的完成标准 +- **应急预案**: 准备应对可能的障碍 + +--- + +## 聊天记录 (Chat History) ``` - Example #1: travel consultation - - **Chat History**: - user: Hi, I'm planning a trip to Italy next month. What can I do there? - ai_agent: That sounds exciting! I can help you with that. Do you prefer exploring cities or enjoying scenic landscapes? - user: Can you help me figure out the best time to visit Rome and what to pack? - user: Actually I'm also wondering — do I need any special visas or documents as an American citizen? - - - **Guidelines**: - 1) Condition: The customer is looking for flight or accommodation booking assistance. Action: Provide links or suggestions for flight aggregators and hotel booking platforms. - 2) Condition: The customer ask for activities recommendations. Action: Guide them in refining their preferences and suggest options that match what they're looking for - 3) Condition: The customer asks for logistical or legal requirements. Action: Provide a clear answer or direct them to a trusted official source if uncertain. - - - **Expected Result**: - ```json - { - "checks": [ - { - "guideline_id": "", - "rationale": "The customer now asked about visas and documents which are legal requirements" - } - ] - } - ``` - - Example #2: Course Consultation - - **Chat History**: - user:Hi, I'm interested in your Python programming course, but I'm not sure if I'm ready for it. - ai_agent:Happy to help! Could you share a bit about your background or experience with programming so far? - user:I've done some HTML and CSS, but never written real code before. - ai_agent:Thanks for sharing! That gives me a good idea. Our Python course is beginner-friendly, but it does assume you're comfortable with logic and problem solving. Would you like me to recommend a short prep course first? - user:That sounds useful. But I'm also wondering — is the course self-paced? I work full time. - - - **Guidelines**: - 1) Condition: The customer mentions a constraint that related to commitment to the course. Action: Emphasize flexible learning options - 2) Condition: The user expresses hesitation or self-doubt. Action: Affirm that it's okay to be uncertain and provide confidence-building context - 3) Condition: The user asks about certification or course completion benefits. Action: Clearly explain what the user receives - - - **Expected Result**: - ```json - { - "checks": [ - { - "guideline_id": "", - "rationale": "In the most recent message the customer mentions that they work full time which is a constraint" - }, - { - "guideline_id": "", - "rationale": "In the most recent message the user still sounds hesitating about their fit to the course" - } - ] - } - ``` - - Example #3: Login issue - - **Chat History**: - user:I'm having trouble logging into my account. - ai_agent:I'm sorry to hear that. Can you tell me what happens when you try to log in? - user:It says my password is incorrect. - ai_agent:Have you tried resetting your password? - user:Yes, I did, but I can't access my mail to complete the reset. - - - **Guidelines**: - 1) Condition When the user is having a problem with login. Action: Help then identify the problem and solve it - - - **Expected Result**: - ```json - { - "checks": [ - { - "guideline_id": "", - "rationale": "In the most recent message the customer is still pursuing their login problem, making the mail access problem a sub-issue rather than a new topic" - } - ] - } - - Example #4: Return Policy - - **Chat History**: - user: Hi, I'm thinking about ordering this coat, but I need to know — what's your return policy? - ai_agent: You can return items within 30 days either in-store or using our prepaid return label. - user: And what happens if I already wore it once? - - - **Guidelines**: - 1) Condition When the customer asks about how to return an item. Action: Mention both in-store and delivery service return options. - - - **Expected Result**: - ```json - { - "checks": [ - { - "guideline_id": "", - "rationale": "In the most recent message the customer asks about what happens when they wore the item, which an inquiry regarding returning an item" - } - ] - } -``` - -Terms: -{terms} - -Chat History: {chat_history} +``` -Guidelines List: +## 指南列表 (Guidelines List) +``` {guidelines_text} - -OUTPUT FORMAT: -The content in JSON format needs to be wrapped in "```json" and "```". -Only include guidelines that actually apply. -```json -{ - "checks": [ - { - "guideline_id": "1", - "rationale": "" - } - ] -} ``` -If no guidelines apply, return an empty checks array: -```json -{ - "checks": [] -} -``` - -Rationale Text Language: +## 语言要求 (Language) {language} + +--- + +## 输出格式 +请按照以下结构输出你的思考结果： + +### 🎯 核心目标 +[简洁描述用户的主要目标] + +### 📋 现状分析 +- **关键信息**: [从聊天记录中提取的重要信息] +- **限制条件**: [需要遵守的规则和约束] +- **可用资源**: [可以利用的工具和资源] + +### 🧠 任务拆解 +1. [第一步的具体行动] +2. [第二步的具体行动] +3. [第三步的具体行动] + - 子步骤 3.1 + - 子步骤 3.2 + +### ⚡ 执行计划 +- **立即行动**: [当前应该执行的第一个步骤] +- **后续步骤**: [接下来的行动计划] +- **检查标准**: [如何验证每个步骤的完成] + diff --git a/routes/chat.py b/routes/chat.py index 8f5bb29..de76514 100644 --- a/routes/chat.py +++ b/routes/chat.py @@ -16,7 +16,7 @@ from utils.api_models import ChatRequestV2 from utils.fastapi_utils import ( process_messages, extract_block_from_system_prompt, format_messages_to_chat_history, create_project_directory, extract_api_key_from_auth, generate_v2_auth_token, fetch_bot_config, - call_guideline_llm, _get_optimal_batch_size, process_guideline_batch, get_content_from_messages, call_preamble_llm, get_preamble_text, get_language_text, + call_guideline_llm, _get_optimal_batch_size, process_guideline, get_content_from_messages, call_preamble_llm, get_preamble_text, get_language_text, create_stream_chunk ) @@ -80,7 +80,7 @@ async def process_guidelines_and_terms( tuple: (agent, processed_system_prompt, guideline_reasoning, terms_analysis) """ # 提取system_prompt中的guideline和terms - processed_system_prompt, guidelines_list, terms_list = extract_block_from_system_prompt(system_prompt) + processed_system_prompt, guidelines, terms_list = extract_block_from_system_prompt(system_prompt) # 处理terms terms_analysis = "" @@ -109,47 +109,22 @@ async def process_guidelines_and_terms( # 处理guidelines guideline_reasoning = "" - active_guidelines = "" agent = None - if guidelines_list: - logger.info(f"Processing guidelines: {len(guidelines_list)} guidelines") - chat_history = format_messages_to_chat_history(messages) - - guidelines_count = len(guidelines_list) - batch_count = _get_optimal_batch_size(guidelines_count) - guidelines_per_batch = max(1, guidelines_count // batch_count) - - # 分批处理guidelines - batches = [] - for i in range(0, guidelines_count, guidelines_per_batch): - batch_guidelines = guidelines_list[i:i + guidelines_per_batch] - batch_strings = [] - for guideline in batch_guidelines: - guideline_str = f"{guideline['guideline_id']}) Condition: {guideline['condition']} Action: {guideline['action']}" - batch_strings.append(guideline_str) - batches.append(batch_strings) - - # 确保批次数量不超过要求的并发数 - while len(batches) > batch_count: - batches[-2].extend(batches[-1]) - batches.pop() - + if guidelines: # 创建所有任务 tasks = [] - - # 添加guideline批次任务 - for batch in batches: - task = process_guideline_batch( - guidelines_batch=batch, - chat_history=chat_history, - terms=terms_analysis, - language=language, - model_name=model_name, - api_key=api_key, - model_server=model_server - ) - tasks.append(task) + chat_history = format_messages_to_chat_history(messages) + guideline_task = process_guideline( + guidelines=guidelines, + chat_history=chat_history, + terms=terms_analysis, + language=language, + model_name=model_name, + api_key=api_key, + model_server=model_server + ) + tasks.append(guideline_task) # 添加agent创建任务 agent_task = agent_manager.get_or_create_agent( @@ -172,60 +147,8 @@ async def process_guidelines_and_terms( # 处理结果 agent = all_results[-1] # agent创建的结果 - batch_results = all_results[:-1] # guideline批次的结果 - - # 合并guideline分析结果 - all_checks = [] - for i, result in enumerate(batch_results): - if isinstance(result, Exception): - logger.error(f"Guideline batch {i} failed: {result}") - continue - if result and isinstance(result, dict) and 'checks' in result: - applicable_checks = [check for check in result['checks']] - all_checks.extend(applicable_checks) - elif result and isinstance(result, str) and result.strip(): - logger.info(f"Non-JSON result from batch {i}: {result}") - - if all_checks: - # 首先创建guideline_id到action字段的映射，同时支持string和int类型的guideline_id - guideline_map = {} - for guideline in guidelines_list: - guideline_id = guideline.get('guideline_id') - if guideline_id is not None: - # 同时存储字符串和整数类型的键，确保匹配成功 - guideline_map[str(guideline_id)] = guideline - - # 补全all_checks中缺失的action字段 - completed_checks = [] - for check in all_checks: - completed_check = check.copy() - guideline_id = check.get('guideline_id') - - # 如果action字段缺失或为空，从guidelines_list中获取 - if str(guideline_id) in guideline_map: - completed_check['action'] = guideline_map[str(guideline_id)].get('action', "") - completed_check['condition'] = guideline_map[str(guideline_id)].get('condition', "") - - completed_checks.append(completed_check) - - guideline_reasoning = "\n".join([item["rationale"] for item in completed_checks]) - - # 使用补全后的checks生成active_guidelines，添加安全检查 - active_guidelines_parts = [] - for item in completed_checks: - if 'action' in item and item['action']: - active_guidelines_parts.append( - "Condition:" + item["condition"] + "\nRationale:" + item["rationale"] + "\nAction:" + item["action"] - ) - else: - # 如果仍然缺少action字段，使用默认值 - active_guidelines_parts.append( - "Condition:" + item["condition"] + "\nRationale:" + item["rationale"] + "\nAction:无具体操作要求" - ) - active_guidelines = "\n".join(active_guidelines_parts) - - logger.info(f"Guideline analysis completed: {len(guideline_reasoning)} chars, processed {len(completed_checks)} checks") - + guideline_reasoning = all_results[0] # guideline批次的结果 + logger.info(f"Guideline analysis completed: {len(guideline_reasoning)} chars") else: # 没有guidelines，直接创建agent agent = await agent_manager.get_or_create_agent( @@ -242,7 +165,7 @@ async def process_guidelines_and_terms( user_identifier=user_identifier ) - return agent, processed_system_prompt, guideline_reasoning, active_guidelines + return agent, processed_system_prompt, guideline_reasoning async def enhanced_generate_stream_response( @@ -307,7 +230,7 @@ async def enhanced_generate_stream_response( logger.error(f"Error generating preamble text: {e}") # 等待guideline分析任务完成 - agent, system_prompt, guideline_reasoning, active_guidelines = await guidelines_task + agent, system_prompt, guideline_reasoning = await guidelines_task # 立即发送guideline_reasoning if guideline_reasoning: @@ -319,8 +242,8 @@ async def enhanced_generate_stream_response( final_messages = messages.copy() final_messages = append_user_last_message(final_messages, f"\n\nlanguage:{get_language_text(language)}") - if active_guidelines: - final_messages = append_user_last_message(final_messages, f"\n\nActive Guidelines:\n{active_guidelines}\nPlease follow these guidelines in your response.") + if guideline_reasoning: + final_messages = append_user_last_message(final_messages, f"\n\nGuidelines:\n{guideline_reasoning}\nPlease follow these guidelines in your response.") # 第三阶段：agent响应流式传输 logger.info(f"Starting agent stream response") @@ -413,7 +336,7 @@ async def create_agent_and_generate_response( _, system_prompt = get_preamble_text(language, system_prompt) # 使用公共函数处理所有逻辑 - agent, system_prompt, guideline_reasoning, active_guidelines = await process_guidelines_and_terms( + agent, system_prompt, guideline_reasoning = await process_guidelines_and_terms( bot_id=bot_id, api_key=api_key, model_name=model_name, @@ -433,9 +356,8 @@ async def create_agent_and_generate_response( final_messages = messages.copy() final_messages = append_user_last_message(final_messages, f"\n\nlanguage:{get_language_text(language)}") pre_message_list = [] - if active_guidelines: - final_messages = append_user_last_message(final_messages, f"\n\nActive Guidelines:\n{active_guidelines}\nPlease follow these guidelines in your response.") if guideline_reasoning: + final_messages = append_user_last_message(final_messages, f"\n\nGuidelines:\n{guideline_reasoning}\nPlease follow these guidelines in your response.") pre_message_list.append({"role": "assistant","reasoning_content": guideline_reasoning+ "\n"}) # 非流式响应 diff --git a/utils/fastapi_utils.py b/utils/fastapi_utils.py index bf12736..6dc05f7 100644 --- a/utils/fastapi_utils.py +++ b/utils/fastapi_utils.py @@ -284,19 +284,21 @@ def format_messages_to_chat_history(messages: List[Dict[str, str]]) -> str: Returns: str: 格式化的聊天记录 """ + # 只取最后的6句消息 chat_history = [] for message in messages: role = message.get('role', '') content = message.get('content', '') + if len(content) > 0: + if role == 'user': + chat_history.append(f"user: {content}") + elif role == 'assistant': + chat_history.append(f"assistant: {content}") - if role == 'user': - chat_history.append(f"user: {content}") - elif role == 'assistant': - chat_history.append(f"assistant: {content}") - # 忽略其他角色（如function等） - - return "\n".join(chat_history) + recent_chat_history = chat_history[-6:] if len(chat_history) > 6 else chat_history + print("\n".join(recent_chat_history)) + return "\n".join(recent_chat_history) def create_project_directory(dataset_ids: Optional[List[str]], bot_id: str, robot_type: str = "general_agent") -> Optional[str]: @@ -602,8 +604,8 @@ def _get_optimal_batch_size(guidelines_count: int) -> int: return 5 -async def process_guideline_batch( - guidelines_batch: List[str], +async def process_guideline( + guidelines: str, chat_history: str, terms: str, language: str, @@ -616,32 +618,8 @@ async def process_guideline_batch( for attempt in range(max_retries): try: - # 调用LLM分析这批guidelines - batch_guidelines_text = "\n".join(guidelines_batch) logger.info(f"Start processed guideline batch on attempt {attempt + 1}") - batch_analysis = await call_guideline_llm(chat_history, batch_guidelines_text, terms, language, model_name, api_key, model_server) - - # 从响应中提取 ```json 和 ``` 包裹的内容 - json_pattern = r'```json\s*\n(.*?)\n```' - json_matches = re.findall(json_pattern, batch_analysis, re.DOTALL) - - if json_matches: - try: - # 解析第一个找到的JSON对象 - json_data = json.loads(json_matches[0]) - logger.info(f"Successfully processed guideline batch on attempt {attempt + 1}") - return json_data # 返回解析后的JSON对象 - except json.JSONDecodeError as e: - logger.error(f"Error parsing JSON from guideline analysis on attempt {attempt + 1}: {e}") - if attempt == max_retries - 1: - return batch_analysis # 最后一次尝试失败，返回原始文本 - continue - else: - logger.warning(f"No JSON format found in guideline analysis on attempt {attempt + 1}") - if attempt == max_retries - 1: - return batch_analysis # 最后一次尝试失败，返回原始文本 - continue - + return await call_guideline_llm(chat_history, guidelines, terms, language, model_name, api_key, model_server) except Exception as e: logger.error(f"Error processing guideline batch on attempt {attempt + 1}: {e}") if attempt == max_retries - 1: @@ -664,7 +642,7 @@ def extract_block_from_system_prompt(system_prompt: Optional[str]) -> tuple[str, if not system_prompt: return "", [], [] - guidelines_list = [] + guidelines = "" terms_list = [] # 首先分割所有的代码块 @@ -675,12 +653,7 @@ def extract_block_from_system_prompt(system_prompt: Optional[str]) -> tuple[str, block_type, content = match.groups() if block_type == 'guideline': - try: - guidelines = parse_guidelines_text(content.strip()) - guidelines_list.extend(guidelines) - blocks_to_remove.append(match.group(0)) - except Exception as e: - logger.error(f"Error parsing guidelines: {e}") + guidelines = content.strip() elif block_type == 'terms': try: @@ -698,7 +671,7 @@ def extract_block_from_system_prompt(system_prompt: Optional[str]) -> tuple[str, # 清理多余的空行 cleaned_prompt = re.sub(r'\n\s*\n\s*\n', '\n\n', cleaned_prompt).strip() - return cleaned_prompt, guidelines_list, terms_list + return cleaned_prompt, guidelines, terms_list def parse_guidelines_text(text: str) -> List[Dict[str, Any]]: