File size: 4,616 Bytes
62da328 |
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 |
# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
from typing import Optional
from camel.agents.chat_agent import ChatAgent
from camel.messages import BaseMessage
from camel.models import BaseModelBackend
from camel.prompts import TextPrompt
from camel.types import RoleType
from camel.utils import create_chunks
# AgentOps decorator setting
try:
import os
if os.getenv("AGENTOPS_API_KEY") is not None:
from agentops import track_agent
else:
raise ImportError
except (ImportError, AttributeError):
from camel.utils import track_agent
@track_agent(name="SearchAgent")
class SearchAgent(ChatAgent):
r"""An agent that summarizes text based on a query and evaluates the
relevance of an answer.
Args:
model (BaseModelBackend, optional): The model backend to use for
generating responses. (default: :obj:`OpenAIModel` with
`GPT_4O_MINI`)
"""
def __init__(
self,
model: Optional[BaseModelBackend] = None,
) -> None:
system_message = BaseMessage(
role_name="Assistant",
role_type=RoleType.ASSISTANT,
meta_dict=None,
content="You are a helpful assistant.",
)
super().__init__(system_message, model=model)
def summarize_text(self, text: str, query: str) -> str:
r"""Summarize the information from the text, base on the query.
Args:
text (str): Text to summarize.
query (str): What information you want.
Returns:
str: Strings with information.
"""
self.reset()
summary_prompt = TextPrompt(
'''Gather information from this text that relative to the
question, but do not directly answer the question.\nquestion:
{query}\ntext '''
)
summary_prompt = summary_prompt.format(query=query)
# Max length of each chunk
max_len = 3000
results = ""
chunks = create_chunks(text, max_len)
# Summarize
for i, chunk in enumerate(chunks, start=1):
prompt = summary_prompt + str(i) + ": " + chunk
user_msg = BaseMessage.make_user_message(
role_name="User",
content=prompt,
)
result = self.step(user_msg).msg.content
results += result + "\n"
# Final summarization
final_prompt = TextPrompt(
'''Here are some summarized texts which split from one text. Using
the information to answer the question. If can't find the answer,
you must answer "I can not find the answer to the query" and
explain why.\n Query:\n{query}.\n\nText:\n'''
)
final_prompt = final_prompt.format(query=query)
prompt = final_prompt + results
user_msg = BaseMessage.make_user_message(
role_name="User",
content=prompt,
)
response = self.step(user_msg).msg.content
return response
def continue_search(self, query: str, answer: str) -> bool:
r"""Ask whether to continue search or not based on the provided answer.
Args:
query (str): The question.
answer (str): The answer to the question.
Returns:
bool: `True` if the user want to continue search, `False`
otherwise.
"""
prompt = TextPrompt(
"Do you think the ANSWER can answer the QUERY? "
"Use only 'yes' or 'no' to answer.\n"
"===== QUERY =====\n{query}\n\n"
"===== ANSWER =====\n{answer}"
)
prompt = prompt.format(query=query, answer=answer)
user_msg = BaseMessage.make_user_message(
role_name="User",
content=prompt,
)
response = self.step(user_msg).msg.content
if "yes" in str(response).lower():
return False
return True
|