From 2f59b3e7988a9cf10b9da21311a266d29b63c46c Mon Sep 17 00:00:00 2001 From: zxmfke Date: Thu, 20 Mar 2025 16:05:40 +0800 Subject: [PATCH] feat: cot example --- README_COT.md | 102 +++++++++++++++++++++++++++++ app/agent/__init__.py | 2 + app/agent/cot.py | 45 +++++++++++++ app/prompt/cot.py | 15 +++++ examples/cot_agent_example.py | 33 ++++++++++ examples/multi_step_cot_example.py | 55 ++++++++++++++++ select_agent.py | 60 +++++++++++++++++ 7 files changed, 312 insertions(+) create mode 100644 README_COT.md create mode 100644 app/agent/cot.py create mode 100644 app/prompt/cot.py create mode 100644 examples/cot_agent_example.py create mode 100644 examples/multi_step_cot_example.py create mode 100644 select_agent.py diff --git a/README_COT.md b/README_COT.md new file mode 100644 index 0000000..f0eb4f3 --- /dev/null +++ b/README_COT.md @@ -0,0 +1,102 @@ +# Chain of Thought (CoT) 模式实现 + +本文档介绍了 OpenManus 中的 Chain of Thought (CoT) 模式实现及使用方法。 + +## 什么是 Chain of Thought + +Chain of Thought（思维链）是一种提示大语言模型展示其推理过程的方法，通过显示模型如何一步步思考问题，从而得出更准确的结论。与 ReAct（Reasoning and Acting）模式不同，CoT 专注于思考过程的展示，而不涉及工具的使用和执行。 + +## CoT 与 ReAct 的区别 + +|特性|CoT（思维链）|ReAct（推理行动）| +|---|---|---| +|主要目的|展示详细的思考过程|执行操作并与环境交互| +|使用工具|不使用工具|使用各种工具执行操作| +|步骤数|通常只需一步完成|多步骤循环（思考-行动-观察）| +|适用场景|复杂推理、数学问题、逻辑题|任务执行、信息检索、编程等| +|输出重点|思考过程和最终答案|工具调用结果和执行状态| + +## 实现概述 + +CoT 模式的实现包括以下文件： + +1. `app/agent/cot.py` - CoT 代理实现 +2. `app/prompt/cot.py` - CoT 特定的提示模板 + +CoT Agent 继承自 BaseAgent，实现了一个简单的思考步骤，不包含工具调用。 + +## 使用方法 + +### 方法一：使用选择器脚本 + +使用项目根目录的 `select_agent.py` 脚本，可以选择不同的代理模式： + +```bash +python select_agent.py +``` + +然后选择 "2" 使用 CoT 模式。 + +### 方法二：直接运行示例 + +项目包含两个示例脚本： + +1. 基本 CoT 示例： + +```bash +python examples/cot_agent_example.py +``` + +2. 多步骤 CoT 对话示例： + +```bash +python examples/multi_step_cot_example.py +``` + +### 方法三：在代码中使用 + +在你的代码中使用 CoT 代理： + +```python +from app.agent import CoTAgent + +async def main(): + agent = CoTAgent() + result = await agent.run("你的问题或提示") + print(result) +``` + +## 定制 CoT + +要定制 CoT 的行为，可以修改以下部分： + +1. 修改系统提示：编辑 `app/prompt/cot.py` 中的 `SYSTEM_PROMPT` +2. 调整输出格式：通过修改系统提示中的指示来改变输出格式 +3. 增加特定领域知识：根据需要在系统提示中添加特定领域的专业知识或指导 + +## 高级用法 + +### 多步骤推理 + +对于需要多步推理的复杂问题，可以设置更多的步骤： + +```python +agent = CoTAgent(max_steps=3) +``` + +然后使用 `step()` 方法进行逐步推理，在每步之间可以添加新的用户输入。 + +### 与其他模式结合 + +可以考虑将 CoT 与其他模式结合使用，例如： + +- 先使用 CoT 分析问题和制定计划 +- 然后使用 ReAct 执行具体操作 + +## 故障排除 + +如果遇到以下问题： + +1. **回答过于简短**：调整系统提示，强调需要详细的思考过程 +2. **没有遵循格式**：检查系统提示中的格式指导，确保清晰明确 +3. **性能问题**：CoT 适合较复杂的推理任务，对于简单问题可能显得冗长 diff --git a/app/agent/__init__.py b/app/agent/__init__.py index 4b8fb9d..ed53176 100644 --- a/app/agent/__init__.py +++ b/app/agent/__init__.py @@ -1,5 +1,6 @@ from app.agent.base import BaseAgent from app.agent.browser import BrowserAgent +from app.agent.cot import CoTAgent from app.agent.planning import PlanningAgent from app.agent.react import ReActAgent from app.agent.swe import SWEAgent @@ -9,6 +10,7 @@ from app.agent.toolcall import ToolCallAgent __all__ = [ "BaseAgent", "BrowserAgent", + "CoTAgent", "PlanningAgent", "ReActAgent", "SWEAgent", diff --git a/app/agent/cot.py b/app/agent/cot.py new file mode 100644 index 0000000..66106e5 --- /dev/null +++ b/app/agent/cot.py @@ -0,0 +1,45 @@ +from typing import List, Optional + +from pydantic import Field + +from app.agent.base import BaseAgent +from app.logger import logger +from app.llm import LLM +from app.prompt.cot import NEXT_STEP_PROMPT, SYSTEM_PROMPT +from app.schema import AgentState, Message + + +class CoTAgent(BaseAgent): + """Chain of Thought Agent - Focuses on demonstrating the thinking process of large language models without executing tools""" + + name: str = "cot" + description: str = "An agent that uses Chain of Thought reasoning" + + system_prompt: str = SYSTEM_PROMPT + next_step_prompt: Optional[str] = NEXT_STEP_PROMPT + + llm: LLM = Field(default_factory=LLM) + + max_steps: int = 1 # CoT typically only needs one step to complete reasoning + + async def step(self) -> str: + """Execute one step of chain of thought reasoning""" + logger.info(f"🧠 {self.name} is thinking...") + + # If next_step_prompt exists and this isn't the first message, add it to user messages + if self.next_step_prompt and len(self.messages) > 1: + self.memory.add_message(Message.user_message(self.next_step_prompt)) + + # Use system prompt and user messages + response = await self.llm.ask( + messages=self.messages, + system_msgs=[Message.system_message(self.system_prompt)] if self.system_prompt else None + ) + + # Record assistant's response + self.memory.add_message(Message.assistant_message(response)) + + # Set state to finished after completion + self.state = AgentState.FINISHED + + return response diff --git a/app/prompt/cot.py b/app/prompt/cot.py new file mode 100644 index 0000000..222fab5 --- /dev/null +++ b/app/prompt/cot.py @@ -0,0 +1,15 @@ +SYSTEM_PROMPT = """You are an assistant focused on Chain of Thought reasoning. For each question, please follow these steps: + +1. Break down the problem: Divide complex problems into smaller, more manageable parts +2. Think step by step: Think through each part in detail, showing your reasoning process +3. Synthesize conclusions: Integrate the thinking from each part into a complete solution +4. Provide an answer: Give a final concise answer + +Your response should follow this format: +Thinking: [Detailed thought process, including problem decomposition, reasoning for each step, and analysis] +Answer: [Final answer based on the thought process, clear and concise] + +Remember, the thinking process is more important than the final answer, as it demonstrates how you reached your conclusion. +""" + +NEXT_STEP_PROMPT = "Please continue your thinking based on the conversation above. If you've reached a conclusion, provide your final answer." diff --git a/examples/cot_agent_example.py b/examples/cot_agent_example.py new file mode 100644 index 0000000..ef02912 --- /dev/null +++ b/examples/cot_agent_example.py @@ -0,0 +1,33 @@ +#!/usr/bin/env python3 +""" +CoT Agent Example - Demonstrates how to use Chain of Thought mode for reasoning +""" + +import asyncio +import sys +from pathlib import Path + +# Add project root directory to path +sys.path.append(str(Path(__file__).resolve().parent.parent)) + +from app.agent import CoTAgent +from app.logger import logger + + +async def main(): + # Create CoT agent + agent = CoTAgent() + + # Provide a problem that requires thinking + question = "Zhang, Li, and Wang have a total of 85 apples. Zhang has twice as many apples as Li, and Wang has 5 more apples than Li. How many apples does each person have?" + + logger.info(f"Question: {question}") + + # Run the agent and get results + result = await agent.run(question) + + logger.info(f"Execution complete!") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/multi_step_cot_example.py b/examples/multi_step_cot_example.py new file mode 100644 index 0000000..9858278 --- /dev/null +++ b/examples/multi_step_cot_example.py @@ -0,0 +1,55 @@ +#!/usr/bin/env python3 +""" +Multi-step CoT Agent Example - Demonstrates how to use Chain of Thought mode in multi-turn conversations +""" + +import asyncio +import sys +from pathlib import Path + +# Add project root directory to path +sys.path.append(str(Path(__file__).resolve().parent.parent)) + +from app.agent import CoTAgent +from app.logger import logger +from app.schema import Message + + +async def main(): + # Create CoT agent + agent = CoTAgent(max_steps=3) # Set maximum steps to 3 + + # Initial question + initial_question = "As artificial intelligence technology develops, what ethical challenges might we face?" + + logger.info(f"Initial question: {initial_question}") + + # Add initial question to agent's memory + agent.memory.add_message(Message.user_message(initial_question)) + + # Step 1: Get initial thoughts + logger.info("Step 1: Initial thoughts") + response1 = await agent.step() + print(f"\nResponse:\n{response1}\n") + + # Step 2: Ask follow-up question + follow_up_question = "Among the ethical challenges you mentioned, which one do you think is most urgent to address? Why?" + logger.info(f"Follow-up question: {follow_up_question}") + + agent.memory.add_message(Message.user_message(follow_up_question)) + response2 = await agent.step() + print(f"\nResponse:\n{response2}\n") + + # Step 3: Ask final follow-up + final_question = "Can you suggest some specific solutions for addressing this most urgent ethical challenge?" + logger.info(f"Final question: {final_question}") + + agent.memory.add_message(Message.user_message(final_question)) + response3 = await agent.step() + print(f"\nResponse:\n{response3}\n") + + logger.info("Multi-step CoT conversation complete!") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/select_agent.py b/select_agent.py new file mode 100644 index 0000000..37bec28 --- /dev/null +++ b/select_agent.py @@ -0,0 +1,60 @@ +#!/usr/bin/env python3 +""" +Agent Selector - Allows users to choose different agent modes +""" + +import asyncio +import sys + +from app.agent import CoTAgent +from app.agent.manus import Manus +from app.logger import logger + + +async def run_cot_agent(): + """Run Chain of Thought agent""" + agent = CoTAgent() + prompt = input("Enter your question: ") + if not prompt.strip(): + logger.warning("Empty question provided") + return + + logger.warning("Processing your question...") + result = await agent.run(prompt) + logger.info("Question processing completed") + + +async def run_react_agent(): + """Run ReAct agent (Manus)""" + agent = Manus() + prompt = input("Enter your request: ") + if not prompt.strip(): + logger.warning("Empty request provided") + return + + logger.warning("Processing your request...") + await agent.run(prompt) + logger.info("Request processing completed") + + +async def main(): + print("\nSelect the agent mode to use:") + print("1. ReAct mode - Can use tools to execute tasks") + print("2. Chain of Thought (CoT) mode - Shows detailed thinking process") + + choice = input("\nEnter option (1 or 2): ").strip() + + if choice == "1": + await run_react_agent() + elif choice == "2": + await run_cot_agent() + else: + print("Invalid choice, please enter 1 or 2") + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + logger.warning("Operation interrupted") + sys.exit(0)