en2zmax / ams-agent

Разобрался. Причина — в семантике @agent.tool в pydantic-ai, и в том, что AMS даёт в step один и тот же объект агента на каждой итерации. Почему падает @agent.tool → FunctionToolset.add_function() → add_tool(), а там жёсткая проверка: def add_tool(self, tool): if tool.name in self.tools: raise UserError(f'Tool name conflicts with existing tool: {tool.name!r}') ... self.tools[tool.name] = tool step вызывается много раз, но agent переиспользуется (именно поэтому agent.history и set_system_prompt имеют смысл — история копится в одном объекте). Значит на второй итерации регистрация read_file_lines с тем же именем бросает UserError: Tool name conflicts with existing tool: 'read_file_lines' (или молча затирается — в зависимости от версии, в 2.33 именно исключение). Плюс AMS-обёртка StatefulAgent (это Cython-класс в /opt/ams/app/processors/tools/stateful_agent.so) проксирует .tool на внутренний pydantic_ai.Agent через __getattr__ — так что «re-register» бьёт прямо по тулсету внутреннего агента. Как правильно (рекомендую вариант 1) Вариант 1 — вообще не трогать агент, передавать тулсет на каждый run Тулсет создаётся один раз при импорте модуля, а в ран передаётся через toolsets=. Ничего не накапливается на агенте → конфликта нет по построению: import os from datetime import datetime from pydantic_ai import ModelSettings, RunContext, ModelRetry, FunctionToolset from pydantic_ai.workspaces import LocalWorkspaceBackend from pydantic_ai_harness.filesystem import FileSystem CURRENT_DIR = os.path.dirname(os.path.abspath(__file__)) system_prompt = open(os.path.join(CURRENT_DIR, 'system_prompt.txt')).read() # инструмент объявлен на уровне модуля (один раз) async def read_file_lines(ctx: RunContext, path: str, start_line: int = 1, max_lines: int = 50) -> str: """Read a specific range of lines from a text file...""" if start_line < 1: raise ModelRetry('start_line must be 1 or greater.') try: content = await ctx.workspace.read_text(path) except FileNotFoundError: raise ModelRetry(f"File '{path}' does not exist.") lines = content.splitlines() start_idx = start_line - 1 end_idx = start_idx + max_lines chunk = lines[start_idx:end_idx] if not chunk: return f"EOF: No lines found in range {start_line}-{end_idx}." return f"[Lines {start_line}-{min(end_idx, len(lines))} of {len(lines)}]:\n" + "\n".join(chunk) # создаётся ОДИН раз TOOLSET = FunctionToolset(tools=[read_file_lines]) WORKSPACE = LocalWorkspaceBackend('./logs') def step(agent, state, params): dt = datetime.now().strftime("%d.%m.%Y %H:%M") agent.set_system_prompt(system_prompt) model_settings = ModelSettings(**params['sampling_params']) prompt = params['prompt'] # ... ваш блок с params['file'] ... if not prompt: return with agent.run_stream_sync( prompt, model_settings=model_settings, toolsets=[TOOLSET], # <-- тул приходит на этот ран, не мутируя агента capabilities=[FileSystem()], workspace=WORKSPACE, ) as stream: for chunk in stream.stream_text(delta=True): yield {'type': 'chunk', 'data': chunk} Вариант 2 — оставить @agent.tool, но регистрировать идемпотентно Если хочется именно закрепить тул на агенте — вынесите функцию из step и защититесь флагом на объекте агента (не глобальным bool — в процессе может жить несколько агентов): def _register_tools_once(agent): if getattr(agent, "_file_tools_registered", False): return agent.tool(read_file_lines) agent._file_tools_registered = True и в начале step: _register_tools_once(agent). Отличие: вариант 1 не зависит от того, переиспользуется ли объект агента/обёртки; вариант 2 надёжен, только пока AMS передаёт тот же самый объект agent. Поэтому вариант 1 безопаснее. Ещё два момента - capabilities=[FileSystem()] уже даёт файловые тулы (read_file, edit_file, list_files, grep, с параметрами max_read_lines/max_read_chars и т.п.). Так что ваш read_file_lines может быть избыточен — если нужен именно построчный диапазон + ModelRetry, оставляйте, но дублирующиеся имена (read_file) не создавайте. - Проверьте окружение на сервере. На 192.168.102.120 в вене AMS стоит pydantic_ai 2.33.0, а в нём нет pydantic_ai.workspaces, не установлен pydantic_ai_harness, и параметра workspace= у run_stream_sync нет (есть deps, toolsets, capabilities). Актуальные версии — pydantic-ai 2.56.0 / pydantic-ai-harness 0.56.0. То есть этот handler.py на текущем серверном AMS не заимпортится — нужен либо апгрейд/установка harness, либо адаптация под 2.33 (capabilities=[FileSystem()] там есть, а workspace-механику придётся заменить). То же самое с сигнатурой: в хелпе AMS (/opt/ams/app/help/agents/index.html) контракт — step(params, state, agent), но «имена аргументов являются частью контракта», поэтому ваш порядок (agent, state, params) сработает, если AMS вызывает по ключевым словам.