Evaluate an agent

1
import json
2
import os
3
4
from humanloop import Humanloop
5
from openai import OpenAI
6
7
import wikipedia
8
9
10
openai = OpenAI(api_key="YOUR_OPENAI_KEY")
11
humanloop = Humanloop(api_key="YOUR_HUMANLOOP_KEY")
12
13
14
def search_wikipedia(query: str) -> dict:
15
    """Search the internet to get data sources for a query."""
16
17
    try:
18
        # Let Wikipedia suggest a relevant page
19
        page = wikipedia.page(query)
20
        return {
21
            "title": page.title,
22
            "content": page.content,
23
            "url": page.url,
24
        }
25
    except Exception:
26
        return {
27
            "title": "",
28
            "content": "No results found",
29
            "url": "",
30
        }
31
32
33
TOOL_SCHEMA = {
34
    "type": "function",
35
    "function": {
36
        "name": "search_wikipedia",
37
        "description": "Search the internet to get data sources for a query.",
38
        "parameters": {
39
            "type": "object",
40
            "required": ["query"],
41
            "properties": {
42
                "query": {"type": "string"},
43
            },
44
            "additionalProperties": False,
45
        },
46
    },
47
}
48
49
50
def agent(messages: list[dict], tool_call: bool) -> str:
51
    args = {
52
        "messages": messages,
53
        "model": "gpt-4o",
54
        "temperature": 0,
55
    }
56
    if tool_call:
57
        args["tools"] = [TOOL_SCHEMA]
58
    response = openai.chat.completions.create(**args)
59
    return {
60
        "content": response.choices[0].message.content,
61
        "tool_calls": [
62
            tool_call.to_dict()
63
            for tool_call in response.choices[0].message.tool_calls or []
64
        ],
65
    }
66
67
68
def workflow(question: str) -> str:
69
    source = None
70
    messages = [
71
        {
72
            "role": "system",
73
            "content": (
74
                "You must find a good source to answer a "
75
                "question using the provided Wikipedia tool."
76
            ),
77
        },
78
        {
79
            "role": "user",
80
            "content": question,
81
        },
82
    ]
83
    # Maximum 3 attempts to find a relevant source
84
    for _ in range(3):
85
        response = agent(messages, tool_call=True)
86
87
        if "FINISHED" in (response["content"] or "") or not response["tool_calls"]:
88
            # Model chose a source to answer the question
89
            break
90
91
        tool_call = response["tool_calls"][0]
92
        arguments = json.loads(tool_call["function"]["arguments"])
93
        source = search_wikipedia(**arguments)
94
        messages.extend(
95
            [
96
                {
97
                    "role": "assistant",
98
                    "content": f"Found a source called {source['title']}",
99
                },
100
                {
101
                    "role": "user",
102
                    "content": (
103
                        "Is this a relevant source? Output 'FINISHED' "
104
                        "or rephrase the question to essential "
105
                        "terms and call the tool again to get a "
106
                        "new information source."
107
                    ),
108
                },
109
            ]
110
        )
111
112
    source = source or {
113
        "title": "",
114
        "content": "No relevant source found",
115
        "url": "",
116
    }
117
    answer = agent(
118
        messages=[
119
            {
120
                "role": "system",
121
                "content": (
122
                    "You are a summarizer. Answer "
123
                    "the question based on this source: "
124
                    f"{source['content']}"
125
                ),
126
            },
127
            {
128
                "role": "user",
129
                "content": question,
130
            },
131
        ],
132
        tool_call=False,
133
    )["content"]
134
    return f"{answer}\n\nSource: {source['url']}"

1
def workflow(question: str) -> str:
2
    # ...
3
4
    answer = agent(
5
        messages=[
6
            {
7
                "role": "system",
8
                "content": (
9
                    "You are a summarizer. Answer questions "
10
                    "so a 5-year old can understand. Answer "
11
                    "the question based on this source: "
12
                    f"{source['content']}"
13
                ),
14
            },
15
            {
16
                "role": "user",
17
                "content": question,
18
            },
19
        ],
20
        tool_call=False,
21
    )["content"]
22
    return f"{answer}\n\nSource: {source['url']}"

$
python agent.py
>
>
Flow Version ID: flv_9ECTrfeZYno2OIj9KAqlz
>
Run ID: rn_WnIwPSI7JFKEtwTS0l3mj
>
>
Navigate to your Evaluation:
>
https://app.humanloop.com/project/fl_9CCIoTzySPfUFeIxfYE6g/evaluations/rn_WnIwPSI7JFKEtwTS0l3mj/stats
>
>
Running 'Orchestrator' over the Dataset 'Children Questions' using 8 workers 
>
[######################------------------] 34/50 (68.00%) | ETA: 14
>
>
...
>
>
+------------------------+---------------------+---------------------+
>
|                        |       Control       |        Latest       |
>
+------------------------+---------------------+---------------------+
>
|                 Run ID |        67tEc        |        WnIwP        |
>
+------------------------+---------------------+---------------------+
>
|             Version ID |         None        |         None        |
>
+------------------------+---------------------+---------------------+
>
|                  Added | 2024-11-19 22:05:17 | 2024-11-19 22:24:13 |
>
+------------------------+---------------------+---------------------+
>
|             Evaluators |                     |                     |
>
+------------------------+---------------------+---------------------+
>
| QA Agent/Comprehension |         3.24        |         8.04        |
>
+------------------------+---------------------+---------------------+
>
>
Change of [4.80] for Evaluator QA Agent/Comprehension

1
@humanloop.tool(path="QA Agent/Search Wikipedia")
2
def search_wikipedia(query: str) -> dict:
3
    ...
4
5
@humanloop.prompt(path="QA Agent/Agent")
6
def agent(messages: list[dict], tool_call: bool) -> str:
7
    ...
8
9
@humanloop.flow(path="QA Agent/Workflow")
10
def workflow(question: str) -> str:
11
    ...