aipy，话说的对标着开源的manus，国内能用，内网能用，开源免费的工具。
仓库地址：https://github.com/knownsec/aipyapp
官网：https://www.aipyaipy.com/
开源版本的代码阅读如下
首先来看main函数下的命令，可以通过不同的命令来启动不同的服务。
agent 启动Agent模型的HTTP服务
python 启动Python交互模式
ipython 启动IPython交互模式
gui/mainw 启动gui图形界面
run/默认 启动默认的任务模式，执行命令 json。
从Agent模型的HTTP服务入手，该HTTP服务就是简单的以fastapi为框架，包含了一个任务管理器。核心代码主要集中在任务和任务管理两者主干逻辑上。
核心代码为（aipyapp/aipy/taskmgr.py）
class TaskManager:
MAX_TASKS = 16
def __init__(self, settings, /, display_manager=None):
# 核心配置
self.settings = settings
self.display_manager = display_manager
self.log = logger.bind(src='taskmgr')
# 任务管理
self.tasks = deque(maxlen=self.MAX_TASKS)
# 工作环境
self._init_workenv()
# 初始化各种管理器
self._init_managers()
该类限制了最多有16个任务。其中包含了两个重要的步骤：
初始化工作环境 _init_workenv， 就是简单地针对不同的task进行 os.chdir 以实现切换不同路径的操作。
初始化管理器 _init_managers， 有插件管理，诊断器，客户端管理器，角色管理器，MCP工具管理器，提示词管理器。
tips： Python的 collections.namedtuple() 用于创建具有命名字段的元组子类的工厂函数
接下来，逐个拆解管理器的内容
核心代码在 aipyapp/aipy/plugins.py 下
class PluginManager:
"""Plugin manager class."""
FILE_PATTERN = "p_*.py" # 所有p_*.py形式的文件都是插件文件
def __init__(self):
self.sys_plugin_dir = os.path.join(__pkgpath__, 'plugins')
self.plugin_directories: List[Path] = []
self._plugins: Dict[str, Any] = {}
self.logger = logger.bind(src=self.__class__.__name__)
self.add_plugin_directory(self.sys_plugin_dir) # 把插件文件夹的路径加入到list中
该类负责管理 aipyapp/plugins 下的所有插件，每一个文件就是一个类型的插件。
插件管理器加载插件代码的模块，使用Python的动态加载脚本来实现
try:
spec = importlib.util.spec_from_file_location(name, filepath)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
plugins = []
for cls in module.__dict__.values():
if isinstance(cls, type) and issubclass(cls, Plugin) and cls.name:
plugins.append(cls)
return plugins
核心代码在 aipyapp/aipy/diagnose.py 下
class Diagnose:
def __init__(self, api_url, api_key):
self._api_url = api_url
self._api_key = api_key
self.log = logger.bind(src='diagnose')
self.load_config()
检查当前Agent在执行的时候发生的报错
核心代码在 aipyapp/llm/manager.py 下
class ClientManager(object):
MAX_TOKENS = 8192
def __init__(self, settings: dict, max_tokens: int | None = None):
self.clients = {}
self.default = None
self.current = None
self.max_tokens = max_tokens or self.MAX_TOKENS
self.log = logger.bind(src='client_manager')
self.names = self._init_clients(settings)
self.model_registry = ModelRegistry(__respath__ / "models.yaml")
获得不同模型和对应能力的列表，如支持图片理解，支持图片生成。
核心代码在 aipyapp/aipy/role.py 下
class RoleManager:
def __init__(self, roles_dir: str = None, api_conf: Dict[str, Dict[str, Any]] = None):
self.roles_dir = roles_dir
self.roles: Dict[str, Role] = {}
self.default_role: Role = None
self.current_role: Role = None
self.log = logger.bind(src='roles')
self.api_conf = api_conf
主要是存储 角色名称，简短描述，详细描述 这一个tuple类型的不同角色的列表。
核心代码在 aipyapp/aipy/mcp_tool.py
class MCPToolManager:
def __init__(self, config_path, tt_api_key):
self.config_path = config_path
self.config_reader = MCPConfigReader(config_path, tt_api_key=tt_api_key)
self.user_mcp = self.config_reader.get_user_mcp()
self.sys_mcp = self.config_reader.get_sys_mcp()
self.mcp_servers = self.sys_mcp | self.user_mcp # 优先系统mcp，其次再是内部的mcp
self._tools_dict = {} # 缓存已获取的工具列表
self._inited = False
# 全局启用/禁用用户MCP标志，默认禁用
self._user_mcp_enabled = False
self._sys_mcp_enabled = False
# 服务器状态缓存，记录每个服务器的启用/禁用状态
self._server_status = self._init_server_status()
# 创建全局的LazyMCPClient实例
self._lazy_client = LazyMCPClient(
mcp_servers=self.mcp_servers,
idle_ttl_seconds=300,
suppress_output=True, )
MCP工具有两种，get_user_mcp 读取 mcp.json 文件并返回 MCP 服务器清单，包括禁用的服务器， get_sys_mcp 获取内部 MCP 服务器配置，就是该项目内部配置的MCP服务
核心代码在 aipyapp/aipy/prompts.py
class Prompts:
def __init__(self, template_dir: str = None):
if not template_dir:
template_dir = __respath__ / 'prompts'
self.template_dir = os.path.abspath(template_dir)
self.env = Environment(
trim_blocks=True,
lstrip_blocks=True,
loader=FileSystemLoader(self.template_dir),
#autoescape=select_autoescape(['j2'])
)
self._init_env() # 调用 _init_env 方法注册全局变量
class Plugin(EventListener):
"""插件基类"""
name: str = None
version: str = None
description: str = None
author: str = None
def __init__(self):
self.logger = logger.bind(src=f"plugin.{self.name}")
def _get_methods(self, prefix: str = "fn_") -> Dict[str, Callable]:
"""Get all functions # 获得类中所有方法的函数
Returns:
All functions
"""
return {
name[len(prefix):]: method
for name, method in inspect.getmembers(self, predicate=inspect.ismethod)
if name.startswith(prefix) and len(name) > len(prefix) }
插件对象的基类是
class EventListener(Protocol):
def get_handlers(self) -> Dict[str, EventHandler]:
...
其中 Protocol 是 Python 的typing.Protocol 是为了支持结构子类型而引入的特性，为静态duck类型。 它允许您定义一组方法和属性，类必须实现这些方法和属性才能与协议兼容，而无需显式继承。这对于类型暗示和静态类型检查特别有用。
插件分析--图像识别和分析插件 （aipyapp/plugins/p_image_tool.py）
class ImageToolPlugin(TaskPlugin):
"""Image Tool - Provides image recognition and analysis capabilities."""
name = "image_tool"
version = "1.0.0"
description = "Image Tool - Provides image recognition and analysis capabilities."
author = "AiPy Team"
def init(self):
"""初始化OpenAI客户端"""
api_key = self.config.get('api_key')
base_url = self.config.get('base_url')
model = self.config.get('model', 'gpt-4-vision-preview')
if not api_key:
raise PluginInitError("API Key not configured for OpenAI.")
self.client = OpenAI(
api_key=api_key,
base_url=base_url
)
self.model = model
self.logger.info(f"Initialized OpenAI client with model: {model}")
该类继承自 TaskPlugin基类，主要使用Prompt + LLM去实现两个功能，分析图片和识别图片。
两个功能分为不同的Prompt来实现具体的功能，可以上传图片的URL或者本地的图片。
插件分析--捕获插件 (aipyapp/plugins/p_style_agent.py)
核心代码
class DisplayAgent(RichDisplayPlugin):
"""Agent模式显示插件 - 捕获所有输出用于API返回"""
name = "agent"
version = "1.0.0"
description = "Agent display style"
author = "AiPy Team"
def __init__(self, console, quiet: bool = True):
super().__init__(console, quiet)
# 捕获的输出数据
self.captured_data = {
'messages': [],
'results': [],
'errors': [],
'status': 'running',
'start_time': datetime.now().isoformat(),
'end_time': None,
'metadata': {}
}
调用LLM，任务开始结束，异常处理，工具调用，都会记录下来。
插件分析-网络工具插件 (aipyapp/plugins/p_web_tools.py)
class WebToolsPlugin(TaskPlugin):
"""Web Tools - Provides web page scraping, URL analysis, and HTTP request capabilities."""
name = "web_tools"
version = "1.0.0"
description = "Web Tools - Provides web page scraping, URL analysis, and HTTP request capabilities."
author = "AiPy Team"
def init(self):
"""Initialize network tool configuration."""
self.timeout = self.config.get('timeout', 30)
self.user_agent = self.config.get('user_agent', 'AiPy WebTools/1.0')
self.max_content_length = self.config.get('max_content_length', 1024 * 1024) # 1MB
self.headers = {
'User-Agent': self.user_agent,
**self.config.get('default_headers', {})
}
self.logger.info(f"Initialized network tool, timeout: {self.timeout}s")
有三个功能，获取网页， 实现http请求，对URL的分析。
任务管理的核心类。 核心代码为 aipyapp/aipy/agent_taskmgr.py 上，其父类为TaskManager类。
class AgentTaskManager(TaskManager):
"""Agent模式任务管理器"""
def __init__(self, settings, /, display_manager=None):
# 强制使用agent显示模式和headless设置
super().__init__(settings, display_manager=display_manager)
# Agent特有属性
self.agent_tasks: Dict[str, AgentTask] = {}
self.executor = ThreadPoolExecutor(max_workers=4) # 支持并发
self.log = logger.bind(src='agent_taskmgr')
主要负责管理Agent的任务运行，主要有几个接口，
提交任务， def submit_task(self, instruction, metadata)，关键属性是instruction，依靠该属性建立一个AgentTask。
执行特定的任务，def execute_task(self, task_id)，从已有的任务中选择该任务，并将其投入到异步程序队列中
loop = asyncio.get_event_loop() # 获取当前异步程序中的事件循环对象
await loop.run_in_executor(
self.executor,
self._run_task_sync,
agent_task
)
其中上面真正异步执行任务的函数为
def _run_task_sync(self, agent_task: AgentTask):
"""同步执行任务（在线程池中运行）"""
try:
# 执行任务
agent_task.task.run(agent_task.instruction)
# 确保任务完成
agent_task.task.done()
class Stoppable():
def __init__(self):
self._stop_event = threading.Event()
使用了线程退出控制的基础实现，创建了一个线程安全的停止信号，用于优雅控制子线程的启动和停止，比卖你强制终止线程导致的资源泄露，数据异常等问题。
threading.Event()：Python 内置的线程同步原语，本质是一个线程安全的布尔标志位，默认状态为「未设置（False）」，提供set()（设为 True，触发信号）、is_set()（判断状态）、wait()（阻塞等待信号触发）三个核心方法，实现多线程间的通信。
Task类的核心代码为
class Task(Stoppable):
def __init__(self, manager: TaskManager, data: TaskData | None = None, parent: Task | None = None):
super().__init__()
data = data or TaskData()
# Phase 1: Initialize basic attributes (no dependencies)
self.data = data
self.parent = parent
self.task_id = data.id
self.manager = manager
self.settings = manager.settings
self.log = logger.bind(src='task', id=self.task_id)
self.cwd = manager.cwd / self.task_id if not parent else parent.cwd / self.task_id
self.gui = manager.settings.gui
self._saved = False
self.max_rounds = manager.settings.get('max_rounds', MAX_ROUNDS)
self.role = manager.role_manager.current_role
# Phase 2: Initialize data objects (minimal dependencies)
if parent:
data.context = parent.context.model_copy(deep=True)
data.message_storage = parent.message_storage.model_copy(deep=True)
#data.blocks = parent.blocks.model_copy(deep=True)
self.blocks = data.blocks
self.message_storage = data.message_storage
self.context = data.context
self.events = data.events
# Phase 3: Initialize managers and processors (depend on Phase 2)
self.event_bus = TypedEventBus() if not parent else parent.event_bus
self.context_manager = ContextManager(
self.message_storage,
self.context,
manager.settings.get('context_manager')
)
self.tool_call_processor = ToolCallProcessor() if not parent else parent.tool_call_processor
# Phase 4: Initialize display (depends on event_bus)
if not parent:
if manager.display_manager:
self.display = manager.display_manager.create_display_plugin()
self.event_bus.add_listener(self.display)
else:
self.display = None
else:
self.display = parent.display
# Phase 5: Initialize execution components (depend on task)
self.mcp = manager.mcp
self.prompts = manager.prompts
self.client_manager = manager.client_manager
self.runtime = CliPythonRuntime(self)
self.runner = BlockExecutor()
self.runner.set_python_runtime(self.runtime)
self.client = Client(self)
# Phase 6: Initialize cleaners (depend on context_manager)
self.step_cleaner = SimpleStepCleaner(self.context_manager)
# Phase 7: Initialize plugins (depend on runtime and event_bus)
if not parent:
self._initialize_plugins(manager)
else:
self.plugins = parent.plugins
# Phase 8: Initialize steps last (depend on almost everything)
self.steps: List[Step] = [Step(self, step_data) for step_data in data.steps]
初始化Task比较复杂，涉及到很多部件
其中最重要的Task类函数，是在被AgentTaskManager调用的run接口。 def run(self, instruction, title)
对于run函数来说，首先根据instruction来处理多模态内容并验证模型能力。
核心代码的是在 aipyapp/aipy/step.py
class Step:
def __init__(self, task: Task, data: StepData):
self.task = task
self.log = logger.bind(src='Step')
self._data = data
self._summary = Counter()
def request(self, user_message: ChatMessage) -> Response:
client = self.task.client
self.task.emit('request_started', llm=client.name)
msg = client(user_message)
self.task.emit('response_completed', llm=client.name, msg=msg)
if isinstance(msg.message, ErrorMessage):
response = Response(message=msg)
self.log.error('LLM request error', error=msg.content)
else:
self._summary.update(msg.usage)
response = Response.from_message(msg, parse_mcp=self.task.mcp)
return response
该类是拆解每一个任务为一个接着一个的步骤，每个具体的步骤会根据相关的LLM来实现输入和输出。
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。