zhuchaowe/qwen_agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add cot

Browse Source
This commit is contained in:
朱潮 2025-12-01 21:52:14 +08:00
parent 06404274eb
commit 08b7d0e2b8
3 changed files with 93 additions and 275 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

189
prompt/guideline_prompt.md
View File

@ -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": "<example-id-for-few-shots--do-not-use-this-output>",
"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": "<example-id-for-few-shots--do-not-use-this-output>",
"rationale": "In the most recent message the customer mentions that they work full time which is a constraint"
},
{
"guideline_id": "<example-id-for-few-shots--do-not-use-this-output>",
"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": "<example-id-for-few-shots--do-not-use-this-in-output>",
"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": "<example-id-for-few-shots--do-not-use-this-in-output>",
"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": "<Explain why the conditions are met and what action should be taken>"
}
]
}
```
If no guidelines apply, return an empty checks array:
```json
{
"checks": []
}
```
Rationale Text Language:
## 语言要求 (Language)
{language}
---
## 输出格式
请按照以下结构输出你的思考结果：
### 🎯 核心目标
[简洁描述用户的主要目标]
### 📋 现状分析
- **关键信息**: [从聊天记录中提取的重要信息]
- **限制条件**: [需要遵守的规则和约束]
- **可用资源**: [可以利用的工具和资源]
### 🧠 任务拆解
1. [第一步的具体行动]
2. [第二步的具体行动]
3. [第三步的具体行动]
- 子步骤 3.1
- 子步骤 3.2
### ⚡ 执行计划
- **立即行动**: [当前应该执行的第一个步骤]
- **后续步骤**: [接下来的行动计划]
- **检查标准**: [如何验证每个步骤的完成]

122
routes/chat.py
View File

@ -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"})
# 非流式响应

57
utils/fastapi_utils.py
View File

@ -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]]: