Compare commits

..

20 Commits

Author SHA1 Message Date
Jiayi Zhang fad4722324 Merge pull request #1127 from KaeyoungLIAN/flow-auto-validate
Pre-commit checks / pre-commit-check (push) Failing after 1s
Add Latex beamer Generate and auto validate
2025-05-14 15:36:53 +08:00
Kaeyoung e2335877a8 precommit 2025-05-14 17:32:31 +10:00
Kaeyoung 0b62ef35b4 Update generatePPT.py 2025-05-14 17:22:23 +10:00
Kaeyoung a27732dca8 update according to comments 2025-05-14 17:02:20 +10:00
Kaeyoung 3f1e24f1d6 change file name 2025-05-08 11:02:48 +10:00
Kaeyoung 6026e7a9c6 Delete excessive modifications. 2025-05-08 10:39:26 +10:00
a5507203 b4161badc6 udpate 2025-05-05 22:54:01 +10:00
a5507203 a6b3571077 update 2025-05-05 18:00:44 +10:00
Kaeyoung abb9eb28a1 Update ppt.py 2025-04-22 14:16:11 +10:00
Kaeyoung 28a4a44420 retrieve the modification of latex_generator and task 2025-04-09 11:04:09 +10:00
Kaeyoung d307ee9a8d Add judge terminate logic and set latex_generator as test state 2025-04-09 10:49:39 +10:00
Kaeyoung a890058eb8 add fixtoolcall 2025-04-05 16:10:40 +11:00
Kaeyoung 2a437a2cd9 change .gitignore to push workspace and revert main.py 2025-03-31 15:32:28 +11:00
Kaeyoung 14c15d96d0 fix some files 2025-03-31 15:28:24 +11:00
Kaeyoung caa7ec8e16 delete some comments. 2025-03-29 19:52:54 +11:00
Kaeyoung d82387244a update server and llm 2025-03-29 19:43:02 +11:00
Kaeyoung 1dc96af1c8 Update toolcall.py 2025-03-29 19:40:37 +11:00
Kaeyoung 8a1c0ce89c delete required properties of validator tool. 2025-03-29 19:39:27 +11:00
Kaeyoung 2b9e0ba673 recover terminate and update valiterminate 2025-03-29 19:37:31 +11:00
Kaeyoung b0dfe3b224 update 2025-03-29 19:28:30 +11:00
55 changed files with 1259 additions and 1563 deletions
-3
View File
@@ -8,7 +8,6 @@ English | [中文](README_zh.md) | [한국어](README_ko.md) | [日本語](READM
 
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)  
[![Discord Follow](https://dcbadge.vercel.app/api/server/DYn29wFk9z?style=flat)](https://discord.gg/DYn29wFk9z)
[![Demo](https://img.shields.io/badge/Demo-Hugging%20Face-yellow)](https://huggingface.co/spaces/lyh-917/OpenManusDemo)
# 👋 OpenManus
@@ -162,8 +161,6 @@ and [browser-use](https://github.com/browser-use/browser-use) for providing basi
Additionally, we are grateful to [AAAJ](https://github.com/metauto-ai/agent-as-a-judge), [MetaGPT](https://github.com/geekan/MetaGPT), [OpenHands](https://github.com/All-Hands-AI/OpenHands) and [SWE-agent](https://github.com/SWE-agent/SWE-agent).
We also thank stepfun(阶跃星辰) for supporting our Hugging Face demo space.
OpenManus is built by contributors from MetaGPT. Huge thanks to this agent community!
## Cite
-3
View File
@@ -8,7 +8,6 @@
 
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)  
[![Discord Follow](https://dcbadge.vercel.app/api/server/DYn29wFk9z?style=flat)](https://discord.gg/DYn29wFk9z)
[![Demo](https://img.shields.io/badge/Demo-Hugging%20Face-yellow)](https://huggingface.co/spaces/lyh-917/OpenManusDemo)
# 👋 OpenManus
@@ -162,8 +161,6 @@ Feishuのネットワーキンググループに参加して、他の開発者
さらに、[AAAJ](https://github.com/metauto-ai/agent-as-a-judge)、[MetaGPT](https://github.com/geekan/MetaGPT)、[OpenHands](https://github.com/All-Hands-AI/OpenHands)、[SWE-agent](https://github.com/SWE-agent/SWE-agent)にも感謝します。
また、Hugging Face デモスペースをサポートしてくださった阶跃星辰 (stepfun)にも感謝いたします。
OpenManusはMetaGPTのコントリビューターによって構築されました。このエージェントコミュニティに大きな感謝を!
## 引用
-3
View File
@@ -8,7 +8,6 @@
 
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)  
[![Discord Follow](https://dcbadge.vercel.app/api/server/DYn29wFk9z?style=flat)](https://discord.gg/DYn29wFk9z)
[![Demo](https://img.shields.io/badge/Demo-Hugging%20Face-yellow)](https://huggingface.co/spaces/lyh-917/OpenManusDemo)
# 👋 OpenManus
@@ -162,8 +161,6 @@ Feishu 네트워킹 그룹에 참여하여 다른 개발자들과 경험을 공
또한, [AAAJ](https://github.com/metauto-ai/agent-as-a-judge), [MetaGPT](https://github.com/geekan/MetaGPT), [OpenHands](https://github.com/All-Hands-AI/OpenHands), [SWE-agent](https://github.com/SWE-agent/SWE-agent)에 깊은 감사를 드립니다.
또한 Hugging Face 데모 공간을 지원해 주신 阶跃星辰 (stepfun)에게 감사드립니다.
OpenManus는 MetaGPT 기여자들에 의해 개발되었습니다. 이 에이전트 커뮤니티에 깊은 감사를 전합니다!
## 인용
-3
View File
@@ -8,7 +8,6 @@
 
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)  
[![Discord Follow](https://dcbadge.vercel.app/api/server/DYn29wFk9z?style=flat)](https://discord.gg/DYn29wFk9z)
[![Demo](https://img.shields.io/badge/Demo-Hugging%20Face-yellow)](https://huggingface.co/spaces/lyh-917/OpenManusDemo)
# 👋 OpenManus
@@ -164,8 +163,6 @@ python run_flow.py
此外,我们感谢 [AAAJ](https://github.com/metauto-ai/agent-as-a-judge)[MetaGPT](https://github.com/geekan/MetaGPT)[OpenHands](https://github.com/All-Hands-AI/OpenHands) 和 [SWE-agent](https://github.com/SWE-agent/SWE-agent).
我们也感谢阶跃星辰 (stepfun) 提供的 Hugging Face 演示空间支持。
OpenManus 由 MetaGPT 社区的贡献者共同构建,感谢这个充满活力的智能体开发者社区!
## 引用我们
+2
View File
@@ -1,6 +1,7 @@
from app.agent.base import BaseAgent
from app.agent.browser import BrowserAgent
from app.agent.mcp import MCPAgent
from app.agent.planning import PlanningAgent
from app.agent.react import ReActAgent
from app.agent.swe import SWEAgent
from app.agent.toolcall import ToolCallAgent
@@ -9,6 +10,7 @@ from app.agent.toolcall import ToolCallAgent
__all__ = [
"BaseAgent",
"BrowserAgent",
"PlanningAgent",
"ReActAgent",
"SWEAgent",
"ToolCallAgent",
+88 -83
View File
@@ -1,7 +1,7 @@
import json
from typing import TYPE_CHECKING, Optional
from typing import Any, Optional
from pydantic import Field, model_validator
from pydantic import Field
from app.agent.toolcall import ToolCallAgent
from app.logger import logger
@@ -10,75 +10,6 @@ from app.schema import Message, ToolChoice
from app.tool import BrowserUseTool, Terminate, ToolCollection
# Avoid circular import if BrowserAgent needs BrowserContextHelper
if TYPE_CHECKING:
from app.agent.base import BaseAgent # Or wherever memory is defined
class BrowserContextHelper:
def __init__(self, agent: "BaseAgent"):
self.agent = agent
self._current_base64_image: Optional[str] = None
async def get_browser_state(self) -> Optional[dict]:
browser_tool = self.agent.available_tools.get_tool(BrowserUseTool().name)
if not browser_tool or not hasattr(browser_tool, "get_current_state"):
logger.warning("BrowserUseTool not found or doesn't have get_current_state")
return None
try:
result = await browser_tool.get_current_state()
if result.error:
logger.debug(f"Browser state error: {result.error}")
return None
if hasattr(result, "base64_image") and result.base64_image:
self._current_base64_image = result.base64_image
else:
self._current_base64_image = None
return json.loads(result.output)
except Exception as e:
logger.debug(f"Failed to get browser state: {str(e)}")
return None
async def format_next_step_prompt(self) -> str:
"""Gets browser state and formats the browser prompt."""
browser_state = await self.get_browser_state()
url_info, tabs_info, content_above_info, content_below_info = "", "", "", ""
results_info = "" # Or get from agent if needed elsewhere
if browser_state and not browser_state.get("error"):
url_info = f"\n URL: {browser_state.get('url', 'N/A')}\n Title: {browser_state.get('title', 'N/A')}"
tabs = browser_state.get("tabs", [])
if tabs:
tabs_info = f"\n {len(tabs)} tab(s) available"
pixels_above = browser_state.get("pixels_above", 0)
pixels_below = browser_state.get("pixels_below", 0)
if pixels_above > 0:
content_above_info = f" ({pixels_above} pixels)"
if pixels_below > 0:
content_below_info = f" ({pixels_below} pixels)"
if self._current_base64_image:
image_message = Message.user_message(
content="Current browser screenshot:",
base64_image=self._current_base64_image,
)
self.agent.memory.add_message(image_message)
self._current_base64_image = None # Consume the image after adding
return NEXT_STEP_PROMPT.format(
url_placeholder=url_info,
tabs_placeholder=tabs_info,
content_above_placeholder=content_above_info,
content_below_placeholder=content_below_info,
results_placeholder=results_info,
)
async def cleanup_browser(self):
browser_tool = self.agent.available_tools.get_tool(BrowserUseTool().name)
if browser_tool and hasattr(browser_tool, "cleanup"):
await browser_tool.cleanup()
class BrowserAgent(ToolCallAgent):
"""
A browser agent that uses the browser_use library to control a browser.
@@ -105,20 +36,94 @@ class BrowserAgent(ToolCallAgent):
tool_choices: ToolChoice = ToolChoice.AUTO
special_tool_names: list[str] = Field(default_factory=lambda: [Terminate().name])
browser_context_helper: Optional[BrowserContextHelper] = None
_current_base64_image: Optional[str] = None
@model_validator(mode="after")
def initialize_helper(self) -> "BrowserAgent":
self.browser_context_helper = BrowserContextHelper(self)
return self
async def _handle_special_tool(self, name: str, result: Any, **kwargs):
if not self._is_special_tool(name):
return
else:
await self.available_tools.get_tool(BrowserUseTool().name).cleanup()
await super()._handle_special_tool(name, result, **kwargs)
async def get_browser_state(self) -> Optional[dict]:
"""Get the current browser state for context in next steps."""
browser_tool = self.available_tools.get_tool(BrowserUseTool().name)
if not browser_tool:
return None
try:
# Get browser state directly from the tool
result = await browser_tool.get_current_state()
if result.error:
logger.debug(f"Browser state error: {result.error}")
return None
# Store screenshot if available
if hasattr(result, "base64_image") and result.base64_image:
self._current_base64_image = result.base64_image
# Parse the state info
return json.loads(result.output)
except Exception as e:
logger.debug(f"Failed to get browser state: {str(e)}")
return None
async def think(self) -> bool:
"""Process current state and decide next actions using tools, with browser state info added"""
self.next_step_prompt = (
await self.browser_context_helper.format_next_step_prompt()
)
return await super().think()
# Add browser state to the context
browser_state = await self.get_browser_state()
async def cleanup(self):
"""Clean up browser agent resources by calling parent cleanup."""
await self.browser_context_helper.cleanup_browser()
# Initialize placeholder values
url_info = ""
tabs_info = ""
content_above_info = ""
content_below_info = ""
results_info = ""
if browser_state and not browser_state.get("error"):
# URL and title info
url_info = f"\n URL: {browser_state.get('url', 'N/A')}\n Title: {browser_state.get('title', 'N/A')}"
# Tab information
if "tabs" in browser_state:
tabs = browser_state.get("tabs", [])
if tabs:
tabs_info = f"\n {len(tabs)} tab(s) available"
# Content above/below viewport
pixels_above = browser_state.get("pixels_above", 0)
pixels_below = browser_state.get("pixels_below", 0)
if pixels_above > 0:
content_above_info = f" ({pixels_above} pixels)"
if pixels_below > 0:
content_below_info = f" ({pixels_below} pixels)"
# Add screenshot as base64 if available
if self._current_base64_image:
# Create a message with image attachment
image_message = Message.user_message(
content="Current browser screenshot:",
base64_image=self._current_base64_image,
)
self.memory.add_message(image_message)
# Replace placeholders with actual browser state info
self.next_step_prompt = NEXT_STEP_PROMPT.format(
url_placeholder=url_info,
tabs_placeholder=tabs_info,
content_above_placeholder=content_above_info,
content_below_placeholder=content_below_info,
results_placeholder=results_info,
)
# Call parent implementation
result = await super().think()
# Reset the next_step_prompt to its original state
self.next_step_prompt = NEXT_STEP_PROMPT
return result
+19 -27
View File
@@ -1,10 +1,8 @@
from typing import Optional
from pydantic import Field
from pydantic import Field, model_validator
from app.agent.browser import BrowserContextHelper
from app.agent.toolcall import ToolCallAgent
from app.agent.browser import BrowserAgent
from app.config import config
from app.prompt.browser import NEXT_STEP_PROMPT as BROWSER_NEXT_STEP_PROMPT
from app.prompt.manus import NEXT_STEP_PROMPT, SYSTEM_PROMPT
from app.tool import Terminate, ToolCollection
from app.tool.browser_use_tool import BrowserUseTool
@@ -12,8 +10,14 @@ from app.tool.python_execute import PythonExecute
from app.tool.str_replace_editor import StrReplaceEditor
class Manus(ToolCallAgent):
"""A versatile general-purpose agent."""
class Manus(BrowserAgent):
"""
A versatile general-purpose agent that uses planning to solve various tasks.
This agent extends BrowserAgent with a comprehensive set of tools and capabilities,
including Python execution, web browsing, file operations, and information retrieval
to handle a wide range of user requests.
"""
name: str = "Manus"
description: str = (
@@ -33,39 +37,27 @@ class Manus(ToolCallAgent):
)
)
special_tool_names: list[str] = Field(default_factory=lambda: [Terminate().name])
browser_context_helper: Optional[BrowserContextHelper] = None
@model_validator(mode="after")
def initialize_helper(self) -> "Manus":
self.browser_context_helper = BrowserContextHelper(self)
return self
async def think(self) -> bool:
"""Process current state and decide next actions with appropriate context."""
# Store original prompt
original_prompt = self.next_step_prompt
# Only check recent messages (last 3) for browser activity
recent_messages = self.memory.messages[-3:] if self.memory.messages else []
browser_in_use = any(
tc.function.name == BrowserUseTool().name
"browser_use" in msg.content.lower()
for msg in recent_messages
if msg.tool_calls
for tc in msg.tool_calls
if hasattr(msg, "content") and isinstance(msg.content, str)
)
if browser_in_use:
self.next_step_prompt = (
await self.browser_context_helper.format_next_step_prompt()
)
# Override with browser-specific prompt temporarily to get browser context
self.next_step_prompt = BROWSER_NEXT_STEP_PROMPT
# Call parent's think method
result = await super().think()
# Restore original prompt
self.next_step_prompt = original_prompt
return result
async def cleanup(self):
"""Clean up Manus agent resources."""
if self.browser_context_helper:
await self.browser_context_helper.cleanup_browser()
+259
View File
@@ -0,0 +1,259 @@
import time
from typing import Dict, List, Optional
from pydantic import Field, model_validator
from app.agent.toolcall import ToolCallAgent
from app.logger import logger
from app.prompt.planning import NEXT_STEP_PROMPT, PLANNING_SYSTEM_PROMPT
from app.schema import TOOL_CHOICE_TYPE, Message, ToolCall, ToolChoice
from app.tool import PlanningTool, Terminate, ToolCollection
class PlanningAgent(ToolCallAgent):
"""
An agent that creates and manages plans to solve tasks.
This agent uses a planning tool to create and manage structured plans,
and tracks progress through individual steps until task completion.
"""
name: str = "planning"
description: str = "An agent that creates and manages plans to solve tasks"
system_prompt: str = PLANNING_SYSTEM_PROMPT
next_step_prompt: str = NEXT_STEP_PROMPT
available_tools: ToolCollection = Field(
default_factory=lambda: ToolCollection(PlanningTool(), Terminate())
)
tool_choices: TOOL_CHOICE_TYPE = ToolChoice.AUTO # type: ignore
special_tool_names: List[str] = Field(default_factory=lambda: [Terminate().name])
tool_calls: List[ToolCall] = Field(default_factory=list)
active_plan_id: Optional[str] = Field(default=None)
# Add a dictionary to track the step status for each tool call
step_execution_tracker: Dict[str, Dict] = Field(default_factory=dict)
current_step_index: Optional[int] = None
max_steps: int = 20
@model_validator(mode="after")
def initialize_plan_and_verify_tools(self) -> "PlanningAgent":
"""Initialize the agent with a default plan ID and validate required tools."""
self.active_plan_id = f"plan_{int(time.time())}"
if "planning" not in self.available_tools.tool_map:
self.available_tools.add_tool(PlanningTool())
return self
async def think(self) -> bool:
"""Decide the next action based on plan status."""
prompt = (
f"CURRENT PLAN STATUS:\n{await self.get_plan()}\n\n{self.next_step_prompt}"
if self.active_plan_id
else self.next_step_prompt
)
self.messages.append(Message.user_message(prompt))
# Get the current step index before thinking
self.current_step_index = await self._get_current_step_index()
result = await super().think()
# After thinking, if we decided to execute a tool and it's not a planning tool or special tool,
# associate it with the current step for tracking
if result and self.tool_calls:
latest_tool_call = self.tool_calls[0] # Get the most recent tool call
if (
latest_tool_call.function.name != "planning"
and latest_tool_call.function.name not in self.special_tool_names
and self.current_step_index is not None
):
self.step_execution_tracker[latest_tool_call.id] = {
"step_index": self.current_step_index,
"tool_name": latest_tool_call.function.name,
"status": "pending", # Will be updated after execution
}
return result
async def act(self) -> str:
"""Execute a step and track its completion status."""
result = await super().act()
# After executing the tool, update the plan status
if self.tool_calls:
latest_tool_call = self.tool_calls[0]
# Update the execution status to completed
if latest_tool_call.id in self.step_execution_tracker:
self.step_execution_tracker[latest_tool_call.id]["status"] = "completed"
self.step_execution_tracker[latest_tool_call.id]["result"] = result
# Update the plan status if this was a non-planning, non-special tool
if (
latest_tool_call.function.name != "planning"
and latest_tool_call.function.name not in self.special_tool_names
):
await self.update_plan_status(latest_tool_call.id)
return result
async def get_plan(self) -> str:
"""Retrieve the current plan status."""
if not self.active_plan_id:
return "No active plan. Please create a plan first."
result = await self.available_tools.execute(
name="planning",
tool_input={"command": "get", "plan_id": self.active_plan_id},
)
return result.output if hasattr(result, "output") else str(result)
async def run(self, request: Optional[str] = None) -> str:
"""Run the agent with an optional initial request."""
if request:
await self.create_initial_plan(request)
return await super().run()
async def update_plan_status(self, tool_call_id: str) -> None:
"""
Update the current plan progress based on completed tool execution.
Only marks a step as completed if the associated tool has been successfully executed.
"""
if not self.active_plan_id:
return
if tool_call_id not in self.step_execution_tracker:
logger.warning(f"No step tracking found for tool call {tool_call_id}")
return
tracker = self.step_execution_tracker[tool_call_id]
if tracker["status"] != "completed":
logger.warning(f"Tool call {tool_call_id} has not completed successfully")
return
step_index = tracker["step_index"]
try:
# Mark the step as completed
await self.available_tools.execute(
name="planning",
tool_input={
"command": "mark_step",
"plan_id": self.active_plan_id,
"step_index": step_index,
"step_status": "completed",
},
)
logger.info(
f"Marked step {step_index} as completed in plan {self.active_plan_id}"
)
except Exception as e:
logger.warning(f"Failed to update plan status: {e}")
async def _get_current_step_index(self) -> Optional[int]:
"""
Parse the current plan to identify the first non-completed step's index.
Returns None if no active step is found.
"""
if not self.active_plan_id:
return None
plan = await self.get_plan()
try:
plan_lines = plan.splitlines()
steps_index = -1
# Find the index of the "Steps:" line
for i, line in enumerate(plan_lines):
if line.strip() == "Steps:":
steps_index = i
break
if steps_index == -1:
return None
# Find the first non-completed step
for i, line in enumerate(plan_lines[steps_index + 1 :], start=0):
if "[ ]" in line or "[→]" in line: # not_started or in_progress
# Mark current step as in_progress
await self.available_tools.execute(
name="planning",
tool_input={
"command": "mark_step",
"plan_id": self.active_plan_id,
"step_index": i,
"step_status": "in_progress",
},
)
return i
return None # No active step found
except Exception as e:
logger.warning(f"Error finding current step index: {e}")
return None
async def create_initial_plan(self, request: str) -> None:
"""Create an initial plan based on the request."""
logger.info(f"Creating initial plan with ID: {self.active_plan_id}")
messages = [
Message.user_message(
f"Analyze the request and create a plan with ID {self.active_plan_id}: {request}"
)
]
self.memory.add_messages(messages)
response = await self.llm.ask_tool(
messages=messages,
system_msgs=[Message.system_message(self.system_prompt)],
tools=self.available_tools.to_params(),
tool_choice=ToolChoice.AUTO,
)
assistant_msg = Message.from_tool_calls(
content=response.content, tool_calls=response.tool_calls
)
self.memory.add_message(assistant_msg)
plan_created = False
for tool_call in response.tool_calls:
if tool_call.function.name == "planning":
result = await self.execute_tool(tool_call)
logger.info(
f"Executed tool {tool_call.function.name} with result: {result}"
)
# Add tool response to memory
tool_msg = Message.tool_message(
content=result,
tool_call_id=tool_call.id,
name=tool_call.function.name,
)
self.memory.add_message(tool_msg)
plan_created = True
break
if not plan_created:
logger.warning("No plan created from initial request")
tool_msg = Message.assistant_message(
"Error: Parameter `plan_id` is required for command: create"
)
self.memory.add_message(tool_msg)
async def main():
# Configure and run the agent
agent = PlanningAgent(available_tools=ToolCollection(PlanningTool(), Terminate()))
result = await agent.run("Help me plan a trip to the moon")
print(result)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
+64
View File
@@ -0,0 +1,64 @@
import json
import uuid
from openai.types.chat.chat_completion_message_tool_call import (
ChatCompletionMessageToolCall,
Function,
)
from app.agent.toolcall import ToolCallAgent
from app.logger import logger
from app.tool import LatexGenerator, ToolCollection, Validator
class PPTAgent(ToolCallAgent):
"""
Agent that executes a fixed sequence of tools, potentially terminating
early if the validator tool indicates completion.
"""
name: str = "fixed_toolcall"
description: str = (
"an agent that executes a fixed sequence of tools in predefined order, "
"potentially terminating early based on validator feedback."
)
available_tools: ToolCollection = ToolCollection(LatexGenerator(), Validator())
max_steps: int = 7
curr_step: int = 0
async def think(self) -> bool:
"""Process current state and decide next actions using tools"""
# pick which of your tools to call
tool_idx = self.curr_step % len(self.available_tools.tools)
tool_meta = self.available_tools.tools[tool_idx]
payload = {
"request": self.memory.messages[0].content,
"history": str(self.memory.messages),
}
arg_str = json.dumps(payload)
# build the Function descriptor
func_call = Function(
name=tool_meta.name,
arguments=arg_str,
)
# generate a proper call ID
call_id = f"call_{uuid.uuid4().hex}"
# wrap it up in a ChatCompletionMessageToolCall
tool_call = ChatCompletionMessageToolCall(
id=call_id, function=func_call, type="function"
)
# assign to self.tool_calls just like the SDK would
self.tool_calls = tool_calls = [tool_call]
logger.info(
f"🛠️ {self.name} selected {len(tool_calls) if tool_calls else 0} tools to use"
)
self.curr_step += 1
return True
+17 -3
View File
@@ -3,7 +3,7 @@ from typing import List
from pydantic import Field
from app.agent.toolcall import ToolCallAgent
from app.prompt.swe import SYSTEM_PROMPT
from app.prompt.swe import NEXT_STEP_TEMPLATE, SYSTEM_PROMPT
from app.tool import Bash, StrReplaceEditor, Terminate, ToolCollection
@@ -14,11 +14,25 @@ class SWEAgent(ToolCallAgent):
description: str = "an autonomous AI programmer that interacts directly with the computer to solve tasks."
system_prompt: str = SYSTEM_PROMPT
next_step_prompt: str = ""
next_step_prompt: str = NEXT_STEP_TEMPLATE
available_tools: ToolCollection = ToolCollection(
Bash(), StrReplaceEditor(), Terminate()
)
special_tool_names: List[str] = Field(default_factory=lambda: [Terminate().name])
max_steps: int = 20
max_steps: int = 30
bash: Bash = Field(default_factory=Bash)
working_dir: str = "."
async def think(self) -> bool:
"""Process current state and decide next action"""
# Update working directory
result = await self.bash.execute("pwd")
self.working_dir = result.output
self.next_step_prompt = self.next_step_prompt.format(
current_dir=self.working_dir
)
return await super().think()
-24
View File
@@ -1,4 +1,3 @@
import asyncio
import json
from typing import Any, List, Optional, Union
@@ -233,26 +232,3 @@ class ToolCallAgent(ReActAgent):
def _is_special_tool(self, name: str) -> bool:
"""Check if tool name is in special tools list"""
return name.lower() in [n.lower() for n in self.special_tool_names]
async def cleanup(self):
"""Clean up resources used by the agent's tools."""
logger.info(f"🧹 Cleaning up resources for agent '{self.name}'...")
for tool_name, tool_instance in self.available_tools.tool_map.items():
if hasattr(tool_instance, "cleanup") and asyncio.iscoroutinefunction(
tool_instance.cleanup
):
try:
logger.debug(f"🧼 Cleaning up tool: {tool_name}")
await tool_instance.cleanup()
except Exception as e:
logger.error(
f"🚨 Error cleaning up tool '{tool_name}': {e}", exc_info=True
)
logger.info(f"✨ Cleanup complete for agent '{self.name}'.")
async def run(self, request: Optional[str] = None) -> str:
"""Run the agent with cleanup when done."""
try:
return await super().run(request)
finally:
await self.cleanup()
-42
View File
@@ -37,26 +37,6 @@ class ProxySettings(BaseModel):
class SearchSettings(BaseModel):
engine: str = Field(default="Google", description="Search engine the llm to use")
fallback_engines: List[str] = Field(
default_factory=lambda: ["DuckDuckGo", "Baidu", "Bing"],
description="Fallback search engines to try if the primary engine fails",
)
retry_delay: int = Field(
default=60,
description="Seconds to wait before retrying all engines again after they all fail",
)
max_retries: int = Field(
default=3,
description="Maximum number of times to retry all engines when all fail",
)
lang: str = Field(
default="en",
description="Language code for search results (e.g., en, zh, fr)",
)
country: str = Field(
default="us",
description="Country code for search results (e.g., us, cn, uk)",
)
class BrowserSettings(BaseModel):
@@ -98,14 +78,6 @@ class SandboxSettings(BaseModel):
)
class MCPSettings(BaseModel):
"""Configuration for MCP (Model Context Protocol)"""
server_reference: str = Field(
"app.mcp.server", description="Module reference for the MCP server"
)
class AppConfig(BaseModel):
llm: Dict[str, LLMSettings]
sandbox: Optional[SandboxSettings] = Field(
@@ -117,7 +89,6 @@ class AppConfig(BaseModel):
search_config: Optional[SearchSettings] = Field(
None, description="Search configuration"
)
mcp_config: Optional[MCPSettings] = Field(None, description="MCP configuration")
class Config:
arbitrary_types_allowed = True
@@ -220,13 +191,6 @@ class Config:
else:
sandbox_settings = SandboxSettings()
mcp_config = raw_config.get("mcp", {})
mcp_settings = None
if mcp_config:
mcp_settings = MCPSettings(**mcp_config)
else:
mcp_settings = MCPSettings()
config_dict = {
"llm": {
"default": default_settings,
@@ -238,7 +202,6 @@ class Config:
"sandbox": sandbox_settings,
"browser_config": browser_settings,
"search_config": search_settings,
"mcp_config": mcp_settings,
}
self._config = AppConfig(**config_dict)
@@ -259,11 +222,6 @@ class Config:
def search_config(self) -> Optional[SearchSettings]:
return self._config.search_config
@property
def mcp_config(self) -> MCPSettings:
"""Get the MCP configuration"""
return self._config.mcp_config
@property
def workspace_root(self) -> Path:
"""Get the workspace root directory"""
+34
View File
@@ -1,4 +1,5 @@
from abc import ABC, abstractmethod
from enum import Enum
from typing import Dict, List, Optional, Union
from pydantic import BaseModel
@@ -6,6 +7,10 @@ from pydantic import BaseModel
from app.agent.base import BaseAgent
class FlowType(str, Enum):
PLANNING = "planning"
class BaseFlow(BaseModel, ABC):
"""Base class for execution flows supporting multiple agents"""
@@ -55,3 +60,32 @@ class BaseFlow(BaseModel, ABC):
@abstractmethod
async def execute(self, input_text: str) -> str:
"""Execute the flow with given input"""
class PlanStepStatus(str, Enum):
"""Enum class defining possible statuses of a plan step"""
NOT_STARTED = "not_started"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
BLOCKED = "blocked"
@classmethod
def get_all_statuses(cls) -> list[str]:
"""Return a list of all possible step status values"""
return [status.value for status in cls]
@classmethod
def get_active_statuses(cls) -> list[str]:
"""Return a list of values representing active statuses (not started or in progress)"""
return [cls.NOT_STARTED.value, cls.IN_PROGRESS.value]
@classmethod
def get_status_marks(cls) -> Dict[str, str]:
"""Return a mapping of statuses to their marker symbols"""
return {
cls.COMPLETED.value: "[✓]",
cls.IN_PROGRESS.value: "[→]",
cls.BLOCKED.value: "[!]",
cls.NOT_STARTED.value: "[ ]",
}
+1 -6
View File
@@ -1,15 +1,10 @@
from enum import Enum
from typing import Dict, List, Union
from app.agent.base import BaseAgent
from app.flow.base import BaseFlow
from app.flow.base import BaseFlow, FlowType
from app.flow.planning import PlanningFlow
class FlowType(str, Enum):
PLANNING = "planning"
class FlowFactory:
"""Factory for creating different types of flows with support for multiple agents"""
+1 -31
View File
@@ -1,47 +1,17 @@
import json
import time
from enum import Enum
from typing import Dict, List, Optional, Union
from pydantic import Field
from app.agent.base import BaseAgent
from app.flow.base import BaseFlow
from app.flow.base import BaseFlow, PlanStepStatus
from app.llm import LLM
from app.logger import logger
from app.schema import AgentState, Message, ToolChoice
from app.tool import PlanningTool
class PlanStepStatus(str, Enum):
"""Enum class defining possible statuses of a plan step"""
NOT_STARTED = "not_started"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
BLOCKED = "blocked"
@classmethod
def get_all_statuses(cls) -> list[str]:
"""Return a list of all possible step status values"""
return [status.value for status in cls]
@classmethod
def get_active_statuses(cls) -> list[str]:
"""Return a list of values representing active statuses (not started or in progress)"""
return [cls.NOT_STARTED.value, cls.IN_PROGRESS.value]
@classmethod
def get_status_marks(cls) -> Dict[str, str]:
"""Return a mapping of statuses to their marker symbols"""
return {
cls.COMPLETED.value: "[✓]",
cls.IN_PROGRESS.value: "[→]",
cls.BLOCKED.value: "[!]",
cls.NOT_STARTED.value: "[ ]",
}
class PlanningFlow(BaseFlow):
"""A flow that manages planning and execution of tasks using agents."""
+2 -3
View File
@@ -10,7 +10,7 @@ from openai import (
OpenAIError,
RateLimitError,
)
from openai.types.chat import ChatCompletion, ChatCompletionMessage
from openai.types.chat.chat_completion_message import ChatCompletionMessage
from tenacity import (
retry,
retry_if_exception_type,
@@ -735,9 +735,8 @@ class LLM:
temperature if temperature is not None else self.temperature
)
params["stream"] = False # Always use non-streaming for tool requests
response: ChatCompletion = await self.client.chat.completions.create(
**params
**params, stream=False
)
# Check if response is valid
+23 -7
View File
@@ -1,19 +1,30 @@
import logging
import sys
logging.basicConfig(level=logging.INFO, handlers=[logging.StreamHandler(sys.stderr)])
import argparse
import asyncio
import atexit
import json
import logging
import os
import sys
from inspect import Parameter, Signature
from typing import Any, Dict, Optional
from mcp.server.fastmcp import FastMCP
from app.logger import logger
# Add directories to Python path (needed for proper importing)
current_dir = os.path.dirname(os.path.abspath(__file__))
parent_dir = os.path.dirname(current_dir)
root_dir = os.path.dirname(parent_dir)
sys.path.insert(0, parent_dir)
sys.path.insert(0, current_dir)
sys.path.insert(0, root_dir)
# Configure logging (using the same format as original)
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("mcp-server")
from app.tool.base import BaseTool
from app.tool.bash import Bash
from app.tool.browser_use_tool import BrowserUseTool
@@ -34,6 +45,11 @@ class MCPServer:
self.tools["editor"] = StrReplaceEditor()
self.tools["terminate"] = Terminate()
from app.logger import logger as app_logger
global logger
logger = app_logger
def register_tool(self, tool: BaseTool, method_name: Optional[str] = None) -> None:
"""Register a tool with parameter validation and documentation."""
tool_name = method_name or tool.name
-15
View File
@@ -1,15 +0,0 @@
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."
+49
View File
@@ -0,0 +1,49 @@
SYSTEM_PROMPT = """
You are a LaTeX Beamer Presentation Generator. Your task is to generate a complete, informative, and ready-to-compile Beamer slide deck in LaTeX, based on the task description and any past drafts or feedback.
## Goals:
- Each slide must be **self-contained**, meaning the audience should understand the slide without external explanations.
- The presentation must **teach** or **explain** the topic in sufficient detail using structured LaTeX slides.
- Each slide must contribute meaningfully to the overall structure and flow of the presentation.
## Requirements:
1. Preamble & Setup
- Start with `\\documentclass{beamer}`.
- Use packages such as `amsmath`, `amsfonts`, and `graphicx`.
- Use the `Madrid` theme unless otherwise specified.
- Include full metadata: `\\title{}`, `\\author{}`, and `\\date{\\today}`.
2. Slide Design
- MUST mark each slide with a comment indicating its number, `% Slide 1`, `% Slide 2`.
- - Slides must follow a **logical order** that ensures smooth flow and coherence.
- AIM for a **minimum of 300 words per slide* Contain **enough detail** (text, bullets, equations, definitions, or examples)
3. Depth of Content
- For important concept, include motivation problem intuitive explanation mathematical formulation or equation (if applicable)
- practical example or application can also be included
4. Completeness & Validity
- Reflect all provided feedback and correct deficiencies from past versions.
- MUST No placeholders or incomplete content.
- Your output will be used directly. Therefore, it must be a ready-to-use result.
- Include `\\end{document}`.
- Ensure valid LaTeX syntax.
5. Style & Clarity
- Maintain consistent formatting and indentation.
- Use bullet points or short paragraphs for clarity.
- Keep math readable and contextualized with supporting text.
**Only output the final LaTeX source code. Do not include explanations, notes, or comments.**
"""
USER_CONTENT = """
## Task
{request}
## Past Drafts & Feedback
{history}
"""
+6
View File
@@ -20,3 +20,9 @@ Remember, you should always include a _SINGLE_ tool call/function call and then
If you'd like to issue two commands at once, PLEASE DO NOT DO THAT! Please instead first submit just the first tool call, and then after receiving a response you'll be able to issue the second tool call.
Note that the environment does NOT support interactive session commands (e.g. python, vim), so please do not invoke them.
"""
NEXT_STEP_TEMPLATE = """{{observation}}
(Open file: {{open_file}})
(Current directory: {{working_dir}})
bash-$
"""
+42
View File
@@ -0,0 +1,42 @@
TEXT_VALIDATION_PROMPT = """
You are a task result evaluator responsible for determining whether a task result meets the task requirements, if not, you need to improve it.
# Objective and Steps
1. **Completeness and Quality Check:**
- Verify that the result includes all required elements of the task.
- Evaluate whether the output meets overall quality criteria (accuracy, clarity, formatting, and completeness).
2. **Change Detection:**
- If this is a subsequent result, compare it with previous iterations.
- If the differences are minimal or the result has not significantly improved, consider it "good enough" for finalization.
3. **Feedback and Escalation:**
- If the result meets the criteria or the improvements are negligible compared to previous iterations, return **"No further feedback"**.
- Otherwise, provide **direct and precise feedback** and **output the improved result in the required format** for finalization.
4. **Ensure Completeness:**
- Your output must meet all requirements of the task.
- Include all necessary details so that the output is self-contained and can be directly used as input for downstream tasks.
5. **Do NOT:**
- Leave any section with placeholders (e.g., "TODO", "Add content here").
- Include any commentary or reminders to the writer or user (e.g., "We can add more later").
- Output partial slides or omit essential details assuming future input.
- **If the result meets the standard:**
- Return **"No further feedback."**.
- **If the result does not meet the standard:**
- add detailed jusification for the change start with "here are some feedbacks" and directly write an improved new result start with "here are the changes".
# Note that: Any output containing incomplete sections, placeholders is not allowed.
"""
USER_CONTENT = """
## Current Task Requirement:
{request}
---
## Current Task Latest Result:
{history}
"""
+38 -3
View File
@@ -96,6 +96,35 @@ class Message(BaseModel):
message["base64_image"] = self.base64_image
return message
def to_string(self) -> str:
"""
Convert the Message instance into a human-readable string format.
Returns:
str: A formatted string representing the message.
"""
# Format the header with role and name
role_str = self.role.upper() if self.role else "UNKNOWN"
name_str = f" ({self.name})" if self.name else ""
header = f"[{role_str}{name_str}]"
# Start with the message content
content = self.content or ""
# Append tool call details if available
if self.tool_calls:
tool_calls_str = "\n".join(
f" ↳ ToolCall: {tc.function.name}({tc.function.arguments}) [id={tc.id}]"
for tc in self.tool_calls
)
content += "\n" + tool_calls_str
# Append information about attached image if any
if self.base64_image:
content += "\n ↳ [Image Attached: base64 content hidden]"
return f"{header}\n{content.strip()}"
@classmethod
def user_message(
cls, content: str, base64_image: Optional[str] = None
@@ -170,9 +199,6 @@ class Memory(BaseModel):
def add_messages(self, messages: List[Message]) -> None:
"""Add multiple messages to memory"""
self.messages.extend(messages)
# Optional: Implement message limit
if len(self.messages) > self.max_messages:
self.messages = self.messages[-self.max_messages :]
def clear(self) -> None:
"""Clear all messages"""
@@ -185,3 +211,12 @@ class Memory(BaseModel):
def to_dict_list(self) -> List[dict]:
"""Convert messages to list of dicts"""
return [msg.to_dict() for msg in self.messages]
def to_string(self) -> str:
"""
Convert the memory's list of messages to a readable string format.
Returns:
str: A formatted string representing the entire conversation history.
"""
return "\n\n".join(msg.to_string() for msg in self.messages)
+5 -4
View File
@@ -2,23 +2,24 @@ from app.tool.base import BaseTool
from app.tool.bash import Bash
from app.tool.browser_use_tool import BrowserUseTool
from app.tool.create_chat_completion import CreateChatCompletion
from app.tool.deep_research import DeepResearch
from app.tool.latex_generator import LatexGenerator
from app.tool.planning import PlanningTool
from app.tool.str_replace_editor import StrReplaceEditor
from app.tool.terminate import Terminate
from app.tool.tool_collection import ToolCollection
from app.tool.web_search import WebSearch
from app.tool.validator import Validator
__all__ = [
"BaseTool",
"Bash",
"BrowserUseTool",
"DeepResearch",
"Terminate",
"ValiTerminate",
"StrReplaceEditor",
"WebSearch",
"ToolCollection",
"CreateChatCompletion",
"PlanningTool",
"Validator",
"LatexGenerator",
]
+133 -79
View File
@@ -16,21 +16,34 @@ from app.tool.base import BaseTool, ToolResult
from app.tool.web_search import WebSearch
_BROWSER_DESCRIPTION = """\
A powerful browser automation tool that allows interaction with web pages through various actions.
* This tool provides commands for controlling a browser session, navigating web pages, and extracting information
* It maintains state across calls, keeping the browser session alive until explicitly closed
* Use this when you need to browse websites, fill forms, click buttons, extract content, or perform web searches
* Each action requires specific parameters as defined in the tool's dependencies
_BROWSER_DESCRIPTION = """
Interact with a web browser to perform various actions such as navigation, element interaction, content extraction, and tab management. This tool provides a comprehensive set of browser automation capabilities:
Key capabilities include:
* Navigation: Go to specific URLs, go back, search the web, or refresh pages
* Interaction: Click elements, input text, select from dropdowns, send keyboard commands
* Scrolling: Scroll up/down by pixel amount or scroll to specific text
* Content extraction: Extract and analyze content from web pages based on specific goals
* Tab management: Switch between tabs, open new tabs, or close tabs
Navigation:
- 'go_to_url': Go to a specific URL in the current tab
- 'go_back': Go back
- 'refresh': Refresh the current page
- 'web_search': Search the query in the current tab, the query should be a search query like humans search in web, concrete and not vague or super long. More the single most important items.
Note: When using element indices, refer to the numbered elements shown in the current browser state.
Element Interaction:
- 'click_element': Click an element by index
- 'input_text': Input text into a form element
- 'scroll_down'/'scroll_up': Scroll the page (with optional pixel amount)
- 'scroll_to_text': If you dont find something which you want to interact with, scroll to it
- 'send_keys': Send strings of special keys like Escape,Backspace, Insert, PageDown, Delete, Enter, Shortcuts such as `Control+o`, `Control+Shift+T` are supported as well. This gets used in keyboard.press.
- 'get_dropdown_options': Get all options from a dropdown
- 'select_dropdown_option': Select dropdown option for interactive element index by the text of the option you want to select
Content Extraction:
- 'extract_content': Extract page content to retrieve specific information from the page, e.g. all company names, a specifc description, all information about, links with companies in structured format or simply links
Tab Management:
- 'switch_tab': Switch to a specific tab
- 'open_tab': Open a new tab with a URL
- 'close_tab': Close the current tab
Utility:
- 'wait': Wait for a specified number of seconds
"""
Context = TypeVar("Context")
@@ -253,19 +266,32 @@ class BrowserUseTool(BaseTool, Generic[Context]):
return ToolResult(
error="Query is required for 'web_search' action"
)
# Execute the web search and return results directly without browser navigation
search_response = await self.web_search_tool.execute(
query=query, fetch_content=True, num_results=1
)
# Navigate to the first search result
first_search_result = search_response.results[0]
url_to_navigate = first_search_result.url
search_results = await self.web_search_tool.execute(query)
page = await context.get_current_page()
await page.goto(url_to_navigate)
await page.wait_for_load_state()
if search_results:
# Navigate to the first search result
first_result = search_results[0]
if isinstance(first_result, dict) and "url" in first_result:
url_to_navigate = first_result["url"]
elif isinstance(first_result, str):
url_to_navigate = first_result
else:
return ToolResult(
error=f"Invalid search result format: {first_result}"
)
return search_response
page = await context.get_current_page()
await page.goto(url_to_navigate)
await page.wait_for_load_state()
return ToolResult(
output=f"Searched for '{query}' and navigated to first result: {url_to_navigate}\nAll results:"
+ "\n".join([str(r) for r in search_results])
)
else:
return ToolResult(
error=f"No search results found for '{query}'"
)
# Element interaction actions
elif action == "click_element":
@@ -377,71 +403,99 @@ class BrowserUseTool(BaseTool, Generic[Context]):
return ToolResult(
error="Goal is required for 'extract_content' action"
)
page = await context.get_current_page()
import markdownify
try:
# Get page content and convert to markdown for better processing
html_content = await page.content()
content = markdownify.markdownify(await page.content())
# Import markdownify here to avoid global import
try:
import markdownify
prompt = f"""\
content = markdownify.markdownify(html_content)
except ImportError:
# Fallback if markdownify is not available
content = html_content
# Create prompt for LLM
prompt_text = """
Your task is to extract the content of the page. You will be given a page and a goal, and you should extract all relevant information around this goal from the page. If the goal is vague, summarize the page. Respond in json format.
Extraction goal: {goal}
Page content:
{content[:max_content_length]}
{page}
"""
messages = [{"role": "system", "content": prompt}]
# Define extraction function schema
extraction_function = {
"type": "function",
"function": {
"name": "extract_content",
"description": "Extract specific information from a webpage based on a goal",
"parameters": {
"type": "object",
"properties": {
"extracted_content": {
"type": "object",
"description": "The content extracted from the page according to the goal",
"properties": {
"text": {
"type": "string",
"description": "Text content extracted from the page",
},
"metadata": {
"type": "object",
"description": "Additional metadata about the extracted content",
"properties": {
"source": {
"type": "string",
"description": "Source of the extracted content",
}
},
},
},
}
},
"required": ["extracted_content"],
},
},
}
# Use LLM to extract content with required function calling
response = await self.llm.ask_tool(
messages,
tools=[extraction_function],
tool_choice="required",
)
if response and response.tool_calls:
args = json.loads(response.tool_calls[0].function.arguments)
extracted_content = args.get("extracted_content", {})
return ToolResult(
output=f"Extracted from page:\n{extracted_content}\n"
# Format the prompt with the goal and content
max_content_length = min(50000, len(content))
formatted_prompt = prompt_text.format(
goal=goal, page=content[:max_content_length]
)
return ToolResult(output="No content was extracted from the page.")
# Create a proper message list for the LLM
from app.schema import Message
messages = [Message.user_message(formatted_prompt)]
# Define extraction function for the tool
extraction_function = {
"type": "function",
"function": {
"name": "extract_content",
"description": "Extract specific information from a webpage based on a goal",
"parameters": {
"type": "object",
"properties": {
"extracted_content": {
"type": "object",
"description": "The content extracted from the page according to the goal",
}
},
"required": ["extracted_content"],
},
},
}
# Use LLM to extract content with required function calling
response = await self.llm.ask_tool(
messages,
tools=[extraction_function],
tool_choice="required",
)
# Extract content from function call response
if (
response
and response.tool_calls
and len(response.tool_calls) > 0
):
# Get the first tool call arguments
tool_call = response.tool_calls[0]
# Parse the JSON arguments
try:
args = json.loads(tool_call.function.arguments)
extracted_content = args.get("extracted_content", {})
# Format extracted content as JSON string
content_json = json.dumps(
extracted_content, indent=2, ensure_ascii=False
)
msg = f"Extracted from page:\n{content_json}\n"
except Exception as e:
msg = f"Error parsing extraction result: {str(e)}\nRaw response: {tool_call.function.arguments}"
else:
msg = "No content was extracted from the page."
return ToolResult(output=msg)
except Exception as e:
# Provide a more helpful error message
error_msg = f"Failed to extract content: {str(e)}"
try:
# Try to return a portion of the page content as fallback
return ToolResult(
output=f"{error_msg}\nHere's a portion of the page content:\n{content[:2000]}..."
)
except:
# If all else fails, just return the error
return ToolResult(error=error_msg)
# Tab management actions
elif action == "switch_tab":
-537
View File
@@ -1,537 +0,0 @@
import asyncio
import json
import re
import time
from typing import List, Optional, Set
from pydantic import BaseModel, ConfigDict, Field, model_validator
from app.exceptions import ToolError
from app.llm import LLM
from app.logger import logger
from app.schema import ToolChoice
from app.tool.base import BaseTool, ToolResult
from app.tool.web_search import SearchResult, WebSearch
# Prompts for LLM interactions
OPTIMIZE_QUERY_PROMPT = """
You are a research assistant helping to optimize a search query for web research.
Your task is to reformulate the given query to be more effective for web searches.
Make it specific, use relevant keywords, and ensure it's clear and concise.
Original query: {query}
Provide only the optimized query text without any explanation or additional formatting.
"""
EXTRACT_INSIGHTS_PROMPT = """
Analyze the following content and extract key insights related to the research query.
For each insight, assess its relevance to the query on a scale of 0.0 to 1.0.
Research query: {query}
Content to analyze:
{content}
Extract up to 3 most important insights from this content. For each insight:
1. Provide the insight content
2. Provide relevance score (0.0-1.0)
"""
GENERATE_FOLLOW_UPS_PROMPT = """
Based on the insights discovered so far, generate follow-up research queries to explore gaps or related areas.
These should help deepen our understanding of the topic.
Original query: {original_query}
Current query: {current_query}
Key insights so far:
{insights}
Generate up to 3 specific follow-up queries that would help address gaps in our current knowledge.
Each query should be concise and focused on a specific aspect of the research topic.
"""
# Constants for insight parsing
DEFAULT_RELEVANCE_SCORE = 1.0
FALLBACK_RELEVANCE_SCORE = 0.7
FALLBACK_CONTENT_LIMIT = 500
# Pattern to detect start of an insight (number., -, *, •) and capture content
INSIGHT_MARKER_PATTERN = re.compile(r"^\s*(?:\d+\.|-|\*|•)\s*(.*)")
# Pattern to detect relevance score, capturing the number (case-insensitive)
RELEVANCE_SCORE_PATTERN = re.compile(r"relevance.*?:.*?(\d\.?\d*)", re.IGNORECASE)
class ResearchInsight(BaseModel):
"""A single insight discovered during research."""
model_config = ConfigDict(frozen=True) # Make insights immutable
content: str = Field(description="The insight content")
source_url: str = Field(description="URL where this insight was found")
source_title: Optional[str] = Field(default=None, description="Title of the source")
relevance_score: float = Field(
default=1.0, description="Relevance score (0.0-1.0)", ge=0.0, le=1.0
)
def __str__(self) -> str:
"""Format insight as string with source attribution."""
source = self.source_title or self.source_url
return f"{self.content} [Source: {source}]"
class ResearchContext(BaseModel):
"""Research context for tracking research progress."""
query: str = Field(description="The original research query")
insights: List[ResearchInsight] = Field(
default_factory=list, description="Key insights discovered"
)
follow_up_queries: List[str] = Field(
default_factory=list, description="Generated follow-up queries"
)
visited_urls: Set[str] = Field(
default_factory=set, description="URLs visited during research"
)
current_depth: int = Field(
default=0, description="Current depth of research exploration", ge=0
)
max_depth: int = Field(
default=2, description="Maximum depth of research to reach", ge=1
)
class ResearchSummary(ToolResult):
"""Comprehensive summary of deep research results."""
model_config = ConfigDict(arbitrary_types_allowed=True)
query: str = Field(description="The original research query")
insights: List[ResearchInsight] = Field(
default_factory=list, description="Key insights discovered"
)
visited_urls: Set[str] = Field(
default_factory=set, description="URLs visited during research"
)
depth_reached: int = Field(
default=0, description="Maximum depth of research reached", ge=0
)
@model_validator(mode="after")
def populate_output(self) -> "ResearchSummary":
"""Populate the output field after validation."""
# Group and sort insights by relevance
grouped_insights = {
"Key Findings": [i for i in self.insights if i.relevance_score >= 0.8],
"Additional Information": [
i for i in self.insights if 0.5 <= i.relevance_score < 0.8
],
"Supplementary Information": [
i for i in self.insights if i.relevance_score < 0.5
],
}
sections = [
f"# Research: {self.query}\n",
f"**Sources**: {len(self.visited_urls)} | **Depth**: {self.depth_reached + 1}\n",
]
for section_title, insights in grouped_insights.items():
if insights:
sections.append(f"## {section_title}")
for i, insight in enumerate(insights, 1):
sections.extend(
[
insight.content,
f"> Source: [{insight.source_title or 'Link'}]({insight.source_url})\n",
]
)
# Assign the formatted string to the 'output' field inherited from ToolResult
self.output = "\n".join(sections)
return self
class DeepResearch(BaseTool):
"""Advanced research tool that explores a topic through iterative web searches."""
name: str = "deep_research"
description: str = """
Performs comprehensive research on a topic through multi-level web searches
and content analysis. Returns a structured summary of findings with source
attribution and relevance ratings.
"""
parameters: dict = {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The research question or topic to investigate.",
},
"max_depth": {
"type": "integer",
"description": "Maximum depth of iterative research (1-5). Default is 2.",
"default": 2,
},
"results_per_search": {
"type": "integer",
"description": "Number of search results to analyze per search (1-20). Default is 5.",
"default": 5,
},
"max_insights": {
"type": "integer",
"description": "Maximum number of insights to return. Default is 20.",
"default": 20,
},
"time_limit_seconds": {
"type": "integer",
"description": "Maximum execution time in seconds. Default is 120.",
"default": 120,
},
},
"required": ["query"],
}
# Dependency injection for easier testing
search_tool: WebSearch = Field(default_factory=WebSearch)
llm: LLM = Field(default_factory=LLM)
async def execute(
self,
query: str,
max_depth: int = 2,
results_per_search: int = 5,
max_insights: int = 20,
time_limit_seconds: int = 120,
) -> ResearchSummary:
"""Execute deep research on the given query."""
# Normalize parameters
max_depth = max(1, min(max_depth, 5))
results_per_search = max(1, min(results_per_search, 20))
# Initialize research context and set deadline
context = ResearchContext(query=query, max_depth=max_depth)
deadline = time.time() + time_limit_seconds
try:
# Initiate research process with optimized query
optimized_query = await self._generate_optimized_query(query)
await self._research_graph(
context=context,
query=optimized_query,
results_count=results_per_search,
deadline=deadline,
)
except ToolError as e:
logger.error(f"Research error: {str(e)}")
# Prepare final summary
return ResearchSummary(
query=query,
insights=sorted(
context.insights, key=lambda x: x.relevance_score, reverse=True
)[:max_insights],
visited_urls=context.visited_urls,
depth_reached=context.current_depth,
)
async def _generate_optimized_query(self, query: str) -> str:
"""Generate an optimized search query using LLM."""
try:
prompt = OPTIMIZE_QUERY_PROMPT.format(query=query)
response = await self.llm.ask_tool(
[{"role": "user", "content": prompt}],
tools=[
{
"type": "function",
"function": {
"name": "optimize_query",
"description": "Generate an optimized search query",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The optimized search query",
}
},
"required": ["query"],
},
},
}
],
tool_choice=ToolChoice.REQUIRED,
stream=False,
)
# Extract the query from the tool_call response
if response and response.tool_calls and len(response.tool_calls) > 0:
tool_call = response.tool_calls[0]
arguments = json.loads(tool_call.function.arguments)
optimized_query = arguments.get("query", "")
else:
# Fallback to original query if tool call failed
logger.warning("Tool call failed to return a valid response")
return query
if not optimized_query:
logger.warning("Generated empty optimized query, using original")
return query
logger.info(f"Optimized query: '{optimized_query}'")
return optimized_query
except Exception as e:
logger.warning(f"Failed to optimize query: {str(e)}")
return query # Fall back to original query on error
async def _research_graph(
self,
context: ResearchContext,
query: str,
results_count: int,
deadline: float,
) -> None:
"""Run a complete research cycle (search, analyze, generate follow-ups)."""
# Check termination conditions
if time.time() >= deadline or context.current_depth >= context.max_depth:
return
# Log current research step
logger.info(f"Research cycle at depth {context.current_depth + 1}")
# 1. Web search
search_results = await self._search_web(query, results_count)
if not search_results:
return
# 2. Extract insights
new_insights = await self._extract_insights(
context, search_results, context.query, deadline
)
if not new_insights:
return
# 3. Generate follow-up queries
follow_up_queries = await self._generate_follow_ups(
new_insights, query, context.query
)
context.follow_up_queries.extend(follow_up_queries)
# Update depth and proceed to next level
context.current_depth += 1
# 4. Continue research with follow-up queries
if follow_up_queries and context.current_depth < context.max_depth:
tasks = [] # Create a list to hold the tasks
for follow_up in follow_up_queries[:2]: # Limit branching factor
if time.time() >= deadline:
break
# Create a coroutine for the recursive research call
task = self._research_graph(
context=context,
query=follow_up,
results_count=max(1, results_count - 1), # Reduce result count
deadline=deadline,
)
tasks.append(task) # Add the task to the list
# Run all the created tasks concurrently
if tasks:
await asyncio.gather(*tasks)
async def _search_web(self, query: str, results_count: int) -> List[SearchResult]:
"""Perform web search for the given query."""
search_response = await self.search_tool.execute(
query=query, num_results=results_count, fetch_content=True
)
return [] if search_response.error else search_response.results
async def _extract_insights(
self,
context: ResearchContext,
results: List[SearchResult],
original_query: str,
deadline: float,
) -> List[ResearchInsight]:
"""Extract insights from search results."""
all_insights = []
for rst in results:
# Skip if URL already visited or time exceeded
if rst.url in context.visited_urls or time.time() >= deadline:
continue
context.visited_urls.add(rst.url)
# Skip if no content available
if not rst.raw_content:
continue
# Extract insights using LLM
insights = await self._analyze_content(
content=rst.raw_content[:10000], # Limit content size
url=rst.url,
title=rst.title,
query=original_query,
)
all_insights.extend(insights)
context.insights.extend(insights)
# Log discovered insights
logger.info(f"Extracted {len(insights)} insights from {rst.url}")
return all_insights
async def _generate_follow_ups(
self, insights: List[ResearchInsight], current_query: str, original_query: str
) -> List[str]:
"""Generate follow-up queries based on insights."""
if not insights:
return []
# Format insights for the prompt
insights_text = "\n".join([f"- {insight.content}" for insight in insights[:5]])
# Create prompt for generating follow-up queries
prompt = GENERATE_FOLLOW_UPS_PROMPT.format(
original_query=original_query,
current_query=current_query,
insights=insights_text,
)
# Get follow-up queries from LLM using structured output
response = await self.llm.ask_tool(
[{"role": "user", "content": prompt}],
tools=[
{
"type": "function",
"function": {
"name": "generate_follow_ups",
"description": "Generate follow-up queries based on research insights",
"parameters": {
"type": "object",
"properties": {
"follow_up_queries": {
"type": "array",
"items": {"type": "string"},
"description": "List of follow-up queries (max 3) that would help address gaps in current knowledge",
"maxItems": 3,
}
},
"required": ["follow_up_queries"],
},
},
}
],
tool_choice=ToolChoice.REQUIRED,
stream=False,
)
# Extract queries from the tool response
queries = []
if response and response.tool_calls and len(response.tool_calls) > 0:
tool_call = response.tool_calls[0]
arguments = json.loads(tool_call.function.arguments)
queries = arguments.get("follow_up_queries", [])
# Ensure we don't return more than 3 queries
return queries[:3]
async def _analyze_content(
self, content: str, url: str, title: str, query: str
) -> List[ResearchInsight]:
"""Extract insights from content based on relevance to query."""
prompt = EXTRACT_INSIGHTS_PROMPT.format(
query=query, content=content[:5000] # Limit content size
)
response = await self.llm.ask_tool(
[{"role": "user", "content": prompt}],
tools=[
{
"type": "function",
"function": {
"name": "extract_insights",
"description": "Extract key insights from content with relevance scores",
"parameters": {
"type": "object",
"properties": {
"insights": {
"type": "array",
"items": {
"type": "object",
"properties": {
"content": {
"type": "string",
"description": "The insight content",
},
"relevance_score": {
"type": "number",
"description": "Relevance score between 0.0 and 1.0",
"minimum": 0.0,
"maximum": 1.0,
},
},
"required": ["content", "relevance_score"],
},
"description": "List of key insights extracted from the content",
"maxItems": 3,
}
},
"required": ["insights"],
},
},
}
],
tool_choice=ToolChoice.REQUIRED,
stream=False,
)
insights = []
# Process structured JSON response
if response and response.tool_calls and len(response.tool_calls) > 0:
tool_call = response.tool_calls[0]
arguments = json.loads(tool_call.function.arguments)
extracted_insights = arguments.get("insights", [])
for insight_data in extracted_insights:
insights.append(
ResearchInsight(
content=insight_data.get("content", ""),
source_url=url,
source_title=title,
relevance_score=insight_data.get(
"relevance_score", FALLBACK_RELEVANCE_SCORE
),
)
)
# Fallback: if no structured insights found, use fallback approach
if not insights:
logger.warning(
f"Could not parse structured insights from LLM response for {url}. Using fallback."
)
insights.append(
ResearchInsight(
content=f"Failed to extract structured insights from content about {title or url}."[
:FALLBACK_CONTENT_LIMIT
],
source_url=url,
source_title=title,
relevance_score=FALLBACK_RELEVANCE_SCORE,
)
)
return insights
if __name__ == "__main__":
deep_research = DeepResearch()
result = asyncio.run(
deep_research.execute(
"What is deep learning", max_depth=1, results_per_search=2
)
)
print(result)
+67
View File
@@ -0,0 +1,67 @@
import os
import aiofiles
from app.config import WORKSPACE_ROOT
from app.tool.base import BaseTool
class FileSaver(BaseTool):
name: str = "file_saver"
description: str = """Save content to a local file at a specified path.
Use this tool when you need to save text, code, or generated content to a file on the local filesystem.
The tool accepts content and a file path, and saves the content to that location.
"""
parameters: dict = {
"type": "object",
"properties": {
"content": {
"type": "string",
"description": "(required) The content to save to the file.",
},
"file_path": {
"type": "string",
"description": "(required) The path where the file should be saved, including filename and extension.",
},
"mode": {
"type": "string",
"description": "(optional) The file opening mode. Default is 'w' for write. Use 'a' for append.",
"enum": ["w", "a"],
"default": "w",
},
},
"required": ["content", "file_path"],
}
async def execute(self, content: str, file_path: str, mode: str = "w") -> str:
"""
Save content to a file at the specified path.
Args:
content (str): The content to save to the file.
file_path (str): The path where the file should be saved.
mode (str, optional): The file opening mode. Default is 'w' for write. Use 'a' for append.
Returns:
str: A message indicating the result of the operation.
"""
try:
# Place the generated file in the workspace directory
if os.path.isabs(file_path):
file_name = os.path.basename(file_path)
full_path = os.path.join(WORKSPACE_ROOT, file_name)
else:
full_path = os.path.join(WORKSPACE_ROOT, file_path)
# Ensure the directory exists
directory = os.path.dirname(full_path)
if directory and not os.path.exists(directory):
os.makedirs(directory)
# Write directly to the file
async with aiofiles.open(full_path, mode, encoding="utf-8") as file:
await file.write(content)
return f"Content successfully saved to {full_path}"
except Exception as e:
return f"Error saving file: {str(e)}"
+38
View File
@@ -0,0 +1,38 @@
from pydantic import Field
from app.llm import LLM
from app.prompt.latex_generator import SYSTEM_PROMPT, USER_CONTENT
from app.tool.base import BaseTool
_Latex_Generator_DESCRIPTION = """
This agent generates complete, high-quality LaTeX documents with a focus on Beamer presentations. It accepts topic-specific input and produces fully self-contained LaTeX source code, including all required packages, structures, and rich content elements such as equations, figures, and formatted text. The agent ensures completeness by avoiding any placeholders or incomplete sections.
In addition to generation, the agent supports iterative refinement: it evaluates and improves the generated LaTeX code based on validation feedback to ensure correctness, formatting quality, and logical structure. The final output is ready for immediate compilation and professional presentation use.
"""
class LatexGenerator(BaseTool):
llm: LLM = Field(default_factory=LLM, description="Language model instance")
name: str = "latexgenerator"
description: str = _Latex_Generator_DESCRIPTION
parameters: dict = {}
async def generate(self, request: str, history: str = ""):
"""Abstract method for result validate logic.
Args:
step_result: The result string to validate.
"""
system_content = SYSTEM_PROMPT
user_content = USER_CONTENT.format(request=request, history=history)
feedback = await self.llm.ask(
messages=[{"role": "user", "content": user_content}],
system_msgs=[{"role": "system", "content": system_content}],
)
return feedback
async def execute(self, request: str, history: str = "") -> str:
"""Finish the current execution"""
return await self.generate(request, history)
+4 -49
View File
@@ -1,54 +1,9 @@
from typing import List
from baidusearch.baidusearch import search
from app.tool.search.base import SearchItem, WebSearchEngine
from app.tool.search.base import WebSearchEngine
class BaiduSearchEngine(WebSearchEngine):
def perform_search(
self, query: str, num_results: int = 10, *args, **kwargs
) -> List[SearchItem]:
"""
Baidu search engine.
Returns results formatted according to SearchItem model.
"""
raw_results = search(query, num_results=num_results)
# Convert raw results to SearchItem format
results = []
for i, item in enumerate(raw_results):
if isinstance(item, str):
# If it's just a URL
results.append(
SearchItem(title=f"Baidu Result {i+1}", url=item, description=None)
)
elif isinstance(item, dict):
# If it's a dictionary with details
results.append(
SearchItem(
title=item.get("title", f"Baidu Result {i+1}"),
url=item.get("url", ""),
description=item.get("abstract", None),
)
)
else:
# Try to get attributes directly
try:
results.append(
SearchItem(
title=getattr(item, "title", f"Baidu Result {i+1}"),
url=getattr(item, "url", ""),
description=getattr(item, "abstract", None),
)
)
except Exception:
# Fallback to a basic result
results.append(
SearchItem(
title=f"Baidu Result {i+1}", url=str(item), description=None
)
)
return results
def perform_search(self, query, num_results=10, *args, **kwargs):
"""Baidu search engine."""
return search(query, num_results=num_results)
+4 -27
View File
@@ -1,32 +1,9 @@
from typing import List, Optional
from pydantic import BaseModel, Field
class SearchItem(BaseModel):
"""Represents a single search result item"""
title: str = Field(description="The title of the search result")
url: str = Field(description="The URL of the search result")
description: Optional[str] = Field(
default=None, description="A description or snippet of the search result"
)
def __str__(self) -> str:
"""String representation of a search result item."""
return f"{self.title} - {self.url}"
class WebSearchEngine(BaseModel):
"""Base class for web search engines."""
model_config = {"arbitrary_types_allowed": True}
class WebSearchEngine(object):
def perform_search(
self, query: str, num_results: int = 10, *args, **kwargs
) -> List[SearchItem]:
) -> list[dict]:
"""
Perform a web search and return a list of search items.
Perform a web search and return a list of URLs.
Args:
query (str): The search query to submit to the search engine.
@@ -35,6 +12,6 @@ class WebSearchEngine(BaseModel):
kwargs: Additional keyword arguments.
Returns:
List[SearchItem]: A list of SearchItem objects matching the search query.
List: A list of dict matching the search query.
"""
raise NotImplementedError
+31 -29
View File
@@ -1,10 +1,10 @@
from typing import List, Optional, Tuple
from typing import List
import requests
from bs4 import BeautifulSoup
from app.logger import logger
from app.tool.search.base import SearchItem, WebSearchEngine
from app.tool.search.base import WebSearchEngine
ABSTRACT_MAX_LENGTH = 300
@@ -36,7 +36,7 @@ BING_SEARCH_URL = "https://www.bing.com/search?q="
class BingSearchEngine(WebSearchEngine):
session: Optional[requests.Session] = None
session: requests.Session = None
def __init__(self, **data):
"""Initialize the BingSearch tool with a requests session."""
@@ -44,16 +44,21 @@ class BingSearchEngine(WebSearchEngine):
self.session = requests.Session()
self.session.headers.update(HEADERS)
def _search_sync(self, query: str, num_results: int = 10) -> List[SearchItem]:
def _search_sync(self, query: str, num_results: int = 10) -> List[str]:
"""
Synchronous Bing search implementation to retrieve search results.
Synchronous Bing search implementation to retrieve a list of URLs matching a query.
Args:
query (str): The search query to submit to Bing.
num_results (int, optional): Maximum number of results to return. Defaults to 10.
query (str): The search query to submit to Bing. Must not be empty.
num_results (int, optional): The maximum number of URLs to return. Defaults to 10.
Returns:
List[SearchItem]: A list of search items with title, URL, and description.
List[str]: A list of URLs from the search results, capped at `num_results`.
Returns an empty list if the query is empty or no results are found.
Notes:
- Pagination is handled by incrementing the `first` parameter and following `next_url` links.
- If fewer results than `num_results` are available, all found URLs are returned.
"""
if not query:
return []
@@ -67,21 +72,25 @@ class BingSearchEngine(WebSearchEngine):
next_url, rank_start=len(list_result), first=first
)
if data:
list_result.extend(data)
list_result.extend([item["url"] for item in data])
if not next_url:
break
first += 10
return list_result[:num_results]
def _parse_html(
self, url: str, rank_start: int = 0, first: int = 1
) -> Tuple[List[SearchItem], str]:
def _parse_html(self, url: str, rank_start: int = 0, first: int = 1) -> tuple:
"""
Parse Bing search result HTML to extract search results and the next page URL.
Parse Bing search result HTML synchronously to extract search results and the next page URL.
Args:
url (str): The URL of the Bing search results page to parse.
rank_start (int, optional): The starting rank for numbering the search results. Defaults to 0.
first (int, optional): Unused parameter (possibly legacy). Defaults to 1.
Returns:
tuple: (List of SearchItem objects, next page URL or None)
tuple: A tuple containing:
- list: A list of dictionaries with keys 'title', 'abstract', 'url', and 'rank' for each result.
- str or None: The URL of the next results page, or None if there is no next page.
"""
try:
res = self.session.get(url=url)
@@ -111,14 +120,13 @@ class BingSearchEngine(WebSearchEngine):
abstract = abstract[:ABSTRACT_MAX_LENGTH]
rank_start += 1
# Create a SearchItem object
list_data.append(
SearchItem(
title=title or f"Bing Result {rank_start}",
url=url,
description=abstract,
)
{
"title": title,
"abstract": abstract,
"url": url,
"rank": rank_start,
}
)
except Exception:
continue
@@ -133,12 +141,6 @@ class BingSearchEngine(WebSearchEngine):
logger.warning(f"Error parsing HTML: {e}")
return [], None
def perform_search(
self, query: str, num_results: int = 10, *args, **kwargs
) -> List[SearchItem]:
"""
Bing search engine.
Returns results formatted according to SearchItem model.
"""
def perform_search(self, query, num_results=10, *args, **kwargs):
"""Bing search engine."""
return self._search_sync(query, num_results=num_results)
+4 -52
View File
@@ -1,57 +1,9 @@
from typing import List
from duckduckgo_search import DDGS
from app.tool.search.base import SearchItem, WebSearchEngine
from app.tool.search.base import WebSearchEngine
class DuckDuckGoSearchEngine(WebSearchEngine):
def perform_search(
self, query: str, num_results: int = 10, *args, **kwargs
) -> List[SearchItem]:
"""
DuckDuckGo search engine.
Returns results formatted according to SearchItem model.
"""
raw_results = DDGS().text(query, max_results=num_results)
results = []
for i, item in enumerate(raw_results):
if isinstance(item, str):
# If it's just a URL
results.append(
SearchItem(
title=f"DuckDuckGo Result {i + 1}", url=item, description=None
)
)
elif isinstance(item, dict):
# Extract data from the dictionary
results.append(
SearchItem(
title=item.get("title", f"DuckDuckGo Result {i + 1}"),
url=item.get("href", ""),
description=item.get("body", None),
)
)
else:
# Try to extract attributes directly
try:
results.append(
SearchItem(
title=getattr(item, "title", f"DuckDuckGo Result {i + 1}"),
url=getattr(item, "href", ""),
description=getattr(item, "body", None),
)
)
except Exception:
# Fallback
results.append(
SearchItem(
title=f"DuckDuckGo Result {i + 1}",
url=str(item),
description=None,
)
)
return results
async def perform_search(self, query, num_results=10, *args, **kwargs):
"""DuckDuckGo search engine."""
return DDGS.text(query, num_results=num_results)
+4 -28
View File
@@ -1,33 +1,9 @@
from typing import List
from googlesearch import search
from app.tool.search.base import SearchItem, WebSearchEngine
from app.tool.search.base import WebSearchEngine
class GoogleSearchEngine(WebSearchEngine):
def perform_search(
self, query: str, num_results: int = 10, *args, **kwargs
) -> List[SearchItem]:
"""
Google search engine.
Returns results formatted according to SearchItem model.
"""
raw_results = search(query, num_results=num_results, advanced=True)
results = []
for i, item in enumerate(raw_results):
if isinstance(item, str):
# If it's just a URL
results.append(
{"title": f"Google Result {i+1}", "url": item, "description": ""}
)
else:
results.append(
SearchItem(
title=item.title, url=item.url, description=item.description
)
)
return results
def perform_search(self, query, num_results=10, *args, **kwargs):
"""Google search engine."""
return search(query, num_results=num_results)
+182
View File
@@ -0,0 +1,182 @@
import asyncio
import os
import shlex
from typing import Optional
from app.tool.base import BaseTool, CLIResult
class Terminal(BaseTool):
name: str = "execute_command"
description: str = """Request to execute a CLI command on the system.
Use this when you need to perform system operations or run specific commands to accomplish any step in the user's task.
You must tailor your command to the user's system and provide a clear explanation of what the command does.
Prefer to execute complex CLI commands over creating executable scripts, as they are more flexible and easier to run.
Commands will be executed in the current working directory.
Note: You MUST append a `sleep 0.05` to the end of the command for commands that will complete in under 50ms, as this will circumvent a known issue with the terminal tool where it will sometimes not return the output when the command completes too quickly.
"""
parameters: dict = {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "(required) The CLI command to execute. This should be valid for the current operating system. Ensure the command is properly formatted and does not contain any harmful instructions.",
}
},
"required": ["command"],
}
process: Optional[asyncio.subprocess.Process] = None
current_path: str = os.getcwd()
lock: asyncio.Lock = asyncio.Lock()
async def execute(self, command: str) -> CLIResult:
"""
Execute a terminal command asynchronously with persistent context.
Args:
command (str): The terminal command to execute.
Returns:
str: The output, and error of the command execution.
"""
# Split the command by & to handle multiple commands
commands = [cmd.strip() for cmd in command.split("&") if cmd.strip()]
final_output = CLIResult(output="", error="")
for cmd in commands:
sanitized_command = self._sanitize_command(cmd)
# Handle 'cd' command internally
if sanitized_command.lstrip().startswith("cd "):
result = await self._handle_cd_command(sanitized_command)
else:
async with self.lock:
try:
self.process = await asyncio.create_subprocess_shell(
sanitized_command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=self.current_path,
)
stdout, stderr = await self.process.communicate()
result = CLIResult(
output=stdout.decode().strip(),
error=stderr.decode().strip(),
)
except Exception as e:
result = CLIResult(output="", error=str(e))
finally:
self.process = None
# Combine outputs
if result.output:
final_output.output += (
(result.output + "\n") if final_output.output else result.output
)
if result.error:
final_output.error += (
(result.error + "\n") if final_output.error else result.error
)
# Remove trailing newlines
final_output.output = final_output.output.rstrip()
final_output.error = final_output.error.rstrip()
return final_output
async def execute_in_env(self, env_name: str, command: str) -> CLIResult:
"""
Execute a terminal command asynchronously within a specified Conda environment.
Args:
env_name (str): The name of the Conda environment.
command (str): The terminal command to execute within the environment.
Returns:
str: The output, and error of the command execution.
"""
sanitized_command = self._sanitize_command(command)
# Construct the command to run within the Conda environment
# Using 'conda run -n env_name command' to execute without activating
conda_command = f"conda run -n {shlex.quote(env_name)} {sanitized_command}"
return await self.execute(conda_command)
async def _handle_cd_command(self, command: str) -> CLIResult:
"""
Handle 'cd' commands to change the current path.
Args:
command (str): The 'cd' command to process.
Returns:
TerminalOutput: The result of the 'cd' command.
"""
try:
parts = shlex.split(command)
if len(parts) < 2:
new_path = os.path.expanduser("~")
else:
new_path = os.path.expanduser(parts[1])
# Handle relative paths
if not os.path.isabs(new_path):
new_path = os.path.join(self.current_path, new_path)
new_path = os.path.abspath(new_path)
if os.path.isdir(new_path):
self.current_path = new_path
return CLIResult(
output=f"Changed directory to {self.current_path}", error=""
)
else:
return CLIResult(output="", error=f"No such directory: {new_path}")
except Exception as e:
return CLIResult(output="", error=str(e))
@staticmethod
def _sanitize_command(command: str) -> str:
"""
Sanitize the command for safe execution.
Args:
command (str): The command to sanitize.
Returns:
str: The sanitized command.
"""
# Example sanitization: restrict certain dangerous commands
dangerous_commands = ["rm", "sudo", "shutdown", "reboot"]
try:
parts = shlex.split(command)
if any(cmd in dangerous_commands for cmd in parts):
raise ValueError("Use of dangerous commands is restricted.")
except Exception:
# If shlex.split fails, try basic string comparison
if any(cmd in command for cmd in dangerous_commands):
raise ValueError("Use of dangerous commands is restricted.")
# Additional sanitization logic can be added here
return command
async def close(self):
"""Close the persistent shell process if it exists."""
async with self.lock:
if self.process:
self.process.terminate()
try:
await asyncio.wait_for(self.process.wait(), timeout=5)
except asyncio.TimeoutError:
self.process.kill()
await self.process.wait()
finally:
self.process = None
async def __aenter__(self):
"""Enter the asynchronous context manager."""
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Exit the asynchronous context manager and close the process."""
await self.close()
+40
View File
@@ -0,0 +1,40 @@
from pydantic import Field
from app.llm import LLM
from app.prompt.validator import TEXT_VALIDATION_PROMPT, USER_CONTENT
from app.tool.base import BaseTool
_VALIDATE_DESCRIPTION = """
This tool evaluates the quality and completeness of a subtask result against a set of predefined criteria.
It checks whether the result fully satisfies task requirements, maintains high quality in terms of clarity, accuracy, and formatting,
and determines whether improvements have been made in comparison to prior versions.
If the result is satisfactory or improvements are minimal, it returns "The step result has already reached the requirement.".
Otherwise, it provides detailed feedback and a revised version of the result that meets all requirements and is ready for downstream use.
"""
class Validator(BaseTool):
llm: LLM = Field(default_factory=LLM, description="Language model instance")
name: str = "validator"
description: str = _VALIDATE_DESCRIPTION
parameters: dict = {}
async def validate(self, request: str, history: str):
"""Abstract method for result validate logic.
Args:
step_result: The result string to validate.
"""
system_content = TEXT_VALIDATION_PROMPT
user_content = USER_CONTENT.format(request=request, history=history)
feedback = await self.llm.ask(
messages=[{"role": "user", "content": user_content}],
system_msgs=[{"role": "system", "content": system_content}],
)
return feedback
async def execute(self, request: str, history: str) -> str:
"""Finish the current execution"""
return await self.validate(request, history)
+40 -357
View File
@@ -1,14 +1,10 @@
import asyncio
from typing import Any, Dict, List, Optional
from typing import List
import requests
from bs4 import BeautifulSoup
from pydantic import BaseModel, ConfigDict, Field, model_validator
from tenacity import retry, stop_after_attempt, wait_exponential
from app.config import config
from app.logger import logger
from app.tool.base import BaseTool, ToolResult
from app.tool.base import BaseTool
from app.tool.search import (
BaiduSearchEngine,
BingSearchEngine,
@@ -16,150 +12,13 @@ from app.tool.search import (
GoogleSearchEngine,
WebSearchEngine,
)
from app.tool.search.base import SearchItem
class SearchResult(BaseModel):
"""Represents a single search result returned by a search engine."""
model_config = ConfigDict(arbitrary_types_allowed=True)
position: int = Field(description="Position in search results")
url: str = Field(description="URL of the search result")
title: str = Field(default="", description="Title of the search result")
description: str = Field(
default="", description="Description or snippet of the search result"
)
source: str = Field(description="The search engine that provided this result")
raw_content: Optional[str] = Field(
default=None, description="Raw content from the search result page if available"
)
def __str__(self) -> str:
"""String representation of a search result."""
return f"{self.title} ({self.url})"
class SearchMetadata(BaseModel):
"""Metadata about the search operation."""
model_config = ConfigDict(arbitrary_types_allowed=True)
total_results: int = Field(description="Total number of results found")
language: str = Field(description="Language code used for the search")
country: str = Field(description="Country code used for the search")
class SearchResponse(ToolResult):
"""Structured response from the web search tool, inheriting ToolResult."""
query: str = Field(description="The search query that was executed")
results: List[SearchResult] = Field(
default_factory=list, description="List of search results"
)
metadata: Optional[SearchMetadata] = Field(
default=None, description="Metadata about the search"
)
@model_validator(mode="after")
def populate_output(self) -> "SearchResponse":
"""Populate output or error fields based on search results."""
if self.error:
return self
result_text = [f"Search results for '{self.query}':"]
for i, result in enumerate(self.results, 1):
# Add title with position number
title = result.title.strip() or "No title"
result_text.append(f"\n{i}. {title}")
# Add URL with proper indentation
result_text.append(f" URL: {result.url}")
# Add description if available
if result.description.strip():
result_text.append(f" Description: {result.description}")
# Add content preview if available
if result.raw_content:
content_preview = result.raw_content[:1000].replace("\n", " ").strip()
if len(result.raw_content) > 1000:
content_preview += "..."
result_text.append(f" Content: {content_preview}")
# Add metadata at the bottom if available
if self.metadata:
result_text.extend(
[
f"\nMetadata:",
f"- Total results: {self.metadata.total_results}",
f"- Language: {self.metadata.language}",
f"- Country: {self.metadata.country}",
]
)
self.output = "\n".join(result_text)
return self
class WebContentFetcher:
"""Utility class for fetching web content."""
@staticmethod
async def fetch_content(url: str, timeout: int = 10) -> Optional[str]:
"""
Fetch and extract the main content from a webpage.
Args:
url: The URL to fetch content from
timeout: Request timeout in seconds
Returns:
Extracted text content or None if fetching fails
"""
headers = {
"WebSearch": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
}
try:
# Use asyncio to run requests in a thread pool
response = await asyncio.get_event_loop().run_in_executor(
None, lambda: requests.get(url, headers=headers, timeout=timeout)
)
if response.status_code != 200:
logger.warning(
f"Failed to fetch content from {url}: HTTP {response.status_code}"
)
return None
# Parse HTML with BeautifulSoup
soup = BeautifulSoup(response.text, "html.parser")
# Remove script and style elements
for script in soup(["script", "style", "header", "footer", "nav"]):
script.extract()
# Get text content
text = soup.get_text(separator="\n", strip=True)
# Clean up whitespace and limit size (100KB max)
text = " ".join(text.split())
return text[:10000] if text else None
except Exception as e:
logger.warning(f"Error fetching content from {url}: {e}")
return None
class WebSearch(BaseTool):
"""Search the web for information using various search engines."""
name: str = "web_search"
description: str = """Search the web for real-time information about any topic.
This tool returns comprehensive search results with relevant information, URLs, titles, and descriptions.
If the primary search engine fails, it automatically falls back to alternative engines."""
description: str = """Perform a web search and return a list of relevant links.
This function attempts to use the primary search engine API to get up-to-date results.
If an error occurs, it falls back to an alternative search engine."""
parameters: dict = {
"type": "object",
"properties": {
@@ -169,23 +28,8 @@ class WebSearch(BaseTool):
},
"num_results": {
"type": "integer",
"description": "(optional) The number of search results to return. Default is 5.",
"default": 5,
},
"lang": {
"type": "string",
"description": "(optional) Language code for search results (default: en).",
"default": "en",
},
"country": {
"type": "string",
"description": "(optional) Country code for search results (default: us).",
"default": "us",
},
"fetch_content": {
"type": "boolean",
"description": "(optional) Whether to fetch full content from result pages. Default is false.",
"default": False,
"description": "(optional) The number of search results to return. Default is 10.",
"default": 10,
},
},
"required": ["query"],
@@ -196,223 +40,62 @@ class WebSearch(BaseTool):
"duckduckgo": DuckDuckGoSearchEngine(),
"bing": BingSearchEngine(),
}
content_fetcher: WebContentFetcher = WebContentFetcher()
async def execute(
self,
query: str,
num_results: int = 5,
lang: Optional[str] = None,
country: Optional[str] = None,
fetch_content: bool = False,
) -> SearchResponse:
async def execute(self, query: str, num_results: int = 10) -> List[str]:
"""
Execute a Web search and return detailed search results.
Execute a Web search and return a list of URLs.
Args:
query: The search query to submit to the search engine
num_results: The number of search results to return (default: 5)
lang: Language code for search results (default from config)
country: Country code for search results (default from config)
fetch_content: Whether to fetch content from result pages (default: False)
query (str): The search query to submit to the search engine.
num_results (int, optional): The number of search results to return. Default is 10.
Returns:
A structured response containing search results and metadata
List[str]: A list of URLs matching the search query.
"""
# Get settings from config
retry_delay = (
getattr(config.search_config, "retry_delay", 60)
if config.search_config
else 60
)
max_retries = (
getattr(config.search_config, "max_retries", 3)
if config.search_config
else 3
)
# Use config values for lang and country if not specified
if lang is None:
lang = (
getattr(config.search_config, "lang", "en")
if config.search_config
else "en"
)
if country is None:
country = (
getattr(config.search_config, "country", "us")
if config.search_config
else "us"
)
search_params = {"lang": lang, "country": country}
# Try searching with retries when all engines fail
for retry_count in range(max_retries + 1):
results = await self._try_all_engines(query, num_results, search_params)
if results:
# Fetch content if requested
if fetch_content:
results = await self._fetch_content_for_results(results)
# Return a successful structured response
return SearchResponse(
status="success",
query=query,
results=results,
metadata=SearchMetadata(
total_results=len(results),
language=lang,
country=country,
),
)
if retry_count < max_retries:
# All engines failed, wait and retry
logger.warning(
f"All search engines failed. Waiting {retry_delay} seconds before retry {retry_count + 1}/{max_retries}..."
)
await asyncio.sleep(retry_delay)
else:
logger.error(
f"All search engines failed after {max_retries} retries. Giving up."
)
# Return an error response
return SearchResponse(
query=query,
error="All search engines failed to return results after multiple retries.",
results=[],
)
async def _try_all_engines(
self, query: str, num_results: int, search_params: Dict[str, Any]
) -> List[SearchResult]:
"""Try all search engines in the configured order."""
engine_order = self._get_engine_order()
failed_engines = []
for engine_name in engine_order:
engine = self._search_engine[engine_name]
logger.info(f"🔎 Attempting search with {engine_name.capitalize()}...")
search_items = await self._perform_search_with_engine(
engine, query, num_results, search_params
)
if not search_items:
continue
if failed_engines:
logger.info(
f"Search successful with {engine_name.capitalize()} after trying: {', '.join(failed_engines)}"
try:
links = await self._perform_search_with_engine(
engine, query, num_results
)
# Transform search items into structured results
return [
SearchResult(
position=i + 1,
url=item.url,
title=item.title
or f"Result {i+1}", # Ensure we always have a title
description=item.description or "",
source=engine_name,
)
for i, item in enumerate(search_items)
]
if failed_engines:
logger.error(f"All search engines failed: {', '.join(failed_engines)}")
if links:
return links
except Exception as e:
print(f"Search engine '{engine_name}' failed with error: {e}")
return []
async def _fetch_content_for_results(
self, results: List[SearchResult]
) -> List[SearchResult]:
"""Fetch and add web content to search results."""
if not results:
return []
# Create tasks for each result
tasks = [self._fetch_single_result_content(result) for result in results]
# Type annotation to help type checker
fetched_results = await asyncio.gather(*tasks)
# Explicit validation of return type
return [
(
result
if isinstance(result, SearchResult)
else SearchResult(**result.dict())
)
for result in fetched_results
]
async def _fetch_single_result_content(self, result: SearchResult) -> SearchResult:
"""Fetch content for a single search result."""
if result.url:
content = await self.content_fetcher.fetch_content(result.url)
if content:
result.raw_content = content
return result
def _get_engine_order(self) -> List[str]:
"""Determines the order in which to try search engines."""
preferred = (
getattr(config.search_config, "engine", "google").lower()
if config.search_config
else "google"
)
fallbacks = (
[engine.lower() for engine in config.search_config.fallback_engines]
if config.search_config
and hasattr(config.search_config, "fallback_engines")
else []
)
"""
Determines the order in which to try search engines.
Preferred engine is first (based on configuration), followed by the remaining engines.
# Start with preferred engine, then fallbacks, then remaining engines
engine_order = [preferred] if preferred in self._search_engine else []
engine_order.extend(
[
fb
for fb in fallbacks
if fb in self._search_engine and fb not in engine_order
]
)
engine_order.extend([e for e in self._search_engine if e not in engine_order])
Returns:
List[str]: Ordered list of search engine names.
"""
preferred = "google"
if config.search_config and config.search_config.engine:
preferred = config.search_config.engine.lower()
engine_order = []
if preferred in self._search_engine:
engine_order.append(preferred)
for key in self._search_engine:
if key not in engine_order:
engine_order.append(key)
return engine_order
@retry(
stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10)
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
)
async def _perform_search_with_engine(
self,
engine: WebSearchEngine,
query: str,
num_results: int,
search_params: Dict[str, Any],
) -> List[SearchItem]:
"""Execute search with the given engine and parameters."""
return await asyncio.get_event_loop().run_in_executor(
None,
lambda: list(
engine.perform_search(
query,
num_results=num_results,
lang=search_params.get("lang"),
country=search_params.get("country"),
)
),
) -> List[str]:
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None, lambda: list(engine.perform_search(query, num_results=num_results))
)
if __name__ == "__main__":
web_search = WebSearch()
search_response = asyncio.run(
web_search.execute(
query="Python programming", fetch_content=True, num_results=1
)
)
print(search_response.to_tool_result())
@@ -1,16 +0,0 @@
# Global LLM configuration
[llm]
model = "claude-3-7-sonnet-latest" # The LLM model to use
base_url = "https://api.anthropic.com/v1/" # API endpoint URL
api_key = "YOUR_API_KEY" # Your API key
max_tokens = 8192 # Maximum number of tokens in the response
temperature = 0.0 # Controls randomness
# Optional configuration for specific LLM models
[llm.vision]
model = "claude-3-7-sonnet-20250219" # The vision model to use
base_url = "https://api.anthropic.com/v1/" # API endpoint URL for vision model
api_key = "YOUR_API_KEY" # Your API key for vision model
max_tokens = 8192 # Maximum number of tokens in the response
temperature = 0.0 # Controls randomness for vision model
-18
View File
@@ -1,18 +0,0 @@
# Global LLM configuration
[llm] #AZURE OPENAI:
api_type= 'azure'
model = "gpt-4o-mini" # The LLM model to use
base_url = "{YOUR_AZURE_ENDPOINT.rstrip('/')}/openai/deployments/{AZURE_DEPLOYMENT_ID}" # API endpoint URL
api_key = "YOUR_API_KEY" # Your API key
max_tokens = 8096 # Maximum number of tokens in the response
temperature = 0.0 # Controls randomness
api_version="AZURE API VERSION" #"2024-08-01-preview" # Azure Openai version if AzureOpenai
# Optional configuration for specific LLM models
[llm.vision]
model = "gpt-4o" # The vision model to use
base_url = "{YOUR_AZURE_ENDPOINT.rstrip('/')}/openai/deployments/{AZURE_DEPLOYMENT_ID}"
api_key = "YOUR_API_KEY" # Your API key for vision model
max_tokens = 8192 # Maximum number of tokens in the response
temperature = 0.0 # Controls randomness for vision model
-16
View File
@@ -1,16 +0,0 @@
# Global LLM configuration
[llm]
model = "gemini-2.0-flash" # The LLM model to use
base_url = "https://generativelanguage.googleapis.com/v1beta/openai/" # API endpoint URL
api_key = "YOUR_API_KEY" # Your API key
temperature = 0.0 # Controls randomness
max_tokens = 8096 # Maximum number of tokens in the response
# Optional configuration for specific LLM models for Google
[llm.vision]
model = "gemini-2.0-flash-exp" # The vision model to use
base_url = "https://generativelanguage.googleapis.com/v1beta/openai/" # API endpoint URL for vision model
api_key = "YOUR_API_KEY" # Your API key for vision model
max_tokens = 8192 # Maximum number of tokens in the response
temperature = 0.0 # Controls randomness for vision model
-17
View File
@@ -1,17 +0,0 @@
# Global LLM configuration
[llm] #OLLAMA:
api_type = 'ollama'
model = "llama3.2" # The LLM model to use
base_url = "http://localhost:11434/v1" # API endpoint URL
api_key = "ollama" # Your API key
max_tokens = 4096 # Maximum number of tokens in the response
temperature = 0.0 # Controls randomness
[llm.vision] #OLLAMA VISION:
api_type = 'ollama'
model = "llama3.2-vision" # The vision model to use
base_url = "http://localhost:11434/v1" # API endpoint URL for vision model
api_key = "ollama" # Your API key for vision model
max_tokens = 4096 # Maximum number of tokens in the response
temperature = 0.0 # Controls randomness for vision model
+12 -27
View File
@@ -1,10 +1,10 @@
# Global LLM configuration
[llm]
model = "claude-3-7-sonnet-20250219" # The LLM model to use
base_url = "https://api.anthropic.com/v1/" # API endpoint URL
api_key = "YOUR_API_KEY" # Your API key
max_tokens = 8192 # Maximum number of tokens in the response
temperature = 0.0 # Controls randomness
model = "claude-3-7-sonnet-20250219" # The LLM model to use
base_url = "https://api.anthropic.com/v1/" # API endpoint URL
api_key = "YOUR_API_KEY" # Your API key
max_tokens = 8192 # Maximum number of tokens in the response
temperature = 0.0 # Controls randomness
# [llm] # Amazon Bedrock
# api_type = "aws" # Required
@@ -17,7 +17,7 @@ temperature = 0.0 # Controls randomness
# [llm] #AZURE OPENAI:
# api_type= 'azure'
# model = "YOUR_MODEL_NAME" #"gpt-4o-mini"
# base_url = "{YOUR_AZURE_ENDPOINT.rstrip('/')}/openai/deployments/{AZURE_DEPLOYMENT_ID}"
# base_url = "{YOUR_AZURE_ENDPOINT.rstrip('/')}/openai/deployments/{AZURE_DEPOLYMENT_ID}"
# api_key = "AZURE API KEY"
# max_tokens = 8096
# temperature = 0.0
@@ -33,11 +33,11 @@ temperature = 0.0 # Controls randomness
# Optional configuration for specific LLM models
[llm.vision]
model = "claude-3-7-sonnet-20250219" # The vision model to use
base_url = "https://api.anthropic.com/v1/" # API endpoint URL for vision model
api_key = "YOUR_API_KEY" # Your API key for vision model
max_tokens = 8192 # Maximum number of tokens in the response
temperature = 0.0 # Controls randomness for vision model
model = "claude-3-7-sonnet-20250219" # The vision model to use
base_url = "https://api.anthropic.com/v1/" # API endpoint URL for vision model
api_key = "YOUR_API_KEY" # Your API key for vision model
max_tokens = 8192 # Maximum number of tokens in the response
temperature = 0.0 # Controls randomness for vision model
# [llm.vision] #OLLAMA VISION:
# api_type = 'ollama'
@@ -71,19 +71,8 @@ temperature = 0.0 # Controls randomness for vision mode
# Optional configuration, Search settings.
# [search]
# Search engine for agent to use. Default is "Google", can be set to "Baidu" or "DuckDuckGo" or "Bing".
# Search engine for agent to use. Default is "Google", can be set to "Baidu" or "DuckDuckGo".
#engine = "Google"
# Fallback engine order. Default is ["DuckDuckGo", "Baidu", "Bing"] - will try in this order after primary engine fails.
#fallback_engines = ["DuckDuckGo", "Baidu", "Bing"]
# Seconds to wait before retrying all engines again when they all fail due to rate limits. Default is 60.
#retry_delay = 60
# Maximum number of times to retry all engines when all fail. Default is 3.
#max_retries = 3
# Language code for search results. Options: "en" (English), "zh" (Chinese), etc.
#lang = "en"
# Country code for search results. Options: "us" (United States), "cn" (China), etc.
#country = "us"
## Sandbox configuration
#[sandbox]
@@ -94,7 +83,3 @@ temperature = 0.0 # Controls randomness for vision mode
#cpu_limit = 2.0
#timeout = 300
#network_enabled = true
# MCP (Model Context Protocol) configuration
[mcp]
server_reference = "app.mcp.server" # default server module reference
-3
View File
@@ -1,3 +0,0 @@
"""
OpenManus benchmark system for standardized agent evaluation.
"""
@@ -11,6 +11,6 @@ The Model we use is `claude3.5`.
I need a 7-day Japan itinerary for April 15-23 from Seattle, with a $2500-5000 budget for my fiancée and me. We love historical sites, hidden gems, and Japanese culture (kendo, tea ceremonies, Zen meditation). We want to see Nara's deer and explore cities on foot. I plan to propose during this trip and need a special location recommendation. Please provide a detailed itinerary and a simple HTML travel handbook with maps, attraction descriptions, essential Japanese phrases, and travel tips we can reference throughout our journey.
```
**preview**
![alt text](picturesapan-travel-plan-1.png)
![alt text](./pictures/japan-travel-plan-1.png)
![alt text](picturesapan-travel-plan-2.png)
![alt text](./pictures/japan-travel-plan-2.png)
+30
View File
@@ -0,0 +1,30 @@
import asyncio
from app.agent.ppt import PPTAgent
from app.logger import logger
async def main():
agent = PPTAgent()
try:
prompt = """
1. Lecture slide:
I am a lecturer. I am teaching the machine learning coure for research students. Please generate latex code for lecture slide for different reinforcement learning algorithms.
Note that:
1). Note that the lecture duration is 2 hour, so we need to generate 30 pages.
2). for each reinforcement learning algorithms, the slide should include motivation, problem and intuitive solution and detailed math equations.
3). Please make sure the the lecture have a good self-contain.
"""
if not prompt.strip():
logger.warning("Empty prompt provided.")
return
logger.warning("Processing your request...")
await agent.run(prompt)
logger.info("Request processing completed.")
except KeyboardInterrupt:
logger.warning("Operation interrupted.")
if __name__ == "__main__":
asyncio.run(main())

Before

Width:  |  Height:  |  Size: 164 KiB

After

Width:  |  Height:  |  Size: 164 KiB

Before

Width:  |  Height:  |  Size: 36 KiB

After

Width:  |  Height:  |  Size: 36 KiB

-3
View File
@@ -17,9 +17,6 @@ async def main():
logger.info("Request processing completed.")
except KeyboardInterrupt:
logger.warning("Operation interrupted.")
finally:
# Ensure agent resources are cleaned up before exiting
await agent.cleanup()
if __name__ == "__main__":
+7 -13
View File
@@ -4,38 +4,32 @@ tenacity~=9.0.0
pyyaml~=6.0.2
loguru~=0.7.3
numpy
datasets~=3.4.1
datasets~=3.2.0
fastapi~=0.115.11
tiktoken~=0.9.0
html2text~=2024.2.26
gymnasium~=1.1.1
pillow~=11.1.0
gymnasium~=1.0.0
pillow~=10.4.0
browsergym~=0.13.3
uvicorn~=0.34.0
unidiff~=0.7.5
browser-use~=0.1.40
googlesearch-python~=1.3.0
baidusearch~=1.0.3
duckduckgo_search~=7.5.3
duckduckgo_search~=7.5.1
aiofiles~=24.1.0
pydantic_core~=2.27.2
colorama~=0.4.6
playwright~=1.51.0
playwright~=1.50.0
docker~=7.1.0
pytest~=8.3.5
pytest-asyncio~=0.25.3
mcp~=1.5.0
mcp~=1.4.1
httpx>=0.27.0
tomli>=2.0.0
boto3~=1.37.18
requests~=2.32.3
beautifulsoup4~=4.13.3
huggingface-hub~=0.29.2
setuptools~=75.8.0
boto3~=1.37.16
+2 -1
View File
@@ -2,7 +2,8 @@ import asyncio
import time
from app.agent.manus import Manus
from app.flow.flow_factory import FlowFactory, FlowType
from app.flow.base import FlowType
from app.flow.flow_factory import FlowFactory
from app.logger import logger
+6 -15
View File
@@ -13,14 +13,10 @@ class MCPRunner:
def __init__(self):
self.root_path = config.root_path
self.server_reference = config.mcp_config.server_reference
self.server_script = self.root_path / "app" / "mcp" / "server.py"
self.agent = MCPAgent()
async def initialize(
self,
connection_type: str,
server_url: str | None = None,
) -> None:
async def initialize(self, connection_type: str, server_url: str = None) -> None:
"""Initialize the MCP agent with the appropriate connection."""
logger.info(f"Initializing MCPAgent with {connection_type} connection...")
@@ -28,7 +24,7 @@ class MCPRunner:
await self.agent.initialize(
connection_type="stdio",
command=sys.executable,
args=["-m", self.server_reference],
args=[str(self.server_script)],
)
else: # sse
await self.agent.initialize(connection_type="sse", server_url=server_url)
@@ -51,14 +47,9 @@ class MCPRunner:
async def run_default(self) -> None:
"""Run the agent in default mode."""
prompt = input("Enter your prompt: ")
if not prompt.strip():
logger.warning("Empty prompt provided.")
return
logger.warning("Processing your request...")
await self.agent.run(prompt)
logger.info("Request processing completed.")
await self.agent.run(
"Hello, what tools are available to me? Terminate after you have listed the tools."
)
async def cleanup(self) -> None:
"""Clean up agent resources."""
-11
View File
@@ -1,11 +0,0 @@
# coding: utf-8
# A shortcut to launch OpenManus MCP server, where its introduction also solves other import issues.
from app.mcp.server import MCPServer, parse_args
if __name__ == "__main__":
args = parse_args()
# Create and run server (maintaining original flow)
server = MCPServer()
server.run(transport=args.transport)
+3 -3
View File
@@ -21,10 +21,10 @@ setup(
"pyyaml~=6.0.2",
"loguru~=0.7.3",
"numpy",
"datasets>=3.2,<3.5",
"datasets~=3.2.0",
"html2text~=2024.2.26",
"gymnasium>=1.0,<1.2",
"pillow>=10.4,<11.2",
"gymnasium~=1.0.0",
"pillow~=10.4.0",
"browsergym~=0.13.3",
"uvicorn~=0.34.0",
"unidiff~=0.7.5",