feat: cot example
This commit is contained in:
zxmfke
+102
@@ -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 适合较复杂的推理任务，对于简单问题可能显得冗长
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
@@ -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."
|
||||
@@ -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())
|
||||
@@ -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())
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user