consult_aurai
Get expert AI guidance for programming problems through multi-turn dialogue. Submit error details and code context to receive analysis and step-by-step solutions.
Instructions
请求上级AI的指导(支持交互对齐机制与多轮对话)
这是核心工具,当本地AI遇到编程问题时调用此工具获取上级AI的指导建议。
🔗 相关工具
sync_context:需要上传文档或代码时使用
📄 上传文章、说明文档(.md/.txt)
💻 上传代码文件(避免内容被截断) ⭐ 重要
将
.py/.js/.json等代码文件复制为.txt后上传
report_progress:执行上级 AI 建议后,使用此工具报告进度并获取下一步指导
get_status:查看当前对话状态、迭代次数、配置信息
💡 重要提示:避免内容被截断
如果 code_snippet 或 context 内容过长,请使用 :
# 步骤 1:将代码文件复制为 .txt
shutil.copy('script.py', 'script.txt')
# 步骤 2:上传文件
sync_context(operation='incremental', files=['script.txt'])
# 步骤 3:告诉上级顾问文件已上传
consult_aurai(
error_message='请审查已上传的 script.txt 文件'
)优势:
✅ 避免代码在
context或answers_to_questions字段中被截断✅ 利用文件读取机制,完整传递内容
✅ 支持任意大小的代码文件
[重要] 何时开始新对话?
系统会自动检测,但你也可以手动控制:
自动清空:当上一次对话返回
resolved=true时,系统会自动清空历史手动清空:如果你要讨论一个完全不同的新问题,设置
is_new_question=true
何时设置
[OK] 切换到完全不相关的项目/文件
[OK] 之前的问题已解决,现在遇到全新的问题
[OK] 发现上下文混乱,想重新开始
[X] 不要在同一个问题的多轮对话中使用
交互协议
1. 多轮对齐机制
不要期待一次成功:上级顾问可能会认为信息不足,返回反问问题
仔细阅读
questions_to_answer中的每个问题主动搜集信息(读取文件、检查日志、运行命令)
再次调用 此工具,将答案填入
answers_to_questions参数
2. 首次调用
必须提供:
problem_type:问题类型(runtime_error/syntax_error/design_issue/other)error_message:清晰描述问题或错误context:相关上下文(代码片段、环境信息、已尝试的方案)code_snippet:相关代码(如果有)
3. 后续调用(当返回 status="need_info" 时)
必须提供:
answers_to_questions:对上级顾问反问的详细回答保持其他参数不变(除非有新信息)
4. 诚实原则
禁止瞎编:如果不知道答案,诚实说明"未找到相关信息"
禁止臆测:不要在没有证据的情况下假设解决方案
提供具体证据(文件路径、日志内容、错误堆栈)
响应格式
信息不足时 (status="need_info")
{
"status": "need_info",
"questions_to_answer": ["问题1", "问题2"],
"instruction": "请搜集信息并再次调用"
}提供指导时 (status="success")
{
"status": "success",
"analysis": "问题分析",
"guidance": "解决建议",
"action_items": ["步骤1", "步骤2"],
"resolved": false // 是否已完全解决
}问题解决后
当 resolved=true 时,对话历史会自动清空,下次查询将开始新对话。
[自动] 新对话检测
系统会自动检测新问题:
如果上一次对话的
resolved=true,下次调用consult_aurai时会自动清空历史保证每个独立问题都有干净的上下文,避免干扰
[重要] 明确标注新问题(可选参数)
如果你想强制开始一个新对话,可以设置 is_new_question=true:
效果:立即清空所有之前的对话历史
后果:上级AI将无法看到之前的任何上下文
使用场景:
之前的对话已完全无关
想重新开始讨论一个全新的问题
发现上下文混乱,想重置
示例:
# 第一次咨询(问题A)
consult_aurai(problem_type="runtime_error", error_message="...")
# 继续讨论问题A...
consult_aurai(answers_to_questions="...")
# 切换到问题B(标注为新问题,清空历史)
consult_aurai(
problem_type="design_issue",
error_message="...",
is_new_question=True # [注意] 会清空之前关于问题A的所有对话
)Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| problem_type | Yes | 问题类型: runtime_error, syntax_error, design_issue, other | |
| error_message | Yes | 错误描述 | |
| code_snippet | No | 相关代码片段 | |
| context | No | 上下文信息(支持 JSON 字符串或字典,会自动解析) | |
| attempts_made | No | 已尝试的解决方案 | |
| answers_to_questions | No | 对上级顾问反问的回答(仅在多轮对话时使用) | |
| is_new_question | No | [重要] 是否为新问题(新问题会清空之前的所有对话历史,确保干净的上下文) |
Implementation Reference
- src/mcp_aurai/server.py:121-386 (handler)Main handler function for consult_aurai tool - async function that processes consultation requests, manages conversation history, builds prompts, calls the AI client, and returns guidance or follow-up questions based on the AI's response
async def consult_aurai( problem_type: str = Field( description="问题类型: runtime_error, syntax_error, design_issue, other" ), error_message: str = Field(description="错误描述"), code_snippet: str | None = Field(default=None, description="相关代码片段"), context: Any = Field(default=None, description="上下文信息(支持 JSON 字符串或字典,会自动解析)"), attempts_made: str | None = Field(default=None, description="已尝试的解决方案"), answers_to_questions: str | None = Field( default=None, description="对上级顾问反问的回答(仅在多轮对话时使用)" ), is_new_question: bool = Field( default=False, description="[重要] 是否为新问题(新问题会清空之前的所有对话历史,确保干净的上下文)" ), ) -> dict[str, Any]: """ 请求上级AI的指导(支持交互对齐机制与多轮对话) 这是核心工具,当本地AI遇到编程问题时调用此工具获取上级AI的指导建议。 --- **🔗 相关工具** - **sync_context**:需要上传文档或代码时使用 - 📄 上传文章、说明文档(.md/.txt) - 💻 **上传代码文件(避免内容被截断)** ⭐ 重要 - 将 `.py/.js/.json` 等代码文件复制为 `.txt` 后上传 - **report_progress**:执行上级 AI 建议后,使用此工具报告进度并获取下一步指导 - **get_status**:查看当前对话状态、迭代次数、配置信息 **💡 重要提示:避免内容被截断** 如果 `code_snippet` 或 `context` 内容过长,**请使用 `sync_context` 上传文件**: ```python # 步骤 1:将代码文件复制为 .txt shutil.copy('script.py', 'script.txt') # 步骤 2:上传文件 sync_context(operation='incremental', files=['script.txt']) # 步骤 3:告诉上级顾问文件已上传 consult_aurai( error_message='请审查已上传的 script.txt 文件' ) ``` **优势**: - ✅ 避免代码在 `context` 或 `answers_to_questions` 字段中被截断 - ✅ 利用文件读取机制,完整传递内容 - ✅ 支持任意大小的代码文件 --- ## [重要] 何时开始新对话? **系统会自动检测**,但你也可以手动控制: - **自动清空**:当上一次对话返回 `resolved=true` 时,系统会自动清空历史 - **手动清空**:如果你要讨论一个完全不同的新问题,设置 `is_new_question=true` **何时设置 `is_new_question=true`?** - [OK] 切换到完全不相关的项目/文件 - [OK] 之前的问题已解决,现在遇到全新的问题 - [OK] 发现上下文混乱,想重新开始 - [X] 不要在同一个问题的多轮对话中使用 ## 交互协议 ### 1. 多轮对齐机制 - **不要期待一次成功**:上级顾问可能会认为信息不足,返回反问问题 - 仔细阅读 `questions_to_answer` 中的每个问题 - 主动搜集信息(读取文件、检查日志、运行命令) - **再次调用** 此工具,将答案填入 `answers_to_questions` 参数 ### 2. 首次调用 必须提供: - `problem_type`:问题类型(runtime_error/syntax_error/design_issue/other) - `error_message`:清晰描述问题或错误 - `context`:相关上下文(代码片段、环境信息、已尝试的方案) - `code_snippet`:相关代码(如果有) ### 3. 后续调用(当返回 status="need_info" 时) 必须提供: - `answers_to_questions`:对上级顾问反问的详细回答 - 保持其他参数不变(除非有新信息) ### 4. 诚实原则 - **禁止瞎编**:如果不知道答案,诚实说明"未找到相关信息" - **禁止臆测**:不要在没有证据的情况下假设解决方案 - 提供具体证据(文件路径、日志内容、错误堆栈) ## 响应格式 ### 信息不足时 (status="need_info") ```json { "status": "need_info", "questions_to_answer": ["问题1", "问题2"], "instruction": "请搜集信息并再次调用" } ``` ### 提供指导时 (status="success") ```json { "status": "success", "analysis": "问题分析", "guidance": "解决建议", "action_items": ["步骤1", "步骤2"], "resolved": false // 是否已完全解决 } ``` ### 问题解决后 当 `resolved=true` 时,对话历史会自动清空,下次查询将开始新对话。 ### [自动] 新对话检测 系统会自动检测新问题: - 如果上一次对话的 `resolved=true`,下次调用 `consult_aurai` 时会自动清空历史 - 保证每个独立问题都有干净的上下文,避免干扰 ### [重要] 明确标注新问题(可选参数) 如果你想强制开始一个新对话,可以设置 `is_new_question=true`: - **效果**:立即清空所有之前的对话历史 - **后果**:上级AI将无法看到之前的任何上下文 - **使用场景**: - 之前的对话已完全无关 - 想重新开始讨论一个全新的问题 - 发现上下文混乱,想重置 **示例**: ```python # 第一次咨询(问题A) consult_aurai(problem_type="runtime_error", error_message="...") # 继续讨论问题A... consult_aurai(answers_to_questions="...") # 切换到问题B(标注为新问题,清空历史) consult_aurai( problem_type="design_issue", error_message="...", is_new_question=True # [注意] 会清空之前关于问题A的所有对话 ) ``` """ config = get_aurai_config() logger.info(f"收到consult_aurai请求,问题类型: {problem_type},是否新问题: {is_new_question}") # [新问题] 处理新问题:两种方式触发清空历史 # 方式1:明确标注 is_new_question=true # 方式2:自动检测(上一次对话已解决) should_clear_history = False clear_reason = "" if is_new_question: # 明确标注新问题 should_clear_history = True clear_reason = "下级AI明确标注为新问题" elif _conversation_history: # 自动检测:检查上一次对话是否已解决 last_entry = _conversation_history[-1] last_response = last_entry.get("response", {}) if last_response.get("resolved", False): should_clear_history = True clear_reason = "上一次对话已解决(自动检测)" # 执行清空操作 if should_clear_history: history_count = len(_conversation_history) _conversation_history.clear() logger.info(f"[新问题] 清空对话历史(清除 {history_count} 条记录)") logger.info(f" 原因: {clear_reason}") logger.info(f" 新问题: {problem_type} - {error_message[:100]}...") # 解析 context 参数(支持 JSON 字符串或字典) parsed_context: dict[str, Any] = {} if context: if isinstance(context, str): try: parsed_context = json.loads(context) logger.debug("已解析 JSON 格式的 context") except json.JSONDecodeError as e: logger.warning(f"context JSON 解析失败: {e},使用空字典") parsed_context = {} elif isinstance(context, dict): parsed_context = context # 构建提示词(如果有对反问的回答,加入上下文) current_context = parsed_context or {} if answers_to_questions: current_context["answers_to_questions"] = answers_to_questions prompt = build_consult_prompt( problem_type=problem_type, error_message=error_message, code_snippet=code_snippet, context=current_context, attempts_made=attempts_made, iteration=len(_conversation_history), conversation_history=_get_history(), ) # 调用上级AI,传递对话历史 client = get_aurai_client() response = await client.chat( user_message=prompt, conversation_history=_get_history() ) # 记录到历史 _add_to_history({ "type": "consult", "problem_type": problem_type, "error_message": error_message, "response": response, "had_answers": answers_to_questions is not None, }) # 根据上级顾问的响应状态返回不同格式 if response.get("status") == "aligning": # 模式 A: 信息不足,需要补充 logger.info(f"上级顾问要求补充信息,问题数: {len(response.get('questions', []))}") return { "status": "need_info", "message": "[提示] 上级顾问认为信息不足,请回答以下问题:", "questions_to_answer": response.get("questions", []), "instruction": "请搜集信息,再次调用 consult_aurai,并将答案填入 'answers_to_questions' 字段。", # ⭐ 相关工具提示 "related_tools_hint": { "sync_context": { "description": "如果需要上传文档(.md/.txt)来补充上下文信息", "example": "sync_context(operation='full_sync', files=['path/to/doc.md'])" } } } else: # 模式 B: 信息充足,提供指导 logger.info(f"上级顾问提供指导,resolved: {response.get('resolved', False)}") # 检查问题是否已解决,若解决则清空对话历史 if response.get("resolved", False): history_count = len(_conversation_history) _conversation_history.clear() logger.info(f"[完成] 问题已解决,已清空对话历史(清除了 {history_count} 条记录)") return { "status": "success", "analysis": response.get("analysis"), "guidance": response.get("guidance"), "action_items": response.get("action_items", []), "code_changes": response.get("code_changes", []), "verification": response.get("verification"), "needs_another_iteration": response.get("needs_another_iteration", False), "resolved": response.get("resolved", False), "requires_human_intervention": response.get("requires_human_intervention", False), "hint": "[提示] 如需咨询新问题,下次调用时设置 is_new_question=true。这将清空之前的所有对话历史(包括之前的问题和上级AI的指导),但当前这条新问题会正常处理并保留在新的对话中", } - src/mcp_aurai/server.py:120-120 (registration)Tool registration using @mcp.tool() decorator from FastMCP framework - registers consult_aurai as an MCP tool
@mcp.tool() - src/mcp_aurai/server.py:122-137 (schema)Input schema definition using Field validators - defines parameters (problem_type, error_message, code_snippet, context, attempts_made, answers_to_questions, is_new_question) with descriptions and validation
problem_type: str = Field( description="问题类型: runtime_error, syntax_error, design_issue, other" ), error_message: str = Field(description="错误描述"), code_snippet: str | None = Field(default=None, description="相关代码片段"), context: Any = Field(default=None, description="上下文信息(支持 JSON 字符串或字典,会自动解析)"), attempts_made: str | None = Field(default=None, description="已尝试的解决方案"), answers_to_questions: str | None = Field( default=None, description="对上级顾问反问的回答(仅在多轮对话时使用)" ), is_new_question: bool = Field( default=False, description="[重要] 是否为新问题(新问题会清空之前的所有对话历史,确保干净的上下文)" ), ) -> dict[str, Any]: - src/mcp_aurai/prompts.py:6-133 (helper)build_consult_prompt helper function - constructs the prompt sent to the upper-level AI advisor, including problem information, context, code snippets, conversation history, and response format instructions
def build_consult_prompt( problem_type: str, error_message: str, code_snippet: str | None = None, context: dict[str, Any] | None = None, attempts_made: str | None = None, iteration: int = 0, conversation_history: list[dict[str, str]] | None = None, ) -> str: """ 构建请求上级AI指导的提示词 Args: problem_type: 问题类型 error_message: 错误描述 code_snippet: 相关代码片段 context: 上下文信息 attempts_made: 已尝试的解决方案 iteration: 当前迭代次数 conversation_history: 对话历史 """ context = context or {} # 构建上下文描述 context_desc = [] if file_path := context.get("file_path"): context_desc.append(f"- 文件路径: {file_path}") if line_number := context.get("line_number"): context_desc.append(f"- 行号: {line_number}") if terminal_output := context.get("terminal_output"): context_desc.append(f"- 终端输出:\n```\n{terminal_output}\n```") # 构建对话历史 history_desc = "" if conversation_history: history_desc = "\n## 对话历史\n\n" for i, turn in enumerate(conversation_history[-5:], 1): # 只保留最近5轮 history_desc += f"### 第{i}轮\n" if "action" in turn: history_desc += f"**执行操作**: {turn['action']}\n" if "result" in turn: history_desc += f"**执行结果**: {turn['result']}\n" history_desc += "\n" prompt = f"""# 你是上级AI顾问 ## 角色 你是一位经验丰富的技术顾问,正在指导一位"本地AI助手"解决编程问题。 ## 任务 分析以下问题,提供清晰、可执行的指导。 ## 问题信息 - **问题类型**: {problem_type} - **错误描述**: {error_message} - **当前迭代**: 第 {iteration + 1} 轮 ## 上下文 {chr(10).join(context_desc) if context_desc else "无"} ## 代码片段 {f"```{context.get('language', 'python')}\n{code_snippet}\n```" if code_snippet else "无"} ## 已尝试方案 {attempts_made if attempts_made else "无"} {history_desc} ## 你的回应格式 **重要**: 首先评估信息完整度,然后选择对应的模式返回。 **信息不足时(缺少具体错误、代码、上下文)**: ```json {{ "status": "aligning", "questions": ["需补充的问题1", "需补充的问题2"], "analysis": null, "guidance": null }} ``` **信息充足时**: ```json {{ "status": "guiding", "questions": [], "analysis": "问题分析 - 简明扼要分析问题根源", "guidance": "指导建议 - 具体建议的文字描述", "action_items": ["步骤1", "步骤2", "..."], "code_changes": [ {{ "file": "文件路径", "line": 行号, "old": "原代码", "new": "新代码" }} ], "verification": "验证方法", "needs_another_iteration": false, "resolved": false, "requires_human_intervention": false }} ``` ## 字段说明 - **status**: 必须是 "aligning"(信息不足)或 "guiding"(信息充足) - **questions**: 反问列表(仅在 aligning 模式) - **analysis**: 分析问题根本原因 - **guidance**: 给出具体、可执行的指导建议 - **action_items**: 数组,列出具体的执行步骤 - **code_changes**: 数组(可选),如果需要修改代码,列出具体的代码变更 - **verification**: 验证方法(新增) - **needs_another_iteration**: 布尔值,是否需要继续迭代 - **resolved**: 布尔值,问题是否已解决 - **requires_human_intervention**: 布尔值,是否需要人工介入 ## 重要原则 1. **指导要具体、可执行** - 避免模糊建议 2. **最小改动原则** - 优先考虑最小改动来解决问题 3. **分步骤指导** - 如果问题复杂,分步骤给出建议 4. **承认限制** - 承认无法解决的问题,及时设置 requires_human_intervention=true 5. **避免无限循环** - 如果连续多次建议相同方向,考虑人工介入 现在,请分析上述问题并给出你的指导。 """ return prompt - src/mcp_aurai/llm.py:176-258 (helper)AuraiClient.chat async method - handles communication with the OpenAI-compatible API, sends messages with conversation history, parses JSON responses, and handles errors
async def chat( self, user_message: str, system_prompt: str | None = None, response_format: Literal["text", "json_object"] = "json_object", conversation_history: list[dict] | None = None, ) -> dict: """ 发送聊天请求 Args: user_message: 用户消息 system_prompt: 系统提示词 response_format: 响应格式 conversation_history: 对话历史(用于多轮对话) Returns: 解析后的JSON响应 """ from .prompts import SYSTEM_PROMPT system_prompt = system_prompt or SYSTEM_PROMPT messages = [] if system_prompt: messages.append({"role": "system", "content": system_prompt}) # 添加对话历史 history_messages = self._build_messages_from_history(conversation_history) messages.extend(history_messages) # 添加当前用户消息 messages.append({"role": "user", "content": user_message}) logger.info(f"发送请求到 {self.config.base_url},消息数: {len(messages)}") try: response = self._client.chat.completions.create( model=self.config.model, messages=messages, temperature=self.config.temperature, max_tokens=self.config.max_tokens, ) content = response.choices[0].message.content logger.info(f"收到响应,长度: {len(content)}") # 尝试解析JSON try: # 清理可能存在的markdown代码块标记 content_clean = content.strip() if content_clean.startswith("```json"): content_clean = content_clean[7:] if content_clean.startswith("```"): content_clean = content_clean[3:] if content_clean.endswith("```"): content_clean = content_clean[:-3] content_clean = content_clean.strip() result = json.loads(content_clean) logger.info("成功解析JSON响应") return result except json.JSONDecodeError as e: logger.warning(f"JSON解析失败: {e},返回原始文本") return { "analysis": "解析失败", "guidance": content, "action_items": [], "needs_another_iteration": False, "resolved": False, "requires_human_intervention": True, } except Exception as e: logger.error(f"API请求失败: {e}") return { "analysis": f"请求失败: {str(e)}", "guidance": "请检查API密钥、Base URL和网络连接", "action_items": [], "needs_another_iteration": False, "resolved": False, "requires_human_intervention": True, }