Use external Evaluators

1
from humanloop import Humanloop
2
from openai import OpenAI
3
from openai.types.chat.chat_completion_message import ChatCompletionMessage as Message
4
import wikipedia
5
import json
6
7
8
openai = OpenAI(api_key="ADD YOUR KEY HERE")
9
humanloop = Humanloop(api_key="ADD YOUR KEY HERE")
10
11
12
def search_wikipedia(query: str) -> dict:
13
    """Search Wikipedia to get up-to-date information for a query."""
14
    try:
15
        page = wikipedia.page(query)
16
        return {
17
            "title": page.title,
18
            "content": page.content,
19
            "url": page.url,
20
        }
21
    except Exception as _:
22
        return {
23
            "title": "",
24
            "content": "No results found",
25
            "url": "",
26
        }
27
28
29
def call_model(messages: list[Message]) -> Message:
30
    """Calls the model with the given messages"""
31
    system_message = {
32
        "role": "system",
33
        "content": (
34
            "You are an assistant that help to answer user questions. You should "
35
            "leverage wikipedia to answer questions so that the information is up to date. "
36
            "If the response from Wikipedia does not seem relevant, rephrase the question "
37
            "and call the tool again. Then finally respond to the user."
38
        ),
39
    }
40
    response = openai.chat.completions.create(
41
        model="gpt-4o",
42
        messages=[system_message] + messages,
43
        tools=[
44
            {
45
                "type": "function",
46
                "function": {
47
                    "name": "search_wikipedia",
48
                    "description": "Search the internet to get up to date answers for a query.",
49
                    "parameters": {
50
                        "type": "object",
51
                        "required": ["query"],
52
                        "properties": {
53
                            "query": {"type": "string"},
54
                        },
55
                        "additionalProperties": False,
56
                    },
57
                },
58
            }
59
        ],
60
    )
61
    return response.choices[0].message.to_dict(exclude_unset=False)
62
63
64
def call_agent(query: str) -> str:
65
    """Calls the main agent loop and returns the final result"""
66
    messages = [{"role": "user", "content": query}]
67
    # Retry for a relevant response 3 times at most
68
    for _ in range(3):
69
        response = call_model(messages)
70
        messages.append(response)
71
        if response["tool_calls"]:
72
            # Call wikipedia to get up-to-date information
73
            for tool_call in response["tool_calls"]:
74
                source = search_wikipedia(
75
                    **json.loads(tool_call["function"]["arguments"])
76
                )
77
                messages.append(
78
                    {
79
                        "role": "tool",
80
                        "content": json.dumps(source),
81
                        "tool_call_id": tool_call["id"],
82
                    }
83
                )
84
        else:
85
            # Respond to the user
86
            return response["content"]
87
88
89
if __name__ == "__main__":
90
    result = call_agent("Where does the sun go at night?")
91
    print(result)

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