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

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 

21from typing import Optional 

22 

23from agentos.agent.agent_builder import build_agent 

24from agentos.llm.base import LLMProvider 

25 

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

27 

28 

29@dataclass 

30class AgentResult: 

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

32 success: bool 

33 output: str = "" 

34 error: Optional[str] = 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: Optional[LLMProvider] = None, 

54 max_steps: int = 20, 

55 system_prompt: Optional[str] = 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( 

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

78 ) 

79 

80 logger.info( 

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

82 f"{tool_calls} tool calls, " 

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

84 ) 

85 

86 return AgentResult( 

87 success=raw.success, 

88 output=raw.final_answer or "", 

89 error=raw.error, 

90 total_steps=raw.total_steps, 

91 total_tokens=raw.total_tokens, 

92 total_cost_usd=raw.total_cost_usd, 

93 total_latency_ms=raw.total_duration_ms, 

94 tool_calls=tool_calls, 

95 ) 

96 

97 except Exception as e: 

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

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

100 

101 def get_tool_count(self) -> int: 

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

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

104 

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

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

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