Coverage for agentos/agent/production_agent.py: 0%

1""" 

2ProductionAgent — 生产级一行调用接口。 

3 

4将 LLM Provider → ToolExecutor → Bridge → BaseTool/Skill 全链路封装， 

5提供结构化日志、结果统计，开箱即用。 

6 

7用法: 

8 from agentos.agent.production_agent import ProductionAgent 

9 

10 agent = ProductionAgent() 

11 result = agent.run("分析我的 CSV 文件 sales.csv") 

12 

13 print(result.output) 

14 print(f"{result.total_steps} steps, {result.total_latency_ms:.0f}ms") 

15""" 

16 

17from __future__ import annotations 

18 

19import logging 

20from dataclasses import dataclass 

21 

22from agentos.agent.agent_builder import build_agent 

23from agentos.llm.base import LLMProvider 

24 

25logger = logging.getLogger("agentos.production") 

26 

27 

28@dataclass 

29class AgentResult: 

30 """Agent 执行完整结果。""" 

31 

32 success: bool 

33 output: str = "" 

34 error: str | None = None 

35 total_steps: int = 0 

36 total_tokens: int = 0 

37 total_cost_usd: float = 0.0 

38 total_latency_ms: float = 0.0 

39 tool_calls: int = 0 

40 

41 

42class ProductionAgent: 

43 """生产级 Agent — 一行 run() 搞定一切。 

44 

45 Example: 

46 agent = ProductionAgent() 

47 result = agent.run("帮我计算 hello world 的 SHA256") 

48 print(result.output) 

49 """ 

50 

51 def __init__( 

52 self, 

53 provider: LLMProvider | None = None, 

54 max_steps: int = 20, 

55 system_prompt: str | None = None, 

56 include_skills: bool = True, 

57 verbose: bool = False, 

58 ): 

59 self.verbose = verbose 

60 

61 self._agent = build_agent( 

62 provider=provider, 

63 max_steps=max_steps, 

64 system_prompt=system_prompt, 

65 include_skills=include_skills, 

66 verbose=verbose, 

67 ) 

68 

69 def run(self, task: str) -> AgentResult: 

70 """执行任务并返回结构化结果。""" 

71 logger.info(f"Task: {task[:120]}") 

72 

73 try: 

74 raw = self._agent.run(task) 

75 

76 tool_calls = sum(len(s.tool_calls) for s in raw.steps) 

77 

78 logger.info( 

79 f"Done: {raw.total_steps} steps, " 

80 f"{tool_calls} tool calls, " 

81 f"{raw.total_duration_ms:.0f}ms" 

82 ) 

83 

84 return AgentResult( 

85 success=raw.success, 

86 output=raw.final_answer or "", 

87 error=raw.error, 

88 total_steps=raw.total_steps, 

89 total_tokens=raw.total_tokens, 

90 total_cost_usd=raw.total_cost_usd, 

91 total_latency_ms=raw.total_duration_ms, 

92 tool_calls=tool_calls, 

93 ) 

94 

95 except Exception as e: 

96 logger.error(f"Failed: {e}") 

97 return AgentResult(success=False, error=str(e)) 

98 

99 def get_tool_count(self) -> int: 

100 """返回已注册工具数量。""" 

101 return len(self._agent._executor._tools) 

102 

103 def list_tools(self) -> list[str]: 

104 """返回已注册工具名称列表。""" 

105 return sorted(self._agent._executor._tools.keys())