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

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 

20import time 

21from dataclasses import dataclass, field 

22from typing import Any, Optional 

23 

24from agentos.agent.agent_builder import build_agent 

25from agentos.llm.base import LLMProvider 

26 

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

28 

29 

30@dataclass 

31class AgentResult: 

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

33 success: bool 

34 output: str = "" 

35 error: Optional[str] = None 

36 total_steps: int = 0 

37 total_tokens: int = 0 

38 total_cost_usd: float = 0.0 

39 total_latency_ms: float = 0.0 

40 tool_calls: int = 0 

41 

42 

43class ProductionAgent: 

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

45 

46 Example: 

47 agent = ProductionAgent() 

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

49 print(result.output) 

50 """ 

51 

52 def __init__( 

53 self, 

54 provider: Optional[LLMProvider] = None, 

55 max_steps: int = 20, 

56 system_prompt: Optional[str] = None, 

57 include_skills: bool = True, 

58 verbose: bool = False, 

59 ): 

60 self.verbose = verbose 

61 

62 self._agent = build_agent( 

63 provider=provider, 

64 max_steps=max_steps, 

65 system_prompt=system_prompt, 

66 include_skills=include_skills, 

67 verbose=verbose, 

68 ) 

69 

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

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

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

73 

74 try: 

75 raw = self._agent.run(task) 

76 

77 tool_calls = sum( 

78 len(s.tool_calls) for s in raw.steps 

79 ) 

80 

81 logger.info( 

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

83 f"{tool_calls} tool calls, " 

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

85 ) 

86 

87 return AgentResult( 

88 success=raw.success, 

89 output=raw.final_answer or "", 

90 error=raw.error, 

91 total_steps=raw.total_steps, 

92 total_tokens=raw.total_tokens, 

93 total_cost_usd=raw.total_cost_usd, 

94 total_latency_ms=raw.total_duration_ms, 

95 tool_calls=tool_calls, 

96 ) 

97 

98 except Exception as e: 

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

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

101 

102 def get_tool_count(self) -> int: 

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

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

105 

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

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

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