+
+[](https://www.python.org/)
+[](https://github.com/datawhalechina/hello-agents)
+[](LICENSE)
+
+
+
+## 🎯 项目亮点
+
+- 本地仓库优先：围绕本地代码库分析、修改、验证设计，不依赖远端 SaaS 工作流。
+- 安全修改闭环：通过标准补丁格式执行代码修改，落盘前支持确认与备份。
+- 双界面体验：CLI 适合快速问答，TUI 适合长会话和过程观察。
+- 扩展能力完整：不仅支持内置工具，还支持 Skills 和 MCP 两类扩展机制。
+- 工程化更完整：包含日志、会话导出、计划生成、Todo 跟踪等能力。
+
+## 📊 性能评估
+
+当前项目以功能完整性和交互体验为主，尚未形成统一的量化 benchmark，现阶段可确认的结果包括：
+
+- 已完成 CLI 与 TUI 两套可运行入口
+- 已具备本地代码仓库分析与补丁执行闭环
+- 已支持会话日志、导出、Todo、Skills 与 MCP 扩展能力
+- 后续可补充任务成功率、平均响应时间和补丁应用成功率等指标
+
+## 🔮 未来计划
+
+- [ ] 支持会话恢复与断点续传
+- [ ] 继续细化终端工具为更原子的命令工具
+- [ ] 重构 Note Tool 与 Memory Tool 的交互方式
+- [ ] 完善测试用例与自动化验证流程
+- [ ] 增加更多可直接启用的 MCP 工具模板
+
+## 🤝 贡献指南
+
+欢迎提出 Issue 和 Pull Request。
+
+## 📄 许可证
+
+MIT License
+
+## 👤 作者
+
+- GitHub: [@aug618](https://github.com/aug618)
+- 二次开发来源：YYHDBL-HelloCodeAgentCli
+- 上游项目仓库：https://github.com/aug618/Praxis
+
+## 🙏 致谢
+
+感谢 Datawhale 社区和 Hello-Agents 项目。
+
+同时感谢 YYHDBL-HelloCodeAgentCli 项目为本项目提供二次开发基础。
\ No newline at end of file
diff --git a/Co-creation-projects/aug618-Praxis/__init__.py b/Co-creation-projects/aug618-Praxis/__init__.py
new file mode 100644
index 00000000..c38a0339
--- /dev/null
+++ b/Co-creation-projects/aug618-Praxis/__init__.py
@@ -0,0 +1,66 @@
+"""
+HelloAgents - 灵活、可扩展的多智能体框架
+
+基于OpenAI原生API构建，提供简洁高效的智能体开发体验。
+"""
+
+# 配置第三方库的日志级别，减少噪音
+import logging
+logging.getLogger("httpx").setLevel(logging.WARNING)
+logging.getLogger("qdrant_client").setLevel(logging.WARNING)
+logging.getLogger("urllib3").setLevel(logging.WARNING)
+logging.getLogger("neo4j").setLevel(logging.WARNING)
+logging.getLogger("neo4j.notifications").setLevel(logging.WARNING)
+
+# from .version import __version__, __author__, __email__, __description__
+
+# 核心组件
+from .core.llm import HelloAgentsLLM
+from .core.config import Config
+from .core.message import Message
+from .core.exceptions import HelloAgentsException
+
+# Agent实现
+from .agents.simple_agent import SimpleAgent
+from .agents.react_agent import ReActAgent
+from .agents.reflection_agent import ReflectionAgent
+from .agents.plan_solve_agent import PlanAndSolveAgent
+
+# 工具系统
+from .tools.registry import ToolRegistry, global_registry
+from .tools.builtin.search import SearchTool, search
+# from .tools.builtin.calculator import CalculatorTool, calculate
+from .tools.chain import ToolChain, ToolChainManager
+from .tools.async_executor import AsyncToolExecutor
+
+__all__ = [
+ # 版本信息
+ "__version__",
+ "__author__",
+ "__email__",
+ "__description__",
+
+ # 核心组件
+ "HelloAgentsLLM",
+ "Config",
+ "Message",
+ "HelloAgentsException",
+
+ # Agent范式
+ "SimpleAgent",
+ "ReActAgent",
+ "ReflectionAgent",
+ "PlanAndSolveAgent",
+
+ # 工具系统
+ "ToolRegistry",
+ "global_registry",
+ "SearchTool",
+ "search",
+ # "CalculatorTool",
+ # "calculate",
+ "ToolChain",
+ "ToolChainManager",
+ "AsyncToolExecutor",
+]
+
diff --git a/Co-creation-projects/aug618-Praxis/agents/__init__.py b/Co-creation-projects/aug618-Praxis/agents/__init__.py
new file mode 100644
index 00000000..32f62de8
--- /dev/null
+++ b/Co-creation-projects/aug618-Praxis/agents/__init__.py
@@ -0,0 +1,7 @@
+"""Agent implementations.
+
+Keep this module lightweight: avoid importing optional dependencies at import time.
+Import concrete agents directly, e.g. `from agents.simple_agent import SimpleAgent`.
+"""
+
+__all__ = ["SimpleAgent", "ReActAgent", "ReflectionAgent", "PlanAndSolveAgent"]
diff --git a/Co-creation-projects/aug618-Praxis/agents/plan_solve_agent.py b/Co-creation-projects/aug618-Praxis/agents/plan_solve_agent.py
new file mode 100644
index 00000000..96dc36c9
--- /dev/null
+++ b/Co-creation-projects/aug618-Praxis/agents/plan_solve_agent.py
@@ -0,0 +1,201 @@
+"""Plan and Solve Agent实现 - 分解规划与逐步执行的智能体"""
+
+import ast
+from typing import Optional, List, Dict
+from core.agent import Agent
+from core.llm import HelloAgentsLLM
+from core.config import Config
+from core.message import Message
+
+# 默认规划器提示词模板
+DEFAULT_PLANNER_PROMPT = """
+你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。
+请确保计划中的每个步骤都是一个独立的、可执行的子任务，并且严格按照逻辑顺序排列。
+你的输出必须是一个Python列表，其中每个元素都是一个描述子任务的字符串。
+
+问题: {question}
+
+请严格按照以下格式输出你的计划:
+```python
+["步骤1", "步骤2", "步骤3", ...]
+```
+"""
+
+# 默认执行器提示词模板
+DEFAULT_EXECUTOR_PROMPT = """
+你是一位顶级的AI执行专家。你的任务是严格按照给定的计划，一步步地解决问题。
+你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。
+请你专注于解决"当前步骤"，并仅输出该步骤的最终答案，不要输出任何额外的解释或对话。
+
+# 原始问题:
+{question}
+
+# 完整计划:
+{plan}
+
+# 历史步骤与结果:
+{history}
+
+# 当前步骤:
+{current_step}
+
+请仅输出针对"当前步骤"的回答:
+"""
+
+class Planner:
+ """规划器 - 负责将复杂问题分解为简单步骤"""
+
+ def __init__(self, llm_client: HelloAgentsLLM, prompt_template: Optional[str] = None):
+ self.llm_client = llm_client
+ self.prompt_template = prompt_template if prompt_template else DEFAULT_PLANNER_PROMPT
+
+ def plan(self, question: str, **kwargs) -> List[str]:
+ """
+ 生成执行计划
+
+ Args:
+ question: 要解决的问题
+ **kwargs: LLM调用参数
+
+ Returns:
+ 步骤列表
+ """
+ prompt = self.prompt_template.format(question=question)
+ messages = [{"role": "user", "content": prompt}]
+
+ print("--- 正在生成计划 ---")
+ response_text = self.llm_client.invoke(messages, **kwargs) or ""
+ print(f"✅ 计划已生成:\n{response_text}")
+
+ try:
+ # 提取Python代码块中的列表
+ plan_str = response_text.split("```python")[1].split("```")[0].strip()
+ plan = ast.literal_eval(plan_str)
+ return plan if isinstance(plan, list) else []
+ except (ValueError, SyntaxError, IndexError) as e:
+ print(f"❌ 解析计划时出错: {e}")
+ print(f"原始响应: {response_text}")
+ return []
+ except Exception as e:
+ print(f"❌ 解析计划时发生未知错误: {e}")
+ return []
+
+class Executor:
+ """执行器 - 负责按计划逐步执行"""
+
+ def __init__(self, llm_client: HelloAgentsLLM, prompt_template: Optional[str] = None):
+ self.llm_client = llm_client
+ self.prompt_template = prompt_template if prompt_template else DEFAULT_EXECUTOR_PROMPT
+
+ def execute(self, question: str, plan: List[str], **kwargs) -> str:
+ """
+ 按计划执行任务
+
+ Args:
+ question: 原始问题
+ plan: 执行计划
+ **kwargs: LLM调用参数
+
+ Returns:
+ 最终答案
+ """
+ history = ""
+ final_answer = ""
+
+ print("\n--- 正在执行计划 ---")
+ for i, step in enumerate(plan, 1):
+ print(f"\n-> 正在执行步骤 {i}/{len(plan)}: {step}")
+ prompt = self.prompt_template.format(
+ question=question,
+ plan=plan,
+ history=history if history else "无",
+ current_step=step
+ )
+ messages = [{"role": "user", "content": prompt}]
+
+ response_text = self.llm_client.invoke(messages, **kwargs) or ""
+
+ history += f"步骤 {i}: {step}\n结果: {response_text}\n\n"
+ final_answer = response_text
+ print(f"✅ 步骤 {i} 已完成，结果: {final_answer}")
+
+ return final_answer
+
+class PlanAndSolveAgent(Agent):
+ """
+ Plan and Solve Agent - 分解规划与逐步执行的智能体
+
+ 这个Agent能够：
+ 1. 将复杂问题分解为简单步骤
+ 2. 按照计划逐步执行
+ 3. 维护执行历史和上下文
+ 4. 得出最终答案
+
+ 特别适合多步骤推理、数学问题、复杂分析等任务。
+ """
+
+ def __init__(
+ self,
+ name: str,
+ llm: HelloAgentsLLM,
+ system_prompt: Optional[str] = None,
+ config: Optional[Config] = None,
+ custom_prompts: Optional[Dict[str, str]] = None
+ ):
+ """
+ 初始化PlanAndSolveAgent
+
+ Args:
+ name: Agent名称
+ llm: LLM实例
+ system_prompt: 系统提示词
+ config: 配置对象
+ custom_prompts: 自定义提示词模板 {"planner": "", "executor": ""}
+ """
+ super().__init__(name, llm, system_prompt, config)
+
+ # 设置提示词模板：用户自定义优先，否则使用默认模板
+ if custom_prompts:
+ planner_prompt = custom_prompts.get("planner")
+ executor_prompt = custom_prompts.get("executor")
+ else:
+ planner_prompt = None
+ executor_prompt = None
+
+ self.planner = Planner(self.llm, planner_prompt)
+ self.executor = Executor(self.llm, executor_prompt)
+
+ def run(self, input_text: str, **kwargs) -> str:
+ """
+ 运行Plan and Solve Agent
+
+ Args:
+ input_text: 要解决的问题
+ **kwargs: 其他参数
+
+ Returns:
+ 最终答案
+ """
+ print(f"\n🤖 {self.name} 开始处理问题: {input_text}")
+
+ # 1. 生成计划
+ plan = self.planner.plan(input_text, **kwargs)
+ if not plan:
+ final_answer = "无法生成有效的行动计划，任务终止。"
+ print(f"\n--- 任务终止 ---\n{final_answer}")
+
+ # 保存到历史记录
+ self.add_message(Message(input_text, "user"))
+ self.add_message(Message(final_answer, "assistant"))
+
+ return final_answer
+
+ # 2. 执行计划
+ final_answer = self.executor.execute(input_text, plan, **kwargs)
+ print(f"\n--- 任务完成 ---\n最终答案: {final_answer}")
+
+ # 保存到历史记录
+ self.add_message(Message(input_text, "user"))
+ self.add_message(Message(final_answer, "assistant"))
+
+ return final_answer
diff --git a/Co-creation-projects/aug618-Praxis/agents/react_agent.py b/Co-creation-projects/aug618-Praxis/agents/react_agent.py
new file mode 100644
index 00000000..4a022b39
--- /dev/null
+++ b/Co-creation-projects/aug618-Praxis/agents/react_agent.py
@@ -0,0 +1,465 @@
+"""ReAct Agent实现 - 推理与行动结合的智能体"""
+
+import re
+from typing import Optional, List, Tuple, Callable, Dict, Any
+import json
+import os
+from core.agent import Agent
+from core.llm import HelloAgentsLLM
+from core.config import Config
+from core.message import Message
+from tools.registry import ToolRegistry
+from utils.cli_ui import Spinner, c, PRIMARY, ACCENT, INFO, hr, log_tool_event, clamp_text
+
+# 默认ReAct提示词模板
+DEFAULT_REACT_PROMPT = """你是一个具备推理和行动能力的AI助手。你可以通过思考分析问题，然后调用合适的工具来获取信息，最终给出准确的答案。
+
+## 可用工具
+{tools}
+
+## 工作流程
+请严格按照以下格式进行回应，每次只能执行一个步骤：
+
+**Thought:** 分析当前问题，思考需要什么信息或采取什么行动。
+**Action:** 选择一个行动，格式必须是以下之一：
+- `{{tool_name}}[{{tool_input}}]` - 调用指定工具
+- `Finish[最终答案]` - 当你有足够信息给出最终答案时
+
+## 重要提醒
+1. 每次回应必须包含Thought和Action两部分
+2. 工具调用的格式必须严格遵循：工具名[参数]
+3. 只有当你确信有足够信息回答问题时，才使用Finish
+4. 如果工具返回的信息不够，继续使用其他工具或相同工具的不同参数
+
+## 当前任务
+**Question:** {question}
+
+## 执行历史
+{history}
+
+现在开始你的推理和行动："""
+
+class ReActAgent(Agent):
+ """
+ ReAct (Reasoning and Acting) Agent
+
+ 结合推理和行动的智能体，能够：
+ 1. 分析问题并制定行动计划
+ 2. 调用外部工具获取信息
+ 3. 基于观察结果进行推理
+ 4. 迭代执行直到得出最终答案
+
+ 这是一个经典的Agent范式，特别适合需要外部信息的任务。
+ """
+
+ def __init__(
+ self,
+ name: str,
+ llm: HelloAgentsLLM,
+ tool_registry: Optional[ToolRegistry] = None,
+ system_prompt: Optional[str] = None,
+ config: Optional[Config] = None,
+ max_steps: int = 5,
+ custom_prompt: Optional[str] = None,
+ observation_summarizer: Optional[Callable[[str, str, str], str]] = None,
+ summarize_threshold_chars: int = 2000,
+ finalize_on_max_steps: bool = True,
+ early_stop_on_repeat: bool = True,
+ repeat_action_threshold: int = 2,
+ ):
+ """
+ 初始化ReActAgent
+
+ Args:
+ name: Agent名称
+ llm: LLM实例
+ tool_registry: 工具注册表（可选，如果不提供则创建空的工具注册表）
+ system_prompt: 系统提示词
+ config: 配置对象
+ max_steps: 最大执行步数
+ custom_prompt: 自定义提示词模板
+ """
+ super().__init__(name, llm, system_prompt, config)
+
+ # 如果没有提供tool_registry，创建一个空的
+ if tool_registry is None:
+ self.tool_registry = ToolRegistry()
+ else:
+ self.tool_registry = tool_registry
+
+ self.max_steps = max_steps
+ self.current_history: List[str] = []
+ self.last_trace: List[Dict[str, Any]] = []
+ self.observation_summarizer = observation_summarizer
+ self.summarize_threshold_chars = summarize_threshold_chars
+ self.finalize_on_max_steps = finalize_on_max_steps
+ self.early_stop_on_repeat = early_stop_on_repeat
+ self.repeat_action_threshold = repeat_action_threshold
+
+ # 设置提示词模板：用户自定义优先，否则使用默认模板
+ self.prompt_template = custom_prompt if custom_prompt else DEFAULT_REACT_PROMPT
+
+ def add_tool(self, tool):
+ """
+ 添加工具到工具注册表
+ 支持MCP工具的自动展开
+
+ Args:
+ tool: 工具实例(可以是普通Tool或MCPTool)
+ """
+ # 检查是否是MCP工具
+ if hasattr(tool, 'auto_expand') and tool.auto_expand:
+ # MCP工具会自动展开为多个工具
+ if hasattr(tool, '_available_tools') and tool._available_tools:
+ for mcp_tool in tool._available_tools:
+ # 创建包装工具
+ from tools.base import Tool
+ wrapped_tool = Tool(
+ name=f"{tool.name}_{mcp_tool['name']}",
+ description=mcp_tool.get('description', ''),
+ func=lambda input_text, t=tool, tn=mcp_tool['name']: t.run({
+ "action": "call_tool",
+ "tool_name": tn,
+ "arguments": {"input": input_text}
+ })
+ )
+ self.tool_registry.register_tool(wrapped_tool)
+ print(f"✅ MCP工具 '{tool.name}' 已展开为 {len(tool._available_tools)} 个独立工具")
+ else:
+ self.tool_registry.register_tool(tool)
+ else:
+ self.tool_registry.register_tool(tool)
+
+ def run(self, input_text: str, **kwargs) -> str:
+ """
+ 运行ReAct Agent
+
+ Args:
+ input_text: 用户问题
+ **kwargs: 其他参数
+
+ Returns:
+ 最终答案
+ """
+ # Optional multimodal attachments (OpenAI-compatible content parts).
+ # Example: [{"type":"image_url","image_url":{"url":"data:image/png;base64,..."}}]
+ attachments = kwargs.pop("attachments", None)
+ self.current_history = []
+ self.last_trace = []
+ current_step = 0
+
+ # Avoid dumping huge stitched prompts to console (CLI UX)
+ preview = input_text.replace("\n", " ")
+ if len(preview) > 160:
+ preview = preview[:160] + "..."
+ # print("\n" + hr("=", 80))
+ # print(c(f"🤖 {self.name}", PRIMARY) + " " + c(f"{preview}", INFO))
+ # print(hr("=", 80))
+
+ repeat_count = 0
+ last_action_sig: Optional[str] = None
+
+ while current_step < self.max_steps:
+ current_step += 1
+ print(c(f"\n--- 当轮需求所处步骤：{current_step}/{self.max_steps} （单次需求最多执行{self.max_steps}步） ---", ACCENT))
+
+ # 构建提示词
+ tools_desc = self.tool_registry.get_tools_description()
+ history_str = "\n".join(self.current_history)
+ prompt = self.prompt_template.format(
+ tools=tools_desc,
+ question=input_text,
+ history=history_str
+ )
+
+ # 调用LLM（支持多模态：prompt + images）
+ user_content: Any = prompt
+ if attachments:
+ user_content = [{"type": "text", "text": prompt}, *list(attachments)]
+ messages = [{"role": "user", "content": user_content}]
+ spinner = Spinner("奶浓正在思考...")
+ spinner.start()
+ response_text = self.llm.invoke(messages, **kwargs)
+ spinner.stop()
+
+ if not response_text:
+ print("❌ 错误：LLM未能返回有效响应。")
+ break
+
+ # 解析输出
+ thought, action = self._parse_output(response_text)
+
+ if thought:
+ print()
+ print(c("奶浓的思考:", INFO), thought)
+ print()
+
+
+ if not action:
+ # One forced retry: ask model to rewrite in strict format (helps for greetings / bilingual models)
+ try:
+ repair_sys = (
+ "You MUST output exactly two lines:\n"
+ "Thought: ...\n"
+ "Action: tool_name[tool_input] OR Finish[final answer]\n"
+ "No extra text. No markdown headers."
+ )
+ repair_user = f"Rewrite the following into the required two-line format:\n\n{response_text}"
+ spinner = Spinner("Repairing format…")
+ spinner.start()
+ repair_user_content: Any = repair_user
+ if attachments:
+ repair_user_content = [{"type": "text", "text": repair_user}, *list(attachments)]
+ repaired = self.llm.invoke(
+ [{"role": "system", "content": repair_sys}, {"role": "user", "content": repair_user_content}],
+ max_tokens=200,
+ )
+ spinner.stop()
+ thought, action = self._parse_output(repaired or "")
+ except Exception:
+ pass
+
+ if not action:
+ print("⚠️ 警告：未能解析出有效的Action，流程终止。")
+ break
+
+ # 检查是否完成
+ if action.startswith("Finish"):
+ final_answer = self._parse_action_input(action)
+ print()
+ print(c("奶浓认为是这样的:", PRIMARY))
+ print(final_answer)
+
+ # 保存到历史记录
+ self.add_message(Message(input_text, "user"))
+ self.add_message(Message(final_answer, "assistant"))
+
+ return final_answer
+
+ # 执行工具调用
+ tool_name, tool_input = self._parse_action(action)
+ if not tool_name or tool_input is None:
+ self.current_history.append("Observation: 无效的Action格式，请检查。")
+ continue
+
+ # Cursor-like gating: tools require human confirmation before execution.
+ # Default: confirm terminal commands (can extend via env).
+ confirm_tools = os.getenv("CODE_AGENT_CONFIRM_TOOLS", "terminal").strip()
+ confirm_set = {t.strip() for t in confirm_tools.split(",") if t.strip()}
+ if tool_name in confirm_set:
+ payload = {
+ "tool": tool_name,
+ "tool_input": tool_input,
+ "thought": thought,
+ "step": current_step,
+ }
+ # Return a special marker that UI can intercept and ask user to confirm/deny.
+ return (
+ f"需要用户确认后才能执行工具：{tool_name}\n"
+ f"拟执行输入：{tool_input}\n\n"
+ f"[[CONFIRM_TOOL]]{json.dumps(payload, ensure_ascii=False)}[[/CONFIRM_TOOL]]"
+ )
+
+ #log_tool_event(tool_name, tool_input)
+
+ # 调用工具
+ observation = self.tool_registry.execute_tool(tool_name, tool_input)
+ observation_full = observation
+ observation_summary = None
+ if (
+ self.observation_summarizer is not None
+ and isinstance(observation, str)
+ and len(observation) > self.summarize_threshold_chars
+ ):
+ try:
+ observation_summary = self.observation_summarizer(tool_name, tool_input, observation)
+ if observation_summary and isinstance(observation_summary, str):
+ observation = observation_summary.strip() + "\n...truncated...\n"
+ except Exception:
+ # fall back to raw observation
+ pass
+
+ #log_tool_event(f"{tool_name} result", clamp_text(str(observation), limit=6000))
+
+ # 提前终止：重复相同 action 且无明显进展
+ action_sig = f"{tool_name}|{tool_input}".strip()
+ if self.early_stop_on_repeat:
+ if last_action_sig == action_sig:
+ repeat_count += 1
+ else:
+ repeat_count = 0
+ last_action_sig = action_sig
+
+ if repeat_count >= self.repeat_action_threshold:
+ self.current_history.append("Observation: 已检测到重复行动，建议停止继续工具调用并给出当前能提供的结论/下一步。")
+ break
+
+ # 更新历史
+ self.current_history.append(f"Action: {action}")
+ self.current_history.append(f"Observation: {observation}")
+ self.last_trace.append(
+ {
+ "action": action,
+ "tool_name": tool_name,
+ "tool_input": tool_input,
+ "observation_full_len": len(observation_full) if isinstance(observation_full, str) else None,
+ "observation_summary": observation_summary,
+ }
+ )
+
+ # 未在循环内 Finish：进行兜底收敛
+ if self.finalize_on_max_steps:
+ try:
+ tools_desc = self.tool_registry.get_tools_description()
+ history_str = "\n".join(self.current_history[-24:])
+ finalize_prompt = (
+ "你是一个 ReAct 代理的最终收敛器。现在工具调用阶段结束了。"
+ "请基于已有的 Thought/Action/Observation 历史，给出一个尽可能有用的最终回答。"
+ "要求：\n"
+ "1) 不要再调用工具\n"
+ "2) 明确已完成的证据/发现\n"
+ "3) 如果信息不足，说清楚缺少什么，并给出下一步最小化建议（1-3条）\n"
+ )
+ final_user_text = f"Question:\n{input_text}\n\nTools:\n{tools_desc}\n\nTrace:\n{history_str}"
+ final_user_content: Any = final_user_text
+ if attachments:
+ final_user_content = [{"type": "text", "text": final_user_text}, *list(attachments)]
+ messages = [
+ {"role": "system", "content": finalize_prompt},
+ {"role": "user", "content": final_user_content},
+ ]
+ final_answer = self.llm.invoke(messages, max_tokens=600)
+ if final_answer:
+ self.add_message(Message(input_text, "user"))
+ self.add_message(Message(final_answer, "assistant"))
+ return final_answer
+ except Exception:
+ pass
+
+ print("⏰ 已达到最大步数，流程终止。")
+ final_answer = "抱歉，我无法在限定步数内完成这个任务。你可以缩小范围或指定目标文件/模块。"
+
+ # 保存到历史记录
+ self.add_message(Message(input_text, "user"))
+ self.add_message(Message(final_answer, "assistant"))
+
+ return final_answer
+
+ def _parse_output(self, text: str) -> Tuple[Optional[str], Optional[str]]:
+ """解析LLM输出，提取思考和行动。
+
+ 兼容常见变体：
+ - Thought/Action 的全角冒号（：）
+ - 中文标签：思考/行动
+ - Markdown 强调：**Thought:** / **Action:**
+ """
+ # Normalize to make regex easier
+ t = (text or "").strip()
+
+ # Primary: strict 2-line format, allow markdown markers and fullwidth colon
+ m = re.search(
+ r"(?:\*\*)?(Thought|思考)(?:\*\*)?\s*[:：]\s*(.*?)\n(?:\*\*)?(Action|行动)(?:\*\*)?\s*[:：]\s*(.*)\s*$",
+ t,
+ flags=re.DOTALL,
+ )
+ if m:
+ thought = m.group(2).strip()
+ action = m.group(4).strip()
+ return thought or None, action or None
+
+ # Fallback: find first Thought-like line and first Action-like line anywhere
+ thought_match = re.search(r"(?:\*\*)?(Thought|思考)(?:\*\*)?\s*[:：]\s*(.*)", t)
+ action_match = re.search(r"(?:\*\*)?(Action|行动)(?:\*\*)?\s*[:：]\s*(.*)", t)
+ thought = thought_match.group(2).strip() if thought_match else None
+ action_raw = action_match.group(2).strip() if action_match else None
+
+ # 关键修复：如果 action 中包含另一个 Thought/Action/Observation，截断到该位置
+ # 防止模型一次输出多个 Thought/Action 循环时，把后续内容都当作第一个 Action 的输入
+ if action_raw:
+ stop_patterns = [
+ r"\nThought:", r"\n思考:", r"\nAction:", r"\n行动:",
+ r"\nObservation:", r"\n观察:", r"\n\*\*Thought", r"\n\*\*Action",
+ ]
+ earliest_stop = len(action_raw)
+ for pat in stop_patterns:
+ m = re.search(pat, action_raw, re.IGNORECASE)
+ if m and m.start() < earliest_stop:
+ earliest_stop = m.start()
+ action_raw = action_raw[:earliest_stop].strip()
+
+ return thought, action_raw
+
+ def _parse_action(self, action_text: str) -> Tuple[Optional[str], Optional[str]]:
+ """解析行动文本，提取工具名称和输入
+
+ 使用括号匹配算法而非贪婪正则，正确处理嵌套 JSON。
+ """
+ # 先找工具名
+ name_match = re.match(r"(\w+)\[", action_text)
+ if not name_match:
+ return None, None
+
+ tool_name = name_match.group(1)
+ start = name_match.end() - 1 # '[' 的位置
+
+ # 使用括号匹配找到对应的 ']'
+ depth = 0
+ in_string = False
+ escape = False
+ end_pos = None
+
+ for i, c in enumerate(action_text[start:], start):
+ if escape:
+ escape = False
+ continue
+ if c == '\\' and in_string:
+ escape = True
+ continue
+ if c == '"' and not escape:
+ in_string = not in_string
+ continue
+ if in_string:
+ continue
+ if c == '[':
+ depth += 1
+ elif c == ']':
+ depth -= 1
+ if depth == 0:
+ end_pos = i
+ break
+
+ if end_pos is not None:
+ tool_input = action_text[start + 1:end_pos]
+ return tool_name, tool_input
+
+ # fallback: 如果括号不匹配，尝试简单正则（不跨行）
+ # 注意：不使用 re.DOTALL，这样 . 不会匹配换行符
+ match = re.match(r"(\w+)\[([^\n]*)\]", action_text)
+ if match:
+ return match.group(1), match.group(2)
+
+ return None, None
+
+ def _parse_action_input(self, action_text: str) -> str:
+ """解析行动输入
+
+ 兼容多种 Finish 书写：
+ - Finish[...]
+ - Finish：... / Finish: ...（无方括号）
+ - Finish\n