企业官网建设流程全解析

class BaseAgent(BaseModel, ABC)： async def run(self, request： Optional[str] = None) -> str： """执行代理的主循环""" if self.state != AgentState.IDLE： raise RuntimeError(f"Cannot run agent from state： {self.state}") if request： self.update_memory("user", request) results： List[str] = [] async with self.state_context(AgentState.RUNNING)： while (self.current_step < self.max_steps and self.state != AgentState.FINISHED)： self.current_step += 1 step_result = await self.step() # 检查是否陷入循环 if self.is_stuck()： self.handle_stuck_state() results.append(f"Step {self.current_step}： {step_result}") if self.current_step >= self.max_steps： self.current_step = 0 self.state = AgentState.IDLE results.append(f"Terminated： Reached max steps ({self.max_steps})") return "\n".join(results) if results else "No steps executed" @abstractmethod async def step(self) -> str： """执行单步操作，必须由子类实现"""

class ReActAgent(BaseAgent, ABC)： @abstractmethod async def think(self) -> bool： """处理当前状态并决定下一步行动""" @abstractmethod async def act(self) -> str： """执行决定的行动""" async def step(self) -> str： """执行单步：思考和行动""" should_act = await self.think() if not should_act： return "Thinking complete - no action needed" return await self.act()

class ToolCallAgent(ReActAgent): """能够执行工具调用的代理类""" available_tools: ToolCollection = ToolCollection( CreateChatCompletion(), Terminate() ) tool_choices: TOOL_CHOICE_TYPE = ToolChoice.AUTO special_tool_names: List[str] = Field(default_factory=lambda: [Terminate().name]) async def think(self) -> bool: """处理当前状态并使用工具决定下一步行动""" # 添加下一步提示到用户消息 if self.next_step_prompt: user_msg = Message.user_message(self.next_step_prompt) self.messages += [user_msg] # 请求 LLM 选择工具 response = await self.llm.ask_tool( messages=self.messages, system_msgs=([Message.system_message(self.system_prompt)] if self.system_prompt else None), tools=self.available_tools.to_params(), tool_choice=self.tool_choices, ) # 处理工具调用 self.tool_calls = tool_calls = ( response.tool_calls if response and response.tool_calls else [] ) content = response.content if response and response.content else "" # 添加助手消息到记忆 assistant_msg = ( Message.from_tool_calls(content=content, tool_calls=self.tool_calls) if self.tool_calls else Message.assistant_message(content) ) self.memory.add_message(assistant_msg) # 决定是否应该执行行动 return bool(self.tool_calls or content) async def act(self) -> str: """执行工具调用并处理结果""" if not self.tool_calls: # 返回最后一条消息内容，如果没有工具调用 return self.messages[-1].content or "No content or commands to execute" results = [] for command in self.tool_calls: # 执行工具 result = await self.execute_tool(command) # 记录工具响应到记忆 tool_msg = Message.tool_message( content=result, tool_call_id=command.id, name=command.function.name, base64_image=self._current_base64_image, ) self.memory.add_message(tool_msg) results.append(result) return "\n\n".join(results)

class Manus(ToolCallAgent): """多功能通用智能体，支持本地和 MCP 工具""" name: str = "Manus" description: str = "A versatile agent that can solve various tasks using multiple tools" # 添加各种通用工具到工具集合 available_tools: ToolCollection = Field( default_factory=lambda: ToolCollection( PythonExecute(), BrowserUseTool(), StrReplaceEditor(), AskHuman(), Terminate(), ) )

class BaseTool(ABC, BaseModel): name: str description: str parameters: Optional[dict] = None async def __call__(self, **kwargs) -> Any: """使用给定参数执行工具""" return await self.execute(**kwargs) @abstractmethod async def execute(self, **kwargs) -> Any: """执行工具的具体逻辑，由子类实现""" def to_param(self) -> Dict: """将工具转换为函数调用格式""" return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": self.parameters, }, }

class Terminate(BaseTool): name: str = "terminate" description: str = """Terminate the interaction when the request is met OR if the assistant cannot proceed further with the task. When you have finished all the tasks, call this tool to end the work.""" parameters: dict = { "type": "object", "properties": { "status": { "type": "string", "description": "The finish status of the interaction.", "enum": ["success", "failure"], } }, "required": ["status"], } async def execute(self, status: str) -> str: """完成当前执行""" return f"The interaction has been completed with status: {status}"

class AskHuman(BaseTool): """Add a tool to ask human for help.""" name: str = "ask_human" description: str = "Use this tool to ask human for help." parameters: str = { "type": "object", "properties": { "inquire": { "type": "string", "description": "The question you want to ask human.", } }, "required": ["inquire"], } async def execute(self, inquire: str) -> str: return input(f"""Bot: {inquire}\n\nYou: """).strip()

class ToolCollection: """A collection of defined tools.""" def __init__(self, *tools: BaseTool): self.tools = tools self.tool_map = {tool.name: tool for tool in tools} def to_params(self) -> List[Dict[str, Any]]: return [tool.to_param() for tool in self.tools] async def execute(self, *, name: str, tool_input: Dict[str, Any] = None) -> ToolResult: tool = self.tool_map.get(name) if not tool: return ToolFailure(error=f"Tool {name} is invalid") try: result = await tool(**tool_input) return result except ToolError as e: return ToolFailure(error=e.message) def add_tools(self, *tools: BaseTool): """Add multiple tools to the collection.""" for tool in tools: self.add_tool(tool) return self

private int duplicateThreshold = 2; /** * 处理陷入循环的状态 */ protected void handleStuckState() { String stuckPrompt = "观察到重复响应。考虑新策略，避免重复已尝试过的无效路径。"; this.nextStepPrompt = stuckPrompt + "\n" + (this.nextStepPrompt != null ? this.nextStepPrompt ： ""); System.out.println("Agent detected stuck state. Added prompt： " + stuckPrompt); } /** * 检查代理是否陷入循环 * * @return 是否陷入循环 */ protected boolean isStuck() { List<Message> messages = this.memory.getMessages(); if (messages.size() < 2) { return false; } Message lastMessage = messages.get(messages.size() - 1); if (lastMessage.getContent() == null || lastMessage.getContent().isEmpty()) { return false; } // 计算重复内容出现次数 int duplicateCount = 0; for (int i = messages.size() - 2; i >= 0; i--) { Message msg = messages.get(i); if (msg.getRole() == Role.ASSISTANT && lastMessage.getContent().equals(msg.getContent())) { duplicateCount++; } } return duplicateCount >= this.duplicateThreshold; } // 每一步 step 执行完都要检查是否陷入循环 if (isStuck()) { handleStuckState(); }

public boolean think() { boolean shouldAct = super.think(); // 获取最新的助手消息 Message lastMessage = getMessageList().get(getMessageList().size() - 1); if (lastMessage instanceof AssistantMessage) { String content = lastMessage.getContent(); // 检查是否包含向用户询问的标记 if (content.contains("[ASK_USER]")) { // 提取问题 String question = content.substring(content.indexOf("[ASK_USER]") + 10); // 向用户输出问题 System.out.println("智能体需要你的帮助： " + question); // 获取用户输入 Scanner scanner = new Scanner(System.in); String userAnswer = scanner.nextLine(); // 添加用户回答到消息列表 UserMessage userResponse = new UserMessage("用户回答： " + userAnswer); getMessageList().add(userResponse); // 需要继续思考 return true; } } return shouldAct; }