1
class MyLLM(HelloAgentsLLM):
2
    def __init__(
3
        self,
4
        model: Optional[str] = None,
5
        api_key: Optional[str] = None,
6
        base_url: Optional[str] = None,
7
        provider: Optional[str] = "auto",
8
        **kwargs
9
    ):
10
        # 检查provider是否为我们想处理的'modelscope'
11
        if provider == "modelscope":
12
            print("正在使用自定义的 ModelScope Provider")
13
            # 解析 ModelScope 的凭证
14
            # 验证凭证是否存在
15
            # 设置默认模型和其他参数
16
            # 使用获取的参数创建OpenAI客户端实例
17
        else:
18
            # 如果不是 modelscope, 则完全使用父类的原始逻辑来处理
19
            super().__init__(model=model, api_key=api_key, base_url=base_url, provider=provider, **kwargs)

1
"""Agent基类"""
2
from abc import ABC, abstractmethod
3
from typing import Optional, Any
4
from .message import Message
5
from .llm import HelloAgentsLLM
6
from .config import Config
7

8
class Agent(ABC):
9
    """Agent基类"""
10

11
    def __init__(
12
        self,
13
        name: str,
14
        llm: HelloAgentsLLM,
15
        system_prompt: Optional[str] = None,
16
        config: Optional[Config] = None
17
    ):
18
        self.name = name
19
        self.llm = llm
20
        self.system_prompt = system_prompt
21
        self.config = config or Config()
22
        self._history: list[Message] = []
23

24
    @abstractmethod
25
    def run(self, input_text: str, **kwargs) -> str:
26
        """运行Agent"""
27
        pass
28

29
    def add_message(self, message: Message):
30
        """添加消息到历史记录"""
31
        self._history.append(message)
32

33
    def clear_history(self):
34
        """清空历史记录"""
35
        self._history.clear()
36

37
    def get_history(self) -> list[Message]:
38
        """获取历史记录"""
39
        return self._history.copy()
40

41
    def __str__(self) -> str:
42
        return f"Agent(name={self.name}, provider={self.llm.provider})"

1
MY_REACT_PROMPT = """你是一个具备推理和行动能力的AI助手。你可以通过思考分析问题，然后调用合适的工具来获取信息，最终给出准确的答案。
2

3
## 可用工具
4
{tools}
5

6
## 工作流程
7
请严格按照以下格式进行回应，每次只能执行一个步骤:
8

9
Thought: 分析当前问题，思考需要什么信息或采取什么行动。
10
Action: 选择一个行动，格式必须是以下之一:
11
- `{{tool_name}}[{{tool_input}}]` - 调用指定工具
12
- `Finish[最终答案]` - 当你有足够信息给出最终答案时
13

14
## 重要提醒
15
1. 每次回应必须包含Thought和Action两部分
16
2. 工具调用的格式必须严格遵循:工具名[参数]
17
3. 只有当你确信有足够信息回答问题时，才使用Finish
18
4. 如果工具返回的信息不够，继续使用其他工具或相同工具的不同参数
19

20
## 当前任务
21
**Question:** {question}
22

23
## 执行历史
24
{history}
25

26
现在开始你的推理和行动:
27
"""

1
def run(self, input_text: str, **kwargs) -> str:
2
    """运行ReAct Agent"""
3
    self.current_history = []
4
    current_step = 0
5

6
    print(f"\n🤖 {self.name} 开始处理问题: {input_text}")
7

8
    while current_step < self.max_steps:
9
        current_step += 1
10
        print(f"\n--- 第 {current_step} 步 ---")
11

12
        # 1. 构建提示词
13
        tools_desc = self.tool_registry.get_tools_description()
14
        history_str = "\n".join(self.current_history)
15
        prompt = self.prompt_template.format(
16
            tools=tools_desc,
17
            question=input_text,
18
            history=history_str
19
        )
20

21
        # 2. 调用LLM
22
        messages = [{"role": "user", "content": prompt}]
23
        response_text = self.llm.invoke(messages, **kwargs)
24

25
        # 3. 解析输出
26
        thought, action = self._parse_output(response_text)
27

28
        # 4. 检查完成条件
29
        if action and action.startswith("Finish"):
30
            final_answer = self._parse_action_input(action)
31
            self.add_message(Message(input_text, "user"))
32
            self.add_message(Message(final_answer, "assistant"))
33
            return final_answer
34

35
        # 5. 执行工具调用
36
        if action:
37
            tool_name, tool_input = self._parse_action(action)
38
            observation = self.tool_registry.execute_tool(tool_name, tool_input)
39
            self.current_history.append(f"Action: {action}")
40
            self.current_history.append(f"Observation: {observation}")
41

42
    # 达到最大步数
43
    final_answer = "抱歉，我无法在限定步数内完成这个任务。"
44
    self.add_message(Message(input_text, "user"))
45
    self.add_message(Message(final_answer, "assistant"))
46
    return final_answer

1
class Agent:
2

3
def __init__(self):
4

5
self.calculator = MyCalculator() # 计算器写死在代码里
6

7
self.searcher = TavilySearch() # 搜索引擎也写死在代码里
8

9
def run(self, user_input):
10

11
if "计算" in user_input:
12

13
expr = parse_expr(user_input)
14

15
return self.calculator.run(expr)
16

17
elif "搜索" in user_input:
18

19
query = parse_query(user_input)
20

21
return self.searcher.search(query)