agent工具学习笔记
·
from agentpress.tool import Tool, ToolResult, openapi_schema, usage_example
from agentpress.thread_manager import ThreadManager
import json
class ExpandMessageTool(Tool):
"""Tool for expanding a previous message to the user."""
def __init__(self, thread_id: str, thread_manager: ThreadManager):
super().__init__()
self.thread_manager = thread_manager
self.thread_id = thread_id
@openapi_schema({
"type": "function",
"function": {
"name": "expand_message",
"description": "Expand a message from the previous conversation with the user. Use this tool to expand a message that was truncated in the earlier conversation.",
"parameters": {
"type": "object",
"properties": {
"message_id": {
"type": "string",
"description": "The ID of the message to expand. Must be a UUID."
}
},
"required": ["message_id"]
}
}
})
@usage_example('''
<!-- Example 1: Expand a message that was truncated in the previous conversation -->
<function_calls>
<invoke name="expand_message">
<parameter name="message_id">ecde3a4c-c7dc-4776-ae5c-8209517c5576</parameter>
</invoke>
</function_calls>
<!-- Example 2: Expand a message to create reports or analyze truncated data -->
<function_calls>
<invoke name="expand_message">
<parameter name="message_id">f47ac10b-58cc-4372-a567-0e02b2c3d479</parameter>
</invoke>
</function_calls>
<!-- Example 3: Expand a message when you need the full content for analysis -->
<function_calls>
<invoke name="expand_message">
<parameter name="message_id">550e8400-e29b-41d4-a716-446655440000</parameter>
</invoke>
</function_calls>
''')
async def expand_message(self, message_id: str) -> ToolResult:
"""Expand a message from the previous conversation with the user.
Args:
message_id: The ID of the message to expand
Returns:
ToolResult indicating the message was successfully expanded
"""
try:
client = await self.thread_manager.db.client
message = await client.table('messages').select('*').eq('message_id', message_id).eq('thread_id', self.thread_id).execute()
if not message.data or len(message.data) == 0:
return self.fail_response(f"Message with ID {message_id} not found in thread {self.thread_id}")
message_data = message.data[0]
message_content = message_data['content']
final_content = message_content
if isinstance(message_content, dict) and 'content' in message_content:
final_content = message_content['content']
elif isinstance(message_content, str):
try:
parsed_content = json.loads(message_content)
if isinstance(parsed_content, dict) and 'content' in message_content:
final_content = parsed_content['content']
except json.JSONDecodeError:
pass
return self.success_response({"status": "Message expanded successfully.", "message": final_content})
except Exception as e:
return self.fail_response(f"Error expanding message: {str(e)}")
if __name__ == "__main__":
import asyncio
async def test_expand_message_tool():
expand_message_tool = ExpandMessageTool()
# Test expand message
expand_message_result = await expand_message_tool.expand_message(
message_id="004ab969-ef9a-4656-8aba-e392345227cd"
)
print("Expand message result:", expand_message_result)
asyncio.run(test_expand_message_tool())消息截断问题：当 AI 的响应因为长度限制被截断时，可以使用此工具获取完整内容
上下文扩展：当需要更多历史消息的详细信息时
数据分析：当需要完整数据进行进一步处理时
表名: messages（消息表）
message_id 等于传入的 message_id 参数，thread_id 等于当前对话线程的 ID
选择字段: *（所有字段）目的: 获取特定消息的完整记录
message_content 是从数据库中直接获取的原始消息内容。
直接从数据库查询结果中提取
可能是多种数据类型之一
是未经处理的原始数据
parsed_content 是对 message_content 进行 JSON 解析后的结果。
通过 json.loads(message_content) 生成
只有当 message_content 是 JSON 字符串时才有意义
是解析后的 Python 对象（通常是字典或列表）
import asyncio
from typing import Optional, Dict, Any
import time
import asyncio
from uuid import uuid4
from agentpress.tool import ToolResult, openapi_schema, usage_example
from sandbox.tool_base import SandboxToolsBase
from agentpress.thread_manager import ThreadManager
class SandboxShellTool(SandboxToolsBase):
"""Tool for executing tasks in a Daytona sandbox with browser-use capabilities.
Uses sessions for maintaining state between commands and provides comprehensive process management."""
def __init__(self, project_id: str, thread_manager: ThreadManager):
super().__init__(project_id, thread_manager)
self._sessions: Dict[str, str] = {} # Maps session names to session IDs
self.workspace_path = "/workspace" # Ensure we're always operating in /workspace
async def _ensure_session(self, session_name: str = "default") -> str:
"""Ensure a session exists and return its ID."""
if session_name not in self._sessions:
session_id = str(uuid4())
try:
await self._ensure_sandbox() # Ensure sandbox is initialized
await self.sandbox.process.create_session(session_id)
self._sessions[session_name] = session_id
except Exception as e:
raise RuntimeError(f"Failed to create session: {str(e)}")
return self._sessions[session_name]
async def _cleanup_session(self, session_name: str):
"""Clean up a session if it exists."""
if session_name in self._sessions:
try:
await self._ensure_sandbox() # Ensure sandbox is initialized
await self.sandbox.process.delete_session(self._sessions[session_name])
del self._sessions[session_name]
except Exception as e:
print(f"Warning: Failed to cleanup session {session_name}: {str(e)}")
@openapi_schema({
"type": "function",
"function": {
"name": "execute_command",
"description": "Execute a shell command in the workspace directory. IMPORTANT: Commands are non-blocking by default and run in a tmux session. This is ideal for long-running operations like starting servers or build processes. Uses sessions to maintain state between commands. This tool is essential for running CLI tools, installing packages, and managing system operations.",
"parameters": {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "The shell command to execute. Use this for running CLI tools, installing packages, or system operations. Commands can be chained using &&, ||, and | operators."
},
"folder": {
"type": "string",
"description": "Optional relative path to a subdirectory of /workspace where the command should be executed. Example: 'data/pdfs'"
},
"session_name": {
"type": "string",
"description": "Optional name of the tmux session to use. Use named sessions for related commands that need to maintain state. Defaults to a random session name.",
},
"blocking": {
"type": "boolean",
"description": "Whether to wait for the command to complete. Defaults to false for non-blocking execution.",
"default": False
},
"timeout": {
"type": "integer",
"description": "Optional timeout in seconds for blocking commands. Defaults to 60. Ignored for non-blocking commands.",
"default": 60
}
},
"required": ["command"]
}
}
})
@usage_example('''
<function_calls>
<invoke name="execute_command">
<parameter name="command">npm run dev</parameter>
<parameter name="session_name">dev_server</parameter>
</invoke>
</function_calls>
<!-- Example 2: Running in Specific Directory -->
<function_calls>
<invoke name="execute_command">
<parameter name="command">npm run build</parameter>
<parameter name="folder">frontend</parameter>
<parameter name="session_name">build_process</parameter>
</invoke>
</function_calls>
<!-- Example 3: Blocking command (wait for completion) -->
<function_calls>
<invoke name="execute_command">
<parameter name="command">npm install</parameter>
<parameter name="blocking">true</parameter>
<parameter name="timeout">300</parameter>
</invoke>
</function_calls>
''')
async def execute_command(
self,
command: str,
folder: Optional[str] = None,
session_name: Optional[str] = None,
blocking: bool = False,
timeout: int = 60
) -> ToolResult:
try:
# Ensure sandbox is initialized
await self._ensure_sandbox()
# Set up working directory
cwd = self.workspace_path
if folder:
folder = folder.strip('/')
cwd = f"{self.workspace_path}/{folder}"
# Generate a session name if not provided
if not session_name:
session_name = f"session_{str(uuid4())[:8]}"
# Check if tmux session already exists
check_session = await self._execute_raw_command(f"tmux has-session -t {session_name} 2>/dev/null || echo 'not_exists'")
session_exists = "not_exists" not in check_session.get("output", "")
if not session_exists:
# Create a new tmux session
await self._execute_raw_command(f"tmux new-session -d -s {session_name}")
# Ensure we're in the correct directory and send command to tmux
full_command = f"cd {cwd} && {command}"
wrapped_command = full_command.replace('"', '\\"') # Escape double quotes
if blocking:
# For blocking execution, use a more reliable approach
# Add a unique marker to detect command completion
marker = f"COMMAND_DONE_{str(uuid4())[:8]}"
completion_command = f"{command} ; echo {marker}"
wrapped_completion_command = completion_command.replace('"', '\\"')
# Send the command with completion marker
await self._execute_raw_command(f'tmux send-keys -t {session_name} "cd {cwd} && {wrapped_completion_command}" Enter')
start_time = time.time()
final_output = ""
while (time.time() - start_time) < timeout:
# Wait a shorter interval for more responsive checking
await asyncio.sleep(0.5)
# Check if session still exists (command might have exited)
check_result = await self._execute_raw_command(f"tmux has-session -t {session_name} 2>/dev/null || echo 'ended'")
if "ended" in check_result.get("output", ""):
break
# Get current output and check for our completion marker
output_result = await self._execute_raw_command(f"tmux capture-pane -t {session_name} -p -S - -E -")
current_output = output_result.get("output", "")
if self._is_command_completed(current_output, marker):
final_output = current_output
break
# If we didn't get the marker, capture whatever output we have
if not final_output:
output_result = await self._execute_raw_command(f"tmux capture-pane -t {session_name} -p -S - -E -")
final_output = output_result.get("output", "")
# Kill the session after capture
await self._execute_raw_command(f"tmux kill-session -t {session_name}")
return self.success_response({
"output": final_output,
"session_name": session_name,
"cwd": cwd,
"completed": True
})
else:
# Send command to tmux session for non-blocking execution
await self._execute_raw_command(f'tmux send-keys -t {session_name} "{wrapped_command}" Enter')
# For non-blocking, just return immediately
return self.success_response({
"session_name": session_name,
"cwd": cwd,
"message": f"Command sent to tmux session '{session_name}'. Use check_command_output to view results.",
"completed": False
})
except Exception as e:
# Attempt to clean up session in case of error
if session_name:
try:
await self._execute_raw_command(f"tmux kill-session -t {session_name}")
except:
pass
return self.fail_response(f"Error executing command: {str(e)}")
async def _execute_raw_command(self, command: str) -> Dict[str, Any]:
"""Execute a raw command directly in the sandbox."""
# Ensure session exists for raw commands
session_id = await self._ensure_session("raw_commands")
# Execute command in session
from daytona_sdk import SessionExecuteRequest
req = SessionExecuteRequest(
command=command,
var_async=False,
cwd=self.workspace_path
)
response = await self.sandbox.process.execute_session_command(
session_id=session_id,
req=req,
timeout=30 # Short timeout for utility commands
)
logs = await self.sandbox.process.get_session_command_logs(
session_id=session_id,
command_id=response.cmd_id
)
return {
"output": logs,
"exit_code": response.exit_code
}
@openapi_schema({
"type": "function",
"function": {
"name": "check_command_output",
"description": "Check the output of a previously executed command in a tmux session. Use this to monitor the progress or results of non-blocking commands.",
"parameters": {
"type": "object",
"properties": {
"session_name": {
"type": "string",
"description": "The name of the tmux session to check."
},
"kill_session": {
"type": "boolean",
"description": "Whether to terminate the tmux session after checking. Set to true when you're done with the command.",
"default": False
}
},
"required": ["session_name"]
}
}
})
@usage_example('''
<function_calls>
<invoke name="check_command_output">
<parameter name="session_name">dev_server</parameter>
</invoke>
</function_calls>
<!-- Example 2: Check final output and kill session -->
<function_calls>
<invoke name="check_command_output">
<parameter name="session_name">build_process</parameter>
<parameter name="kill_session">true</parameter>
</invoke>
</function_calls>
''')
async def check_command_output(
self,
session_name: str,
kill_session: bool = False
) -> ToolResult:
try:
# Ensure sandbox is initialized
await self._ensure_sandbox()
# Check if session exists
check_result = await self._execute_raw_command(f"tmux has-session -t {session_name} 2>/dev/null || echo 'not_exists'")
if "not_exists" in check_result.get("output", ""):
return self.fail_response(f"Tmux session '{session_name}' does not exist.")
# Get output from tmux pane
output_result = await self._execute_raw_command(f"tmux capture-pane -t {session_name} -p -S - -E -")
output = output_result.get("output", "")
# Kill session if requested
if kill_session:
await self._execute_raw_command(f"tmux kill-session -t {session_name}")
termination_status = "Session terminated."
else:
termination_status = "Session still running."
return self.success_response({
"output": output,
"session_name": session_name,
"status": termination_status
})
except Exception as e:
return self.fail_response(f"Error checking command output: {str(e)}")
@openapi_schema({
"type": "function",
"function": {
"name": "terminate_command",
"description": "Terminate a running command by killing its tmux session.",
"parameters": {
"type": "object",
"properties": {
"session_name": {
"type": "string",
"description": "The name of the tmux session to terminate."
}
},
"required": ["session_name"]
}
}
})
@usage_example('''
<function_calls>
<invoke name="terminate_command">
<parameter name="session_name">dev_server</parameter>
</invoke>
</function_calls>
''')
async def terminate_command(
self,
session_name: str
) -> ToolResult:
try:
# Ensure sandbox is initialized
await self._ensure_sandbox()
# Check if session exists
check_result = await self._execute_raw_command(f"tmux has-session -t {session_name} 2>/dev/null || echo 'not_exists'")
if "not_exists" in check_result.get("output", ""):
return self.fail_response(f"Tmux session '{session_name}' does not exist.")
# Kill the session
await self._execute_raw_command(f"tmux kill-session -t {session_name}")
return self.success_response({
"message": f"Tmux session '{session_name}' terminated successfully."
})
except Exception as e:
return self.fail_response(f"Error terminating command: {str(e)}")
@openapi_schema({
"type": "function",
"function": {
"name": "list_commands",
"description": "List all running tmux sessions and their status.",
"parameters": {
"type": "object",
"properties": {}
}
}
})
@usage_example('''
<function_calls>
<invoke name="list_commands">
</invoke>
</function_calls>
''')
async def list_commands(self) -> ToolResult:
try:
# Ensure sandbox is initialized
await self._ensure_sandbox()
# List all tmux sessions
result = await self._execute_raw_command("tmux list-sessions 2>/dev/null || echo 'No sessions'")
output = result.get("output", "")
if "No sessions" in output or not output.strip():
return self.success_response({
"message": "No active tmux sessions found.",
"sessions": []
})
# Parse session list
sessions = []
for line in output.split('\n'):
if line.strip():
parts = line.split(':')
if parts:
session_name = parts[0].strip()
sessions.append(session_name)
return self.success_response({
"message": f"Found {len(sessions)} active sessions.",
"sessions": sessions
})
except Exception as e:
return self.fail_response(f"Error listing commands: {str(e)}")
def _is_command_completed(self, current_output: str, marker: str) -> bool:
"""
Check if command execution is completed by comparing marker from end to start.
Args:
current_output: Current output content
marker: Completion marker
Returns:
bool: True if command completed, False otherwise
"""
if not current_output or not marker:
return False
# Find the last complete marker match position to start comparison
# Avoid terminal prompt output at the end
marker_end_pos = -1
for i in range(len(current_output) - len(marker), -1, -1):
if current_output[i:i+len(marker)] == marker:
marker_end_pos = i + len(marker) - 1
break
# Start comparison from found marker position or end of output
if marker_end_pos != -1:
output_idx = marker_end_pos
marker_idx = len(marker) - 1
else:
output_idx = len(current_output) - 1
marker_idx = len(marker) - 1
# Compare characters from end to start
while marker_idx >= 0 and output_idx >= 0:
# Skip newlines in current_output
if current_output[output_idx] == '\n':
output_idx -= 1
continue
# Compare characters
if current_output[output_idx] != marker[marker_idx]:
return False
# Continue comparison
output_idx -= 1
marker_idx -= 1
# If marker not fully matched
if marker_idx >= 0:
return False
# Check if preceded by "echo " (command just started)
check_count = 0
echo_chars = "echo "
echo_idx = len(echo_chars) - 1
while output_idx >= 0 and check_count < 5:
# Skip newlines
if current_output[output_idx] == '\n':
output_idx -= 1
continue
check_count += 1
# Check for "echo " pattern
if echo_idx >= 0 and current_output[output_idx] == echo_chars[echo_idx]:
echo_idx -= 1
else:
echo_idx = len(echo_chars) - 1
output_idx -= 1
# If "echo " found, command just started
if echo_idx < 0:
return False
return True
async def cleanup(self):
"""Clean up all sessions."""
for session_name in list(self._sessions.keys()):
await self._cleanup_session(session_name)
# Also clean up any tmux sessions
try:
await self._ensure_sandbox()
await self._execute_raw_command("tmux kill-server 2>/dev/null || true")
except:
pass会话管理初始化使用 tmux 会话来维持命令执行状态，支持多个并行会话，自动管理会话生命周期
为什么要使用 tmux 会话来维持命令执行状态？
许多命令需要长时间运行，比如：启动开发服务器 (npm run dev)，构建大型项目 (npm run build)，运行测试套件，数据处理任务， 如果不使用会话，AI 需要一直等待命令完成
await execute_command("npm run dev")
使用 tmux 会话，命令在后台运行，立即返回 AI 可以继续执行其他任务多任务并行处理
await execute_command("npm run dev", session_name="dev_server")
cwd = f"{self.workspace_path}/{folder}"默认在 /workspace 目录下执行，支持在子目录中执行命令
，确保命令在正确的上下文中执行
execute_command - 执行命令（非阻塞执行（默认））
async def execute_command(self, command: str, folder: Optional[str] = None,
session_name: Optional[str] = None, blocking: bool = False,
timeout: int = 60) -> ToolResult:
if blocking:阻塞执行
except Exception as e:
# Attempt to clean up session in case of error
if session_name:
try:
await self._execute_raw_command(f"tmux kill-session -t {session_name}")
except:
pass
错误处理和资源清理
check_command_output - 检查命令输出
查看正在运行的命令输出，可选择性终止会话先检测再终止
terminate_command - 终止命令
强制终止正在运行的命令，清理会话资源，不需要检测直接确认终止
它们终止会话的核心代码操作是相同的！
list_commands - 列出命令显示 所有活动的会话，监控系统状态
_is_command_completed智能完成检测使用特殊标记来检测命令是否完成执行，避免误判
项目最核心的一段代码
import os
import json
import asyncio
import datetime
from typing import Optional, Dict, List, Any, AsyncGenerator
from dataclasses import dataclass
from agent.tools.message_tool import MessageTool
from agent.tools.sb_deploy_tool import SandboxDeployTool
from agent.tools.sb_expose_tool import SandboxExposeTool
from agent.tools.web_search_tool import SandboxWebSearchTool
from dotenv import load_dotenv
from utils.config import config
from agent.agent_builder_prompt import get_agent_builder_prompt
from agentpress.thread_manager import ThreadManager
from agentpress.response_processor import ProcessorConfig
from agent.tools.sb_shell_tool import SandboxShellTool
from agent.tools.sb_files_tool import SandboxFilesTool
from agent.tools.data_providers_tool import DataProvidersTool
from agent.tools.expand_msg_tool import ExpandMessageTool
from agent.prompt import get_system_prompt
from utils.logger import logger
from utils.auth_utils import get_account_id_from_thread
from services.billing import check_billing_status
from agent.tools.sb_vision_tool import SandboxVisionTool
from agent.tools.sb_image_edit_tool import SandboxImageEditTool
from agent.tools.sb_presentation_outline_tool import SandboxPresentationOutlineTool
from agent.tools.sb_presentation_tool_v2 import SandboxPresentationToolV2
from services.langfuse import langfuse
from langfuse.client import StatefulTraceClient
from agent.tools.mcp_tool_wrapper import MCPToolWrapper
from agent.tools.task_list_tool import TaskListTool
from agentpress.tool import SchemaType
from agent.tools.sb_sheets_tool import SandboxSheetsTool
from agent.tools.sb_web_dev_tool import SandboxWebDevTool
load_dotenv()
@dataclass
class AgentConfig:
thread_id: str
project_id: str
stream: bool
native_max_auto_continues: int = 25
max_iterations: int = 100
model_name: str = "openrouter/moonshotai/kimi-k2"
enable_thinking: Optional[bool] = False
reasoning_effort: Optional[str] = 'low'
enable_context_manager: bool = True
agent_config: Optional[dict] = None
trace: Optional[StatefulTraceClient] = None
is_agent_builder: Optional[bool] = False
target_agent_id: Optional[str] = None
class ToolManager:
def __init__(self, thread_manager: ThreadManager, project_id: str, thread_id: str):
self.thread_manager = thread_manager
self.project_id = project_id
self.thread_id = thread_id
def register_all_tools(self, agent_id: Optional[str] = None, disabled_tools: Optional[List[str]] = None):
"""Register all available tools by default, with optional exclusions.
Args:
agent_id: Optional agent ID for agent builder tools
disabled_tools: List of tool names to exclude from registration
"""
disabled_tools = disabled_tools or []
logger.debug(f"Registering tools with disabled list: {disabled_tools}")
# Core tools - always enabled
self._register_core_tools()
# Sandbox tools
self._register_sandbox_tools(disabled_tools)
# Data and utility tools
self._register_utility_tools(disabled_tools)
# Agent builder tools - register if agent_id provided
if agent_id:
self._register_agent_builder_tools(agent_id, disabled_tools)
# Browser tool
self._register_browser_tool(disabled_tools)
logger.debug(f"Tool registration complete. Registered tools: {list(self.thread_manager.tool_registry.tools.keys())}")
def _register_core_tools(self):
"""Register core tools that are always available."""
self.thread_manager.add_tool(ExpandMessageTool, thread_id=self.thread_id, thread_manager=self.thread_manager)
self.thread_manager.add_tool(MessageTool)
self.thread_manager.add_tool(TaskListTool, project_id=self.project_id, thread_manager=self.thread_manager, thread_id=self.thread_id)
def _register_sandbox_tools(self, disabled_tools: List[str]):
"""Register sandbox-related tools."""
sandbox_tools = [
('sb_shell_tool', SandboxShellTool, {'project_id': self.project_id, 'thread_manager': self.thread_manager}),
('sb_files_tool', SandboxFilesTool, {'project_id': self.project_id, 'thread_manager': self.thread_manager}),
('sb_deploy_tool', SandboxDeployTool, {'project_id': self.project_id, 'thread_manager': self.thread_manager}),
('sb_expose_tool', SandboxExposeTool, {'project_id': self.project_id, 'thread_manager': self.thread_manager}),
('web_search_tool', SandboxWebSearchTool, {'project_id': self.project_id, 'thread_manager': self.thread_manager}),
('sb_vision_tool', SandboxVisionTool, {'project_id': self.project_id, 'thread_id': self.thread_id, 'thread_manager': self.thread_manager}),
('sb_image_edit_tool', SandboxImageEditTool, {'project_id': self.project_id, 'thread_id': self.thread_id, 'thread_manager': self.thread_manager}),
('sb_presentation_outline_tool', SandboxPresentationOutlineTool, {'project_id': self.project_id, 'thread_manager': self.thread_manager}),
('sb_presentation_tool_v2', SandboxPresentationToolV2, {'project_id': self.project_id, 'thread_manager': self.thread_manager}),
('sb_sheets_tool', SandboxSheetsTool, {'project_id': self.project_id, 'thread_manager': self.thread_manager}),
('sb_web_dev_tool', SandboxWebDevTool, {'project_id': self.project_id, 'thread_id': self.thread_id, 'thread_manager': self.thread_manager}),
]
for tool_name, tool_class, kwargs in sandbox_tools:
if tool_name not in disabled_tools:
self.thread_manager.add_tool(tool_class, **kwargs)
logger.debug(f"Registered {tool_name}")
def _register_utility_tools(self, disabled_tools: List[str]):
"""Register utility and data provider tools."""
if config.RAPID_API_KEY and 'data_providers_tool' not in disabled_tools:
self.thread_manager.add_tool(DataProvidersTool)
logger.debug("Registered data_providers_tool")
def _register_agent_builder_tools(self, agent_id: str, disabled_tools: List[str]):
"""Register agent builder tools."""
from agent.tools.agent_builder_tools.agent_config_tool import AgentConfigTool
from agent.tools.agent_builder_tools.mcp_search_tool import MCPSearchTool
from agent.tools.agent_builder_tools.credential_profile_tool import CredentialProfileTool
from agent.tools.agent_builder_tools.workflow_tool import WorkflowTool
from agent.tools.agent_builder_tools.trigger_tool import TriggerTool
from services.supabase import DBConnection
db = DBConnection()
agent_builder_tools = [
('agent_config_tool', AgentConfigTool),
('mcp_search_tool', MCPSearchTool),
('credential_profile_tool', CredentialProfileTool),
('workflow_tool', WorkflowTool),
('trigger_tool', TriggerTool),
]
for tool_name, tool_class in agent_builder_tools:
if tool_name not in disabled_tools:
self.thread_manager.add_tool(tool_class, thread_manager=self.thread_manager, db_connection=db, agent_id=agent_id)
logger.debug(f"Registered {tool_name}")
def _register_browser_tool(self, disabled_tools: List[str]):
"""Register browser tool."""
if 'browser_tool' not in disabled_tools:
from agent.tools.browser_tool import BrowserTool
self.thread_manager.add_tool(BrowserTool, project_id=self.project_id, thread_id=self.thread_id, thread_manager=self.thread_manager)
logger.debug("Registered browser_tool")
class MCPManager:
def __init__(self, thread_manager: ThreadManager, account_id: str):
self.thread_manager = thread_manager
self.account_id = account_id
async def register_mcp_tools(self, agent_config: dict) -> Optional[MCPToolWrapper]:
all_mcps = []
if agent_config.get('configured_mcps'):
all_mcps.extend(agent_config['configured_mcps'])
if agent_config.get('custom_mcps'):
for custom_mcp in agent_config['custom_mcps']:
custom_type = custom_mcp.get('customType', custom_mcp.get('type', 'sse'))
if custom_type == 'pipedream':
if 'config' not in custom_mcp:
custom_mcp['config'] = {}
if not custom_mcp['config'].get('external_user_id'):
profile_id = custom_mcp['config'].get('profile_id')
if profile_id:
try:
from pipedream import profile_service
from uuid import UUID
profile = await profile_service.get_profile(UUID(self.account_id), UUID(profile_id))
if profile:
custom_mcp['config']['external_user_id'] = profile.external_user_id
except Exception as e:
logger.error(f"Error retrieving external_user_id from profile {profile_id}: {e}")
if 'headers' in custom_mcp['config'] and 'x-pd-app-slug' in custom_mcp['config']['headers']:
custom_mcp['config']['app_slug'] = custom_mcp['config']['headers']['x-pd-app-slug']
elif custom_type == 'composio':
qualified_name = custom_mcp.get('qualifiedName')
if not qualified_name:
qualified_name = f"composio.{custom_mcp['name'].replace(' ', '_').lower()}"
mcp_config = {
'name': custom_mcp['name'],
'qualifiedName': qualified_name,
'config': custom_mcp.get('config', {}),
'enabledTools': custom_mcp.get('enabledTools', []),
'instructions': custom_mcp.get('instructions', ''),
'isCustom': True,
'customType': 'composio'
}
all_mcps.append(mcp_config)
continue
mcp_config = {
'name': custom_mcp['name'],
'qualifiedName': f"custom_{custom_type}_{custom_mcp['name'].replace(' ', '_').lower()}",
'config': custom_mcp['config'],
'enabledTools': custom_mcp.get('enabledTools', []),
'instructions': custom_mcp.get('instructions', ''),
'isCustom': True,
'customType': custom_type
}
all_mcps.append(mcp_config)
if not all_mcps:
return None
mcp_wrapper_instance = MCPToolWrapper(mcp_configs=all_mcps)
try:
await mcp_wrapper_instance.initialize_and_register_tools()
updated_schemas = mcp_wrapper_instance.get_schemas()
for method_name, schema_list in updated_schemas.items():
for schema in schema_list:
self.thread_manager.tool_registry.tools[method_name] = {
"instance": mcp_wrapper_instance,
"schema": schema
}
logger.debug(f"⚡ Registered {len(updated_schemas)} MCP tools (Redis cache enabled)")
return mcp_wrapper_instance
except Exception as e:
logger.error(f"Failed to initialize MCP tools: {e}")
return None
class PromptManager:
@staticmethod
async def build_system_prompt(model_name: str, agent_config: Optional[dict],
is_agent_builder: bool, thread_id: str,
mcp_wrapper_instance: Optional[MCPToolWrapper],
client=None) -> dict:
default_system_content = get_system_prompt()
if "anthropic" not in model_name.lower():
sample_response_path = os.path.join(os.path.dirname(__file__), 'sample_responses/1.txt')
with open(sample_response_path, 'r') as file:
sample_response = file.read()
default_system_content = default_system_content + "\n\n <sample_assistant_response>" + sample_response + "</sample_assistant_response>"
if is_agent_builder:
system_content = get_agent_builder_prompt()
elif agent_config and agent_config.get('system_prompt'):
system_content = agent_config['system_prompt'].strip()
else:
system_content = default_system_content
# Add agent knowledge base context if available
if client and agent_config and agent_config.get('agent_id'):
try:
logger.debug(f"Retrieving agent knowledge base context for agent {agent_config['agent_id']}")
# Use only agent-based knowledge base context
kb_result = await client.rpc('get_agent_knowledge_base_context', {
'p_agent_id': agent_config['agent_id']
}).execute()
if kb_result.data and kb_result.data.strip():
logger.debug(f"Found agent knowledge base context, adding to system prompt (length: {len(kb_result.data)} chars)")
# logger.debug(f"Knowledge base data object: {kb_result.data[:500]}..." if len(kb_result.data) > 500 else f"Knowledge base data object: {kb_result.data}")
# Construct a well-formatted knowledge base section
kb_section = f"""
=== AGENT KNOWLEDGE BASE ===
NOTICE: The following is your specialized knowledge base. This information should be considered authoritative for your responses and should take precedence over general knowledge when relevant.
{kb_result.data}
=== END AGENT KNOWLEDGE BASE ===
IMPORTANT: Always reference and utilize the knowledge base information above when it's relevant to user queries. This knowledge is specific to your role and capabilities."""
system_content += kb_section
else:
logger.debug("No knowledge base context found for this agent")
except Exception as e:
logger.error(f"Error retrieving knowledge base context for agent {agent_config.get('agent_id', 'unknown')}: {e}")
# Continue without knowledge base context rather than failing
if agent_config and (agent_config.get('configured_mcps') or agent_config.get('custom_mcps')) and mcp_wrapper_instance and mcp_wrapper_instance._initialized:
mcp_info = "\n\n--- MCP Tools Available ---\n"
mcp_info += "You have access to external MCP (Model Context Protocol) server tools.\n"
mcp_info += "MCP tools can be called directly using their native function names in the standard function calling format:\n"
mcp_info += '<function_calls>\n'
mcp_info += '<invoke name="{tool_name}">\n'
mcp_info += '<parameter name="param1">value1</parameter>\n'
mcp_info += '<parameter name="param2">value2</parameter>\n'
mcp_info += '</invoke>\n'
mcp_info += '</function_calls>\n\n'
mcp_info += "Available MCP tools:\n"
try:
registered_schemas = mcp_wrapper_instance.get_schemas()
for method_name, schema_list in registered_schemas.items():
for schema in schema_list:
if schema.schema_type == SchemaType.OPENAPI:
func_info = schema.schema.get('function', {})
description = func_info.get('description', 'No description available')
mcp_info += f"- **{method_name}**: {description}\n"
params = func_info.get('parameters', {})
props = params.get('properties', {})
if props:
mcp_info += f" Parameters: {', '.join(props.keys())}\n"
except Exception as e:
logger.error(f"Error listing MCP tools: {e}")
mcp_info += "- Error loading MCP tool list\n"
mcp_info += "\n🚨 CRITICAL MCP TOOL RESULT INSTRUCTIONS 🚨\n"
mcp_info += "When you use ANY MCP (Model Context Protocol) tools:\n"
mcp_info += "1. ALWAYS read and use the EXACT results returned by the MCP tool\n"
mcp_info += "2. For search tools: ONLY cite URLs, sources, and information from the actual search results\n"
mcp_info += "3. For any tool: Base your response entirely on the tool's output - do NOT add external information\n"
mcp_info += "4. DO NOT fabricate, invent, hallucinate, or make up any sources, URLs, or data\n"
mcp_info += "5. If you need more information, call the MCP tool again with different parameters\n"
mcp_info += "6. When writing reports/summaries: Reference ONLY the data from MCP tool results\n"
mcp_info += "7. If the MCP tool doesn't return enough information, explicitly state this limitation\n"
mcp_info += "8. Always double-check that every fact, URL, and reference comes from the MCP tool output\n"
mcp_info += "\nIMPORTANT: MCP tool results are your PRIMARY and ONLY source of truth for external data!\n"
mcp_info += "NEVER supplement MCP results with your training data or make assumptions beyond what the tools provide.\n"
system_content += mcp_info
now = datetime.datetime.now(datetime.timezone.utc)
datetime_info = f"\n\n=== CURRENT DATE/TIME INFORMATION ===\n"
datetime_info += f"Today's date: {now.strftime('%A, %B %d, %Y')}\n"
datetime_info += f"Current UTC time: {now.strftime('%H:%M:%S UTC')}\n"
datetime_info += f"Current year: {now.strftime('%Y')}\n"
datetime_info += f"Current month: {now.strftime('%B')}\n"
datetime_info += f"Current day: {now.strftime('%A')}\n"
datetime_info += "Use this information for any time-sensitive tasks, research, or when current date/time context is needed.\n"
system_content += datetime_info
return {"role": "system", "content": system_content}
class MessageManager:
def __init__(self, client, thread_id: str, model_name: str, trace: Optional[StatefulTraceClient]):
self.client = client
self.thread_id = thread_id
self.model_name = model_name
self.trace = trace
async def build_temporary_message(self) -> Optional[dict]:
temp_message_content_list = []
latest_browser_state_msg = await self.client.table('messages').select('*').eq('thread_id', self.thread_id).eq('type', 'browser_state').order('created_at', desc=True).limit(1).execute()
if latest_browser_state_msg.data and len(latest_browser_state_msg.data) > 0:
try:
browser_content = latest_browser_state_msg.data[0]["content"]
if isinstance(browser_content, str):
browser_content = json.loads(browser_content)
screenshot_base64 = browser_content.get("screenshot_base64")
screenshot_url = browser_content.get("image_url")
browser_state_text = browser_content.copy()
browser_state_text.pop('screenshot_base64', None)
browser_state_text.pop('image_url', None)
if browser_state_text:
temp_message_content_list.append({
"type": "text",
"text": f"The following is the current state of the browser:\n{json.dumps(browser_state_text, indent=2)}"
})
if 'gemini' in self.model_name.lower() or 'anthropic' in self.model_name.lower() or 'openai' in self.model_name.lower():
if screenshot_url:
temp_message_content_list.append({
"type": "image_url",
"image_url": {
"url": screenshot_url,
"format": "image/png"
}
})
elif screenshot_base64:
temp_message_content_list.append({
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{screenshot_base64}",
}
})
except Exception as e:
logger.error(f"Error parsing browser state: {e}")
latest_image_context_msg = await self.client.table('messages').select('*').eq('thread_id', self.thread_id).eq('type', 'image_context').order('created_at', desc=True).limit(1).execute()
if latest_image_context_msg.data and len(latest_image_context_msg.data) > 0:
try:
image_context_content = latest_image_context_msg.data[0]["content"] if isinstance(latest_image_context_msg.data[0]["content"], dict) else json.loads(latest_image_context_msg.data[0]["content"])
base64_image = image_context_content.get("base64")
mime_type = image_context_content.get("mime_type")
file_path = image_context_content.get("file_path", "unknown file")
if base64_image and mime_type:
temp_message_content_list.append({
"type": "text",
"text": f"Here is the image you requested to see: '{file_path}'"
})
temp_message_content_list.append({
"type": "image_url",
"image_url": {
"url": f"data:{mime_type};base64,{base64_image}",
}
})
await self.client.table('messages').delete().eq('message_id', latest_image_context_msg.data[0]["message_id"]).execute()
except Exception as e:
logger.error(f"Error parsing image context: {e}")
if temp_message_content_list:
return {"role": "user", "content": temp_message_content_list}
return None
class AgentRunner:
def __init__(self, config: AgentConfig):
self.config = config
async def setup(self):
if not self.config.trace:
self.config.trace = langfuse.trace(name="run_agent", session_id=self.config.thread_id, metadata={"project_id": self.config.project_id})
self.thread_manager = ThreadManager(
trace=self.config.trace,
is_agent_builder=self.config.is_agent_builder or False,
target_agent_id=self.config.target_agent_id,
agent_config=self.config.agent_config
)
self.client = await self.thread_manager.db.client
self.account_id = await get_account_id_from_thread(self.client, self.config.thread_id)
if not self.account_id:
raise ValueError("Could not determine account ID for thread")
project = await self.client.table('projects').select('*').eq('project_id', self.config.project_id).execute()
if not project.data or len(project.data) == 0:
raise ValueError(f"Project {self.config.project_id} not found")
project_data = project.data[0]
sandbox_info = project_data.get('sandbox', {})
if not sandbox_info.get('id'):
# Sandbox is created lazily by tools when required. Do not fail setup
# if no sandbox is present — tools will call `_ensure_sandbox()`
# which will create and persist the sandbox metadata when needed.
logger.debug(f"No sandbox found for project {self.config.project_id}; will create lazily when needed")
async def setup_tools(self):
tool_manager = ToolManager(self.thread_manager, self.config.project_id, self.config.thread_id)
# Determine agent ID for agent builder tools
agent_id = None
if self.config.agent_config and self.config.agent_config.get('is_suna_default', False):
agent_id = self.config.agent_config['agent_id']
elif self.config.is_agent_builder and self.config.target_agent_id:
agent_id = self.config.target_agent_id
# Convert agent config to disabled tools list
disabled_tools = self._get_disabled_tools_from_config()
# Register all tools with exclusions
tool_manager.register_all_tools(agent_id=agent_id, disabled_tools=disabled_tools)
def _get_disabled_tools_from_config(self) -> List[str]:
"""Convert agent config to list of disabled tools."""
disabled_tools = []
if not self.config.agent_config or 'agentpress_tools' not in self.config.agent_config:
# No tool configuration - enable all tools by default
return disabled_tools
raw_tools = self.config.agent_config['agentpress_tools']
# Handle different formats of tool configuration
if not isinstance(raw_tools, dict):
# If not a dict, assume all tools are enabled
return disabled_tools
# Special case: Suna default agents with empty tool config enable all tools
if self.config.agent_config.get('is_suna_default', False) and not raw_tools:
return disabled_tools
def is_tool_enabled(tool_name: str) -> bool:
try:
tool_config = raw_tools.get(tool_name, True) # Default to True (enabled) if not specified
if isinstance(tool_config, bool):
return tool_config
elif isinstance(tool_config, dict):
return tool_config.get('enabled', True) # Default to True (enabled) if not specified
else:
return True # Default to enabled
except Exception:
return True # Default to enabled
# List of all available tools
all_tools = [
'sb_shell_tool', 'sb_files_tool', 'sb_deploy_tool', 'sb_expose_tool',
'web_search_tool', 'sb_vision_tool', 'sb_presentation_tool', 'sb_image_edit_tool',
'sb_sheets_tool', 'sb_web_dev_tool', 'data_providers_tool', 'browser_tool',
'agent_config_tool', 'mcp_search_tool', 'credential_profile_tool',
'workflow_tool', 'trigger_tool'
]
# Add tools that are explicitly disabled
for tool_name in all_tools:
if not is_tool_enabled(tool_name):
disabled_tools.append(tool_name)
# Special handling for presentation tools
if 'sb_presentation_tool' in disabled_tools:
disabled_tools.extend(['sb_presentation_outline_tool', 'sb_presentation_tool_v2'])
logger.debug(f"Disabled tools from config: {disabled_tools}")
return disabled_tools
async def setup_mcp_tools(self) -> Optional[MCPToolWrapper]:
if not self.config.agent_config:
return None
mcp_manager = MCPManager(self.thread_manager, self.account_id)
return await mcp_manager.register_mcp_tools(self.config.agent_config)
def get_max_tokens(self) -> Optional[int]:
if "sonnet" in self.config.model_name.lower():
return 8192
elif "gpt-4" in self.config.model_name.lower():
return 4096
elif "gemini-2.5-pro" in self.config.model_name.lower():
return 64000
elif "kimi-k2" in self.config.model_name.lower():
return 8192
return None
async def run(self) -> AsyncGenerator[Dict[str, Any], None]:
await self.setup()
await self.setup_tools()
mcp_wrapper_instance = await self.setup_mcp_tools()
system_message = await PromptManager.build_system_prompt(
self.config.model_name, self.config.agent_config,
self.config.is_agent_builder, self.config.thread_id,
mcp_wrapper_instance, self.client
)
iteration_count = 0
continue_execution = True
latest_user_message = await self.client.table('messages').select('*').eq('thread_id', self.config.thread_id).eq('type', 'user').order('created_at', desc=True).limit(1).execute()
if latest_user_message.data and len(latest_user_message.data) > 0:
data = latest_user_message.data[0]['content']
if isinstance(data, str):
data = json.loads(data)
if self.config.trace:
self.config.trace.update(input=data['content'])
message_manager = MessageManager(self.client, self.config.thread_id, self.config.model_name, self.config.trace)
while continue_execution and iteration_count < self.config.max_iterations:
iteration_count += 1
can_run, message, subscription = await check_billing_status(self.client, self.account_id)
if not can_run:
error_msg = f"Billing limit reached: {message}"
yield {
"type": "status",
"status": "stopped",
"message": error_msg
}
break
latest_message = await self.client.table('messages').select('*').eq('thread_id', self.config.thread_id).in_('type', ['assistant', 'tool', 'user']).order('created_at', desc=True).limit(1).execute()
if latest_message.data and len(latest_message.data) > 0:
message_type = latest_message.data[0].get('type')
if message_type == 'assistant':
continue_execution = False
break
temporary_message = await message_manager.build_temporary_message()
max_tokens = self.get_max_tokens()
generation = self.config.trace.generation(name="thread_manager.run_thread") if self.config.trace else None
try:
response = await self.thread_manager.run_thread(
thread_id=self.config.thread_id,
system_prompt=system_message,
stream=self.config.stream,
llm_model=self.config.model_name,
llm_temperature=0,
llm_max_tokens=max_tokens,
tool_choice="auto",
max_xml_tool_calls=1,
temporary_message=temporary_message,
processor_config=ProcessorConfig(
xml_tool_calling=True,
native_tool_calling=False,
execute_tools=True,
execute_on_stream=True,
tool_execution_strategy="parallel",
xml_adding_strategy="user_message"
),
native_max_auto_continues=self.config.native_max_auto_continues,
include_xml_examples=True,
enable_thinking=self.config.enable_thinking,
reasoning_effort=self.config.reasoning_effort,
enable_context_manager=self.config.enable_context_manager,
generation=generation
)
if isinstance(response, dict) and "status" in response and response["status"] == "error":
yield response
break
last_tool_call = None
agent_should_terminate = False
error_detected = False
full_response = ""
try:
if hasattr(response, '__aiter__') and not isinstance(response, dict):
async for chunk in response:
if isinstance(chunk, dict) and chunk.get('type') == 'status' and chunk.get('status') == 'error':
error_detected = True
yield chunk
continue
if chunk.get('type') == 'status':
try:
metadata = chunk.get('metadata', {})
if isinstance(metadata, str):
metadata = json.loads(metadata)
if metadata.get('agent_should_terminate'):
agent_should_terminate = True
content = chunk.get('content', {})
if isinstance(content, str):
content = json.loads(content)
if content.get('function_name'):
last_tool_call = content['function_name']
elif content.get('xml_tag_name'):
last_tool_call = content['xml_tag_name']
except Exception:
pass
if chunk.get('type') == 'assistant' and 'content' in chunk:
try:
content = chunk.get('content', '{}')
if isinstance(content, str):
assistant_content_json = json.loads(content)
else:
assistant_content_json = content
assistant_text = assistant_content_json.get('content', '')
full_response += assistant_text
if isinstance(assistant_text, str):
if '</ask>' in assistant_text or '</complete>' in assistant_text or '</web-browser-takeover>' in assistant_text:
if '</ask>' in assistant_text:
xml_tool = 'ask'
elif '</complete>' in assistant_text:
xml_tool = 'complete'
elif '</web-browser-takeover>' in assistant_text:
xml_tool = 'web-browser-takeover'
last_tool_call = xml_tool
except json.JSONDecodeError:
pass
except Exception:
pass
yield chunk
else:
error_detected = True
if error_detected:
if generation:
generation.end(output=full_response, status_message="error_detected", level="ERROR")
break
if agent_should_terminate or last_tool_call in ['ask', 'complete', 'web-browser-takeover']:
if generation:
generation.end(output=full_response, status_message="agent_stopped")
continue_execution = False
except Exception as e:
error_msg = f"Error during response streaming: {str(e)}"
if generation:
generation.end(output=full_response, status_message=error_msg, level="ERROR")
yield {
"type": "status",
"status": "error",
"message": error_msg
}
break
except Exception as e:
error_msg = f"Error running thread: {str(e)}"
yield {
"type": "status",
"status": "error",
"message": error_msg
}
break
if generation:
generation.end(output=full_response)
asyncio.create_task(asyncio.to_thread(lambda: langfuse.flush()))
async def run_agent(
thread_id: str,
project_id: str,
stream: bool,
thread_manager: Optional[ThreadManager] = None,
native_max_auto_continues: int = 25,
max_iterations: int = 100,
model_name: str = "openrouter/moonshotai/kimi-k2",
enable_thinking: Optional[bool] = False,
reasoning_effort: Optional[str] = 'low',
enable_context_manager: bool = True,
agent_config: Optional[dict] = None,
trace: Optional[StatefulTraceClient] = None,
is_agent_builder: Optional[bool] = False,
target_agent_id: Optional[str] = None
):
effective_model = model_name
if model_name == "openrouter/moonshotai/kimi-k2" and agent_config and agent_config.get('model'):
effective_model = agent_config['model']
logger.debug(f"Using model from agent config: {effective_model} (no user selection)")
elif model_name != "openrouter/moonshotai/kimi-k2":
logger.debug(f"Using user-selected model: {effective_model}")
else:
logger.debug(f"Using default model: {effective_model}")
config = AgentConfig(
thread_id=thread_id,
project_id=project_id,
stream=stream,
native_max_auto_continues=native_max_auto_continues,
max_iterations=max_iterations,
model_name=effective_model,
enable_thinking=enable_thinking,
reasoning_effort=reasoning_effort,
enable_context_manager=enable_context_manager,
agent_config=agent_config,
trace=trace,
is_agent_builder=is_agent_builder,
target_agent_id=target_agent_id
)
runner = AgentRunner(config)
async for chunk in runner.run():
yield chunkasync def run_agent(...):这是整个系统的主入口，所有 Agent 执行都从这里开始
# 创建配置
config = AgentConfig(...)
# 启动 runner
runner = AgentRunner(config)
async for chunk in runner.run():
yield chunk
class ToolManager:
def register_all_tools(self, agent_id: Optional[str] = None, disabled_tools: Optional[List[str]] = None):
# 注册所有工具
self._register_core_tools()
self._register_sandbox_tools(disabled_tools)
管理所有可用工具的注册，根据配置启用/禁用工具，为 Agent 提供工具访问能力
通过 continue_execution 标志控制是否继续执行
while continue_execution and iteration_count < self.config.max_iterations:
iteration_count += 1
账单状态检查
can_run, message, subscription = await check_billing_status(self.client, self.account_id)
if not can_run:检查用户的账单状态,确保用户有权限继续使用服务,防止滥用和未付费使用
停止执行并返回错误信息
消息状态检查
latest_message = await self.client.table('messages').select('*').eq('thread_id', self.config.thread_id).in_('type', ['assistant', 'tool', 'user']).order('created_at', desc=True).limit(1).execute()
if latest_message.data and len(latest_message.data) > 0:
message_type = latest_message.data[0].get('type')
if message_type == 'assistant':
continue_execution = False
break
检查最新的消息类型,如果最新消息是 assistant 类型，说明任务已完成,避免重复执行已完成的任务
用户: "帮我创建一个网站"
AI: "好的，我已经为您创建了网站，链接是..."
[循环检测到最新消息是 assistant 类型(一个状态值)，停止执行]
临时消息构建
temporary_message = await message_manager.build_temporary_message()
构建临时上下文信息,包括浏览器状态、图像上下文等,为当前推理提供额外信息
response = await self.thread_manager.run_thread(...)
这是整个循环最核心的部分，调用 ThreadManager 执行实际的推理过程
processor_config=ProcessorConfig(
xml_tool_calling=True, # 使用 XML 格式工具调用
native_tool_calling=False, # 不使用原生工具调用
execute_tools=True, # 执行工具
execute_on_stream=True, # 流式执行
tool_execution_strategy="parallel", # 并行执行工具
xml_adding_strategy="user_message" # XML 添加策略
)
async for chunk in response:
# 处理每个响应块
if isinstance(chunk, dict) and chunk.get('type') == 'status' and chunk.get('status') == 'error':
error_detected = True
yield chunk
continue
# 检查终止条件
if chunk.get('type') == 'status':
try:
metadata = chunk.get('metadata', {})
if isinstance(metadata, str):
metadata = json.loads(metadata)
if metadata.get('agent_should_terminate'):
agent_should_terminate = True
# 提取工具调用信息
except Exception:
pass
# 处理助手响应
if chunk.get('type') == 'assistant' and 'content' in chunk:
# 解析和累积响应内容
pass
# 流式输出
yield chunk
if agent_should_terminate or last_tool_call in ['ask', 'complete', 'web-browser-takeover']:
if generation:
generation.end(output=full_response, status_message="agent_stopped")
continue_execution = False
调用了特定的终止工具：ask：询问用户,complete：任务完成,web-browser-takeover：浏览器接管
ThreadManager - 线程管理器
def __init__(self, trace: Optional[StatefulTraceClient] = None, is_agent_builder: bool = False, target_agent_id: Optional[str] = None, agent_config: Optional[dict] = None):
"""Initialize ThreadManager.
Args:
trace: Optional trace client for logging
is_agent_builder: Whether this is an agent builder session
target_agent_id: ID of the agent being built (if in agent builder mode)
agent_config: Optional agent configuration with version information
"""
self.db = DBConnection()
self.tool_registry = ToolRegistry()
self.trace = trace
self.is_agent_builder = is_agent_builder
self.target_agent_id = target_agent_id
self.agent_config = agent_config
if not self.trace:
self.trace = langfuse.trace(name="anonymous:thread_manager")
self.response_processor = ResponseProcessor(
tool_registry=self.tool_registry,
add_message_callback=self.add_message,
trace=self.trace,
is_agent_builder=self.is_agent_builder,
target_agent_id=self.target_agent_id,
agent_config=self.agent_config
)
self.context_manager = ContextManager()# 数据库连接 # 工具注册表 # 追踪系统
# 管理特定线程的对话历史
async def get_thread_messages(self, thread_id: str) -> List[Dict]:
# 从数据库获取对话历史
pass
async def add_message(self, thread_id: str, message: Dict) -> str:
# 向对话线程添加消息
pass
async def run_thread(self,
thread_id: str,
system_prompt: dict,
stream: bool = True,
llm_model: str = "default",
# ... 其他参数
) -> Union[AsyncGenerator, Dict]:
# 协调整个推理过程
# 1. 构建上下文
# 2. 调用 LLM
# 3. 处理响应
# 4. 执行工具调用
pass
PromptManager - 提示词管理器
核心工具类
SandboxShellTool - Shell 命令执行
MessageTool - 消息处理
TaskListTool - 任务管理
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
22
0- 0
已为社区贡献3条内容
所有评论(0)