LLM Agent开发指南

1
用户输入
2
   ↓
3
[感知 Perception] ← 记忆
4
   ↓
5
[规划 Planning]  → 分解任务
6
   ↓
7
[决策 Decision]  → 选择行动
8
   ↓
9
[执行 Action]    → 调用工具
10
   ↓
11
[观察 Observation] → 获取结果
12
   ↓
13
(循环直到完成)
14
   ↓
15
返回最终答案

1
User: 埃隆·马斯克的年龄的平方是多少？
2

3
Thought 1: 我需要先找到埃隆·马斯克的出生日期
4
Action 1: Search[埃隆·马斯克 出生日期]
5
Observation 1: 埃隆·马斯克出生于 1971 年 6 月 28 日
6

7
Thought 2: 现在我需要计算他的年龄
8
Action 2: Calculate[2024 - 1971]
9
Observation 2: 53
10

11
Thought 3: 现在计算 53 的平方
12
Action 3: Calculate[53 ** 2]
13
Observation 3: 2809
14

15
Thought 4: 我现在知道答案了
16
Final Answer: 埃隆·马斯克 53 岁，年龄的平方是 2809。

1
import re
2
from typing import Dict, Callable, List, Tuple
3
import openai
4

5
class ReActAgent:
6
    def __init__(self, tools: Dict[str, Callable], model="gpt-4", max_steps=10):
7
        """
8
        Args:
9
            tools: 工具字典 {tool_name: tool_function}
10
            model: 使用的 LLM 模型
11
            max_steps: 最大推理步数
12
        """
13
        self.tools = tools
14
        self.model = model
15
        self.max_steps = max_steps
16

17
    def run(self, question: str) -> str:
18
        """执行 Agent"""
19
        prompt = self._build_initial_prompt(question)
20

21
        for step in range(self.max_steps):
22
            # 调用 LLM 生成下一步
23
            response = self._call_llm(prompt)
24

25
            # 解析 response
26
            thought, action, action_input = self._parse_response(response)
27

28
            print(f"\n--- Step {step + 1} ---")
29
            print(f"Thought: {thought}")
30
            print(f"Action: {action}[{action_input}]")
31

32
            # 检查是否得出最终答案
33
            if action.lower() == "final answer":
34
                return action_input
35

36
            # 执行 Action
37
            observation = self._execute_action(action, action_input)
38
            print(f"Observation: {observation}")
39

40
            # 更新 prompt
41
            prompt += f"\nObservation {step + 1}: {observation}\nThought {step + 2}:"
42

43
        return "达到最大步数限制"
44

45
    def _build_initial_prompt(self, question: str) -> str:
46
        """构建初始 prompt"""
47
        tools_desc = "\n".join([
48
            f"- {name}: {func.__doc__}"
49
            for name, func in self.tools.items()
50
        ])
51

52
        return f"""Answer the following question by reasoning step by step.
53

54
Available tools:
55
{tools_desc}
56

57
Use this format:
58
Thought: [your reasoning]
59
Action: [tool_name[input]]
60
Observation: [tool result will be provided]
61
... (repeat Thought/Action/Observation)
62
Thought: I now know the final answer
63
Action: Final Answer[your answer]
64

65
Question: {question}
66

67
Thought 1:"""
68

69
    def _call_llm(self, prompt: str) -> str:
70
        """调用 LLM"""
71
        response = openai.ChatCompletion.create(
72
            model=self.model,
73
            messages=[{"role": "user", "content": prompt}],
74
            temperature=0,
75
            max_tokens=200
76
        )
77
        return response.choices[0].message.content
78

79
    def _parse_response(self, response: str) -> Tuple[str, str, str]:
80
        """解析 LLM 输出"""
81
        # 提取 Thought
82
        thought_match = re.search(r'Thought:?(.*?)(?=Action:|$)', response, re.DOTALL)
83
        thought = thought_match.group(1).strip() if thought_match else ""
84

85
        # 提取 Action 和 Input
86
        action_match = re.search(r'Action:?\s*(\w+)\[(.*?)\]', response)
87
        if action_match:
88
            action = action_match.group(1)
89
            action_input = action_match.group(2).strip()
90
        else:
91
            action = "Final Answer"
92
            action_input = thought
93

94
        return thought, action, action_input
95

96
    def _execute_action(self, action: str, action_input: str) -> str:
97
        """执行工具调用"""
98
        if action not in self.tools:
99
            return f"Error: Tool '{action}' not found"
100

101
        try:
102
            result = self.tools[action](action_input)
103
            return str(result)
104
        except Exception as e:
105
            return f"Error executing {action}: {str(e)}"
106

107
# === 定义工具 ===
108
def search(query: str) -> str:
109
    """Search the web for information"""
110
    # 实际实现中调用搜索 API（如 Google Search API）
111
    # 这里用模拟数据
112
    mock_results = {
113
        "埃隆·马斯克 出生日期": "埃隆·马斯克出生于 1971 年 6 月 28 日",
114
        "OpenAI CEO": "OpenAI 的 CEO 是 Sam Altman",
115
    }
116
    return mock_results.get(query, "No results found")
117

118
def calculate(expression: str) -> str:
119
    """Evaluate a mathematical expression"""
120
    try:
121
        # 安全的计算（只允许数学运算）
122
        allowed_names = {
123
            'abs': abs, 'round': round, 'min': min, 'max': max,
124
            'pow': pow, 'sum': sum
125
        }
126
        result = eval(expression, {"__builtins__": {}}, allowed_names)
127
        return str(result)
128
    except Exception as e:
129
        return f"Calculation error: {e}"
130

131
def lookup(keyword: str) -> str:
132
    """Look up information in a knowledge base"""
133
    # 实际实现中查询向量数据库或知识图谱
134
    knowledge_base = {
135
        "Python": "Python 是一种高级编程语言，由 Guido van Rossum 创建于 1991 年",
136
        "AI": "人工智能（AI）是计算机科学的一个分支，致力于创建智能机器"
137
    }
138
    return knowledge_base.get(keyword, "Not found in knowledge base")
139

140
# === 使用 Agent ===
141
agent = ReActAgent(
142
    tools={
143
        "Search": search,
144
        "Calculate": calculate,
145
        "Lookup": lookup
146
    }
147
)
148

149
answer = agent.run("埃隆·马斯克的年龄的平方是多少？")
150
print(f"\n=== Final Answer ===\n{answer}")

1
import openai
2
import json
3

4
# 定义工具
5
tools = [
6
    {
7
        "type": "function",
8
        "function": {
9
            "name": "get_current_weather",
10
            "description": "Get the current weather in a given location",
11
            "parameters": {
12
                "type": "object",
13
                "properties": {
14
                    "location": {
15
                        "type": "string",
16
                        "description": "The city and state, e.g. San Francisco, CA"
17
                    },
18
                    "unit": {
19
                        "type": "string",
20
                        "enum": ["celsius", "fahrenheit"]
21
                    }
22
                },
23
                "required": ["location"]
24
            }
25
        }
26
    }
27
]
28

29
# 调用
30
response = openai.ChatCompletion.create(
31
    model="gpt-4",
32
    messages=[{"role": "user", "content": "What's the weather in Boston?"}],
33
    tools=tools,
34
    tool_choice="auto"  # 自动决定是否调用工具
35
)
36

37
# 检查是否需要调用工具
38
message = response.choices[0].message
39

40
if message.tool_calls:
41
    tool_call = message.tool_calls[0]
42
    function_name = tool_call.function.name
43
    function_args = json.loads(tool_call.function.arguments)
44

45
    print(f"调用工具: {function_name}")
46
    print(f"参数: {function_args}")
47

48
    # 执行工具调用
49
    if function_name == "get_current_weather":
50
        weather = get_current_weather(**function_args)
51

52
        # 将结果返回给模型
53
        second_response = openai.ChatCompletion.create(
54
            model="gpt-4",
55
            messages=[
56
                {"role": "user", "content": "What's the weather in Boston?"},
57
                message,
58
                {
59
                    "role": "tool",
60
                    "tool_call_id": tool_call.id,
61
                    "name": function_name,
62
                    "content": json.dumps(weather)
63
                }
64
            ]
65
        )
66

67
        final_answer = second_response.choices[0].message.content
68
        print(f"最终回答: {final_answer}")

1
from typing import List, Dict, Any, Callable
2
from pydantic import BaseModel, Field
3

4
class Tool(BaseModel):
5
    """工具基类"""
6
    name: str
7
    description: str
8
    parameters: Dict[str, Any]
9
    function: Callable
10

11
    def run(self, **kwargs) -> str:
12
        """执行工具"""
13
        return self.function(**kwargs)
14

15
    def to_openai_format(self) -> Dict:
16
        """转换为 OpenAI function calling 格式"""
17
        return {
18
            "type": "function",
19
            "function": {
20
                "name": self.name,
21
                "description": self.description,
22
                "parameters": self.parameters
23
            }
24
        }
25

26
class ToolRegistry:
27
    """工具注册表"""
28
    def __init__(self):
29
        self.tools: Dict[str, Tool] = {}
30

31
    def register(self, tool: Tool):
32
        """注册工具"""
33
        self.tools[tool.name] = tool
34

35
    def get(self, name: str) -> Tool:
36
        """获取工具"""
37
        return self.tools.get(name)
38

39
    def list_tools(self) -> List[Dict]:
40
        """列出所有工具（OpenAI 格式）"""
41
        return [tool.to_openai_format() for tool in self.tools.values()]
42

43
# 创建工具
44
def calculator(expression: str) -> str:
45
    """Evaluate a math expression"""
46
    return str(eval(expression))
47

48
calculator_tool = Tool(
49
    name="calculator",
50
    description="Evaluate mathematical expressions",
51
    parameters={
52
        "type": "object",
53
        "properties": {
54
            "expression": {
55
                "type": "string",
56
                "description": "Math expression to evaluate, e.g. '2 + 2'"
57
            }
58
        },
59
        "required": ["expression"]
60
    },
61
    function=calculator
62
)
63

64
# 注册工具
65
registry = ToolRegistry()
66
registry.register(calculator_tool)

1
from collections import deque
2

3
class ConversationMemory:
4
    def __init__(self, max_turns=10):
5
        self.messages = deque(maxlen=max_turns * 2)  # 每轮2条消息（用户+助手）
6

7
    def add_user_message(self, content: str):
8
        self.messages.append({"role": "user", "content": content})
9

10
    def add_assistant_message(self, content: str):
11
        self.messages.append({"role": "assistant", "content": content})
12

13
    def get_messages(self) -> List[Dict]:
14
        return list(self.messages)
15

16
    def clear(self):
17
        self.messages.clear()
18

19
# 使用
20
memory = ConversationMemory(max_turns=5)
21

22
memory.add_user_message("你好，我叫张三")
23
memory.add_assistant_message("你好张三，很高兴认识你！")
24

25
memory.add_user_message("我的名字是什么？")
26
# Agent 可以从 memory 中回忆起"张三"

1
from langchain.memory import VectorStoreRetrieverMemory
2
from langchain.embeddings import OpenAIEmbeddings
3
from langchain.vectorstores import Chroma
4

5
# 创建向量存储
6
embeddings = OpenAIEmbeddings()
7
vectorstore = Chroma(embedding_function=embeddings)
8

9
# 创建记忆
10
memory = VectorStoreRetrieverMemory(
11
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
12
    memory_key="history",
13
    input_key="input"
14
)
15

16
# 存储记忆
17
memory.save_context(
18
    {"input": "我最喜欢的颜色是蓝色"},
19
    {"output": "好的，我记住了你喜欢蓝色"}
20
)
21

22
# 检索记忆（根据相关性）
23
relevant_memories = memory.load_memory_variables(
24
    {"input": "我喜欢什么颜色？"}
25
)
26
print(relevant_memories["history"])
27
# 输出：你最喜欢的颜色是蓝色

1
from langchain.memory import ConversationEntityMemory
2

3
class EntityMemory:
4
    def __init__(self):
5
        self.entities = {}  # {entity_name: {properties}}
6

7
    def update_entity(self, name: str, properties: Dict):
8
        """更新实体信息"""
9
        if name not in self.entities:
10
            self.entities[name] = {}
11
        self.entities[name].update(properties)
12

13
    def get_entity(self, name: str) -> Dict:
14
        """获取实体信息"""
15
        return self.entities.get(name, {})
16

17
    def extract_and_update(self, text: str):
18
        """从文本中提取实体并更新"""
19
        # 使用 NER 或 LLM 提取实体
20
        # 示例：
21
        entities = extract_entities_from_text(text)  # 需要实现
22
        for entity in entities:
23
            self.update_entity(entity["name"], entity["properties"])
24

25
# 使用
26
memory = EntityMemory()
27
memory.update_entity("张三", {"职业": "工程师", "年龄": 30})
28
memory.update_entity("北京", {"类型": "城市", "国家": "中国"})
29

30
print(memory.get_entity("张三"))
31
# {'职业': '工程师', '年龄': 30}

1
from typing import List
2

3
class PlanExecuteAgent:
4
    def __init__(self, planner_llm, executor_agent):
5
        self.planner = planner_llm
6
        self.executor = executor_agent
7

8
    def run(self, objective: str) -> str:
9
        # 步骤 1: 制定计划
10
        plan = self.create_plan(objective)
11
        print(f"=== Plan ===\n{plan}\n")
12

13
        # 步骤 2: 执行计划
14
        results = []
15
        for step in plan:
16
            print(f"执行: {step}")
17
            result = self.executor.run(step)
18
            results.append(result)
19
            print(f"结果: {result}\n")
20

21
        # 步骤 3: 汇总结果
22
        final_answer = self.summarize_results(objective, results)
23
        return final_answer
24

25
    def create_plan(self, objective: str) -> List[str]:
26
        """使用 LLM 制定计划"""
27
        prompt = f"""You are a planning expert. Break down the following objective into concrete, actionable steps.
28

29
Objective: {objective}
30

31
Steps (number each step):"""
32

33
        response = self.planner(prompt)
34

35
        # 解析步骤
36
        steps = []
37
        for line in response.split('\n'):
38
            if line.strip() and line[0].isdigit():
39
                steps.append(line.split('.', 1)[1].strip())
40

41
        return steps
42

43
    def summarize_results(self, objective: str, results: List[str]) -> str:
44
        """汇总执行结果"""
45
        results_text = "\n".join([f"{i+1}. {r}" for i, r in enumerate(results)])
46

47
        prompt = f"""Based on the following results, provide a final answer to the objective.
48

49
Objective: {objective}
50

51
Results:
52
{results_text}
53

54
Final Answer:"""
55

56
        return self.planner(prompt)
57

58
# 使用
59
planner_llm = lambda prompt: openai.ChatCompletion.create(
60
    model="gpt-4",
61
    messages=[{"role": "user", "content": prompt}],
62
    temperature=0
63
).choices[0].message.content
64

65
executor = ReActAgent(tools={...})
66

67
agent = PlanExecuteAgent(planner_llm, executor)
68
answer = agent.run("研究 GPT-4 的技术细节并写一份总结报告")

1
class HierarchicalAgent:
2
    """分层 Agent：高层规划 + 低层执行"""
3

4
    def __init__(self, high_level_llm, low_level_agents: List):
5
        self.high_level = high_level_llm
6
        self.low_level = low_level_agents
7

8
    def run(self, goal: str) -> str:
9
        # 高层分解任务
10
        subtasks = self.decompose_task(goal)
11

12
        # 分配给低层 Agent
13
        results = []
14
        for subtask in subtasks:
15
            agent = self.select_agent(subtask)
16
            result = agent.run(subtask)
17
            results.append(result)
18

19
        # 整合结果
20
        return self.integrate_results(results)
21

22
    def decompose_task(self, goal: str) -> List[str]:
23
        """任务分解"""
24
        prompt = f"""Decompose the following goal into 3-5 subtasks:
25

26
Goal: {goal}
27

28
Subtasks:"""
29
        response = self.high_level(prompt)
30
        return parse_subtasks(response)
31

32
    def select_agent(self, subtask: str):
33
        """选择合适的 Agent"""
34
        # 可以基于关键词、相似度等
35
        for agent in self.low_level:
36
            if agent.can_handle(subtask):
37
                return agent
38
        return self.low_level[0]  # 默认

1
from langchain.agents import create_react_agent, AgentExecutor
2
from langchain.tools import Tool
3
from langchain_openai import ChatOpenAI
4
from langchain import hub
5

6
# 1. 定义工具
7
def search_web(query: str) -> str:
8
    """Search the web"""
9
    return f"Search results for: {query}"
10

11
def calculator(expression: str) -> str:
12
    """Calculate math expression"""
13
    return str(eval(expression))
14

15
tools = [
16
    Tool(
17
        name="Search",
18
        func=search_web,
19
        description="Useful for searching information"
20
    ),
21
    Tool(
22
        name="Calculator",
23
        func=calculator,
24
        description="Useful for math calculations"
25
    )
26
]
27

28
# 2. 创建 LLM
29
llm = ChatOpenAI(model="gpt-4", temperature=0)
30

31
# 3. 获取 prompt 模板
32
prompt = hub.pull("hwchase17/react")
33

34
# 4. 创建 Agent
35
agent = create_react_agent(llm, tools, prompt)
36

37
# 5. 创建 Agent Executor
38
agent_executor = AgentExecutor(
39
    agent=agent,
40
    tools=tools,
41
    verbose=True,
42
    max_iterations=10
43
)
44

45
# 6. 运行
46
result = agent_executor.invoke({
47
    "input": "What is 25 * 4? Then search for information about that number."
48
})
49

50
print(result["output"])

1
from langchain.memory import ConversationBufferMemory
2

3
# 创建记忆
4
memory = ConversationBufferMemory(
5
    memory_key="chat_history",
6
    return_messages=True
7
)
8

9
# 创建带记忆的 Agent
10
agent_executor = AgentExecutor(
11
    agent=agent,
12
    tools=tools,
13
    memory=memory,
14
    verbose=True
15
)
16

17
# 多轮对话
18
agent_executor.invoke({"input": "我的名字是张三"})
19
agent_executor.invoke({"input": "我的名字是什么？"})  # 能回忆起"张三"

1
from langchain.agents import BaseSingleActionAgent
2
from langchain.schema import AgentAction, AgentFinish
3

4
class CustomAgent(BaseSingleActionAgent):
5
    def plan(self, intermediate_steps, **kwargs):
6
        """决定下一步行动"""
7
        # 自定义决策逻辑
8
        user_input = kwargs["input"]
9

10
        if "计算" in user_input:
11
            return AgentAction(
12
                tool="Calculator",
13
                tool_input=extract_expression(user_input),
14
                log="使用计算器"
15
            )
16
        elif "搜索" in user_input:
17
            return AgentAction(
18
                tool="Search",
19
                tool_input=extract_query(user_input),
20
                log="使用搜索"
21
            )
22
        else:
23
            return AgentFinish(
24
                return_values={"output": "我不确定如何处理这个请求"},
25
                log="无法处理"
26
            )
27

28
    async def aplan(self, intermediate_steps, **kwargs):
29
        """异步版本"""
30
        return self.plan(intermediate_steps, **kwargs)

1
class AutoGPT:
2
    def __init__(self, llm, tools, memory, max_iterations=25):
3
        self.llm = llm
4
        self.tools = tools
5
        self.memory = memory
6
        self.max_iterations = max_iterations
7

8
    def run(self, goal: str):
9
        """自主循环执行"""
10
        self.memory.add("Goal", goal)
11

12
        for i in range(self.max_iterations):
13
            # 1. 构建上下文
14
            context = self.build_context()
15

16
            # 2. LLM 决策
17
            action = self.llm.decide_next_action(context)
18

19
            # 3. 执行action
20
            if action["type"] == "FINISH":
21
                return action["result"]
22

23
            result = self.execute_action(action)
24

25
            # 4. 存入记忆
26
            self.memory.add(f"Action {i}", action)
27
            self.memory.add(f"Result {i}", result)
28

29
            # 5. 自我评估
30
            if self.should_stop(result):
31
                break
32

33
        return "达到最大迭代次数"
34

35
    def build_context(self) -> str:
36
        """构建提示上下文"""
37
        return f"""You are an autonomous agent working towards: {self.memory.get('Goal')}
38

39
Memory:
40
{self.memory.summarize()}
41

42
Available tools:
43
{self.tools.describe()}
44

45
What should you do next? Respond in JSON format:
46
{% raw %}{{
47
    "thought": "...",
48
    "action": "...",
49
    "action_input": "..."
50
}}{% endraw %}"""

1
class AgentMemory:
2
    """Agent 的记忆系统"""
3
    def __init__(self, max_items=100):
4
        self.short_term = deque(maxlen=max_items)
5
        self.long_term = {}  # 向量数据库
6

7
    def add(self, key: str, value: str):
8
        self.short_term.append((key, value))
9

10
    def summarize(self, last_n=10) -> str:
11
        """总结最近的记忆"""
12
        recent = list(self.short_term)[-last_n:]
13
        return "\n".join([f"{k}: {v}" for k, v in recent])
14

15
    def search(self, query: str, k=5) -> List[str]:
16
        """搜索相关记忆"""
17
        # 使用向量相似度搜索
18
        pass
19

20
class SelfCriticAgent:
21
    """带自我批评的 Agent"""
22
    def run_with_criticism(self, task: str) -> str:
23
        attempt = 0
24
        max_attempts = 3
25

26
        while attempt < max_attempts:
27
            # 执行任务
28
            result = self.execute(task)
29

30
            # 自我批评
31
            critique = self.critique(task, result)
32

33
            if critique["quality"] > 0.8:
34
                return result
35

36
            # 根据批评改进
37
            task = self.improve_based_on_critique(task, critique)
38
            attempt += 1
39

40
        return result
41

42
    def critique(self, task: str, result: str) -> Dict:
43
        """批评结果"""
44
        prompt = f"""Evaluate the following result:
45

46
Task: {task}
47
Result: {result}
48

49
Provide:
50
1. Quality score (0-1)
51
2. Issues found
52
3. Suggestions for improvement
53

54
JSON format:"""
55

56
        response = self.llm(prompt)
57
        return json.loads(response)

1
class MultiAgentSystem:
2
    def __init__(self):
3
        self.agents = {
4
            "产品经理": ProductManagerAgent(),
5
            "架构师": ArchitectAgent(),
6
            "工程师": EngineerAgent(),
7
            "测试": TesterAgent()
8
        }
9

10
    def run(self, project_description: str):
11
        """软件开发工作流"""
12
        # 1. 产品经理：需求分析
13
        requirements = self.agents["产品经理"].analyze(project_description)
14

15
        # 2. 架构师：设计架构
16
        architecture = self.agents["架构师"].design(requirements)
17

18
        # 3. 工程师：编写代码
19
        code = self.agents["工程师"].implement(architecture)
20

21
        # 4. 测试：测试代码
22
        test_results = self.agents["测试"].test(code)
23

24
        # 5. 如果测试失败，返回修改
25
        if not test_results["passed"]:
26
            code = self.agents["工程师"].fix(code, test_results["issues"])
27

28
        return code
29

30
class ProductManagerAgent:
31
    def analyze(self, description: str) -> Dict:
32
        """分析需求"""
33
        prompt = f"""As a product manager, analyze this project:
34

35
{description}
36

37
Provide:
38
1. Core features
39
2. User stories
40
3. Acceptance criteria
41

42
Format as JSON."""
43

44
        return json.loads(self.llm(prompt))

1
class DebateSystem:
2
    """多个 Agent 辩论得出最佳方案"""
3
    def __init__(self, agents: List[Agent], judge: Agent):
4
        self.agents = agents
5
        self.judge = judge
6

7
    def debate(self, topic: str, rounds=3) -> str:
8
        proposals = []
9

10
        for round_num in range(rounds):
11
            print(f"\n=== Round {round_num + 1} ===")
12

13
            round_proposals = []
14
            for agent in self.agents:
15
                # 每个 Agent 提出方案
16
                proposal = agent.propose(
17
                    topic,
18
                    previous_proposals=proposals
19
                )
20
                round_proposals.append(proposal)
21
                print(f"{agent.name}: {proposal}")
22

23
            proposals.extend(round_proposals)
24

25
        # 评委选出最佳方案
26
        best = self.judge.select_best(topic, proposals)
27
        return best

1
class AgentEvaluator:
2
    def evaluate(self, agent, test_cases):
3
        metrics = {
4
            "success_rate": 0,
5
            "avg_steps": 0,
6
            "avg_cost": 0,
7
            "avg_time": 0
8
        }
9

10
        for case in test_cases:
11
            result = agent.run(case["input"])
12

13
            # 成功率
14
            if self.check_correctness(result, case["expected"]):
15
                metrics["success_rate"] += 1
16

17
            # 步数
18
            metrics["avg_steps"] += agent.step_count
19

20
            # 成本（API 调用次数 × 价格）
21
            metrics["avg_cost"] += agent.total_cost
22

23
            # 时间
24
            metrics["avg_time"] += agent.execution_time
25

26
        # 计算平均值
27
        n = len(test_cases)
28
        metrics["success_rate"] /= n
29
        metrics["avg_steps"] /= n
30
        metrics["avg_cost"] /= n
31
        metrics["avg_time"] /= n
32

33
        return metrics

1
# 优化 Agent 的系统提示
2
system_prompts = [
3
    "You are a helpful assistant.",
4
    "You are an expert problem solver. Think step by step.",
5
    "You are a precise AI agent. Always verify your actions before executing."
6
]
7

8
best_prompt = None
9
best_score = 0
10

11
for prompt in system_prompts:
12
    agent = Agent(system_prompt=prompt)
13
    score = evaluator.evaluate(agent, test_cases)["success_rate"]
14

15
    if score > best_score:
16
        best_score = score
17
        best_prompt = prompt
18

19
print(f"Best prompt: {best_prompt} (score: {best_score})")