File size: 4,549 Bytes
0c2d302
 
f6a6ef0
 
4c3f84c
0c2d302
 
 
 
 
 
 
 
 
 
 
 
 
11d325c
0c2d302
f6a6ef0
0c2d302
 
 
4c3f84c
0c2d302
 
 
 
 
 
 
 
 
 
 
 
 
4c3f84c
0c2d302
 
 
 
 
 
 
 
11d325c
0c2d302
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
f6a6ef0
4129aa0
 
0c2d302
 
 
 
d22c13b
0c2d302
 
 
 
f6a6ef0
4c3f84c
0c2d302
 
 
 
 
 
 
 
 
 
 
 
 
 
 
f6a6ef0
4c3f84c
0c2d302
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
4c3f84c
0c2d302
 
 
 
582c31b
0c2d302
 
 
5f073d9
0c2d302
 
b38c2ca
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
from pathlib import Path

# from dotenv import find_dotenv, load_dotenv
import os
from huggingface_hub import InferenceClient, login
from smolagents import (
    CodeAgent,
    DuckDuckGoSearchTool,
    FinalAnswerTool,
    InferenceClientModel,
    PythonInterpreterTool,
    Tool,
    ToolCallingAgent,
    VisitWebpageTool,
    WikipediaSearchTool,
)

from src.insurance_assistants.complex_rag import RAG
from src.insurance_assistants.consts import PROMPT_PREFIX, VLM_PROMPT

# _ = load_dotenv(dotenv_path=find_dotenv())
rag_app = RAG()
# FIXME Comment the following if you want to reprocess everything
rag_app.vectordb_id = "policy_wordings"
login(os.getenv("HF_TOKEN"))

class InsuranceInfoRetriever(Tool):
    name = "InsuranceInfoRetriever"
    description = "Retrieves information from insurance documents."
    inputs = {
        "query": {"type": "string", "description": "The query to search for."},
    }
    output_type = "string"

    def forward(self, query: str) -> str:
        client = InferenceClient(
            provider="hyperbolic",
            bill_to="VitalNest",
            # token=os.getenv("HF_TOKEN")
        )
        results = rag_app.search_documents(query)
        img_paths = [Path(res[0]) for res in results]

        grouped_images = [rag_app.encode_image_to_base64(pth) for pth in img_paths]
        chat_template = [
            {
                "role": "system",
                "content": VLM_PROMPT,
            },
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {"url": f"data:image/jpeg;base64,{image}"},
                    }
                    for image in grouped_images
                ]
                + [{"type": "text", "text": query}],
            },
        ]
        completion = client.chat.completions.create(
            model="Qwen/Qwen2.5-VL-72B-Instruct",
            messages=chat_template,
            temperature=0.1,
            max_tokens=10_000,
        )
        answer = completion.choices[0].message.content
        if answer:
            answer += f"The information was retrived from the following documents: {img_paths}"
        return answer if answer else ""


insurance_agent = CodeAgent(
    tools=[InsuranceInfoRetriever(), FinalAnswerTool()],
    model=InferenceClientModel(bill_to="VitalNest", temperature=0.1,
                              # token=os.getenv("HF_TOKEN")
                              ),
    additional_authorized_imports=["os", "requests", "bs4", "pil", "base64", "io"],
    max_steps=1,
    verbosity_level=-1,
    name="insurance_agent",
    description="You answer health insurance questions based on the `InsuranceInfoRetriever` "
    "tool. All health insurance questions must be answered by you.",
)
websearch_agent = ToolCallingAgent(
    model=InferenceClientModel(
        model_id="Qwen/Qwen3-30B-A3B", bill_to="VitalNest", temperature=0.1,
        # token=os.getenv("HF_TOKEN")
    ),
    tools=[
        VisitWebpageTool(max_output_length=20_000),
        DuckDuckGoSearchTool(max_results=5),
        FinalAnswerTool(),
    ],
    max_steps=4,
    verbosity_level=-1,
    name="web_search_agent",
    planning_interval=2,
    description="Searches the web with a particular query.",
)

wikipedia_agent = ToolCallingAgent(
    model=InferenceClientModel(
        model_id="Qwen/Qwen3-30B-A3B", bill_to="VitalNest", temperature=0.1,
        # token=os.getenv("HF_TOKEN")
    ),
    tools=[
        WikipediaSearchTool(user_agent="WikiAssistant (merlin@example.com)"),
        FinalAnswerTool(),
    ],
    max_steps=3,
    verbosity_level=-1,
    name="wikipedia_agent",
    description="Searches Wikipedia for a topic.",
)

manager_agent = CodeAgent(
    tools=[FinalAnswerTool(), PythonInterpreterTool()],
    additional_authorized_imports=["os"],
    model=InferenceClientModel(
        model_id="Qwen/Qwen3-235B-A22B",
        bill_to="VitalNest",
        temperature=0.1,
        # token=os.getenv("HF_TOKEN")
    ),
    managed_agents=[websearch_agent, wikipedia_agent, insurance_agent],
    max_steps=10,
    planning_interval=2,
    verbosity_level=0,
    add_base_tools=True,
    name="Versatile_Multi_Agent",
    description="Answer health insurance related questions from pre-defined set of "
    "health insurance documents using the `insurance_agent` team member, search wikipedia and the web for general information.",
)
manager_agent.system_prompt = manager_agent.system_prompt + PROMPT_PREFIX