Spaces:
Sleeping
Sleeping
| from llama_index.core.agent.workflow import AgentWorkflow | |
| from llama_index.core.workflow import Context | |
| from llama_index.core.tools import FunctionTool | |
| from llama_index.llms.huggingface_api import HuggingFaceInferenceAPI | |
| from llama_index.tools.duckduckgo import DuckDuckGoSearchToolSpec | |
| from llama_index.tools.wikipedia import WikipediaToolSpec | |
| from llama_index.core.tools.tool_spec.load_and_search import LoadAndSearchToolSpec | |
| from llama_index.readers.web import SimpleWebPageReader | |
| from llama_index.core.tools.ondemand_loader_tool import OnDemandLoaderTool | |
| from langfuse.llama_index import LlamaIndexInstrumentor | |
| from llama_index.llms.ollama import Ollama | |
| from llama_index.core.agent.workflow import ReActAgent, FunctionAgent | |
| class BasicAgent: | |
| def __init__(self, ollama=False, langfuse=True): | |
| if not ollama: | |
| llm = HuggingFaceInferenceAPI(model_name="Qwen/Qwen2.5-Coder-32B-Instruct") | |
| else: | |
| llm = Ollama(model="mistral:latest", request_timeout=120.0) | |
| # Langfuse | |
| self.langfuse = langfuse | |
| if self.langfuse: | |
| self.instrumentor = LlamaIndexInstrumentor() | |
| self.instrumentor.start() | |
| # Initialize tools | |
| tool_spec = DuckDuckGoSearchToolSpec() | |
| search_tool = FunctionTool.from_defaults(tool_spec.duckduckgo_full_search) | |
| wiki_spec = WikipediaToolSpec() | |
| wiki_search_tool = wiki_spec.to_tool_list()[1] | |
| # Convert into a LoadAndSearchToolSpec because the wikipedia search tool returns | |
| # entire Wikipedia pages and this can pollute the context window of the LLM | |
| wiki_spec = WikipediaToolSpec() | |
| wiki_search_tool = wiki_spec.to_tool_list()[1] | |
| # Convert into a LoadAndSearchToolSpec because the wikipedia search tool returns | |
| # entire Wikipedia pages and this can pollute the context window of the LLM | |
| # TODO this does not work so well. We need to make the retriever return the top 5 chunks or sth. | |
| wiki_search_tool_las = LoadAndSearchToolSpec.from_defaults(wiki_search_tool).to_tool_list() | |
| webpage_tool = OnDemandLoaderTool.from_defaults( | |
| SimpleWebPageReader(html_to_text=True), | |
| name="Webpage search tool", | |
| description="A tool for loading the content of a webpage and querying it for information", | |
| ) | |
| self.agent = AgentWorkflow.from_tools_or_functions( # ReActAgent( | |
| tools=[search_tool], # webpage_tool does not work properly - cookies etc | |
| llm=llm, | |
| verbose=True, | |
| system_prompt = ( | |
| "You are a general AI assistant. I will ask you a question. " | |
| "Report your thoughts, and finish your answer with the following template: " | |
| "FINAL ANSWER: [YOUR FINAL ANSWER]. YOUR FINAL ANSWER should be a number " | |
| "OR as few words as possible OR a comma separated list of numbers and/or " | |
| "strings. If you are asked for a number, don't use comma to write your " | |
| "number neither use units such as $ or percent sign unless specified otherwise. " | |
| "If you are asked for a string, don't use articles, neither abbreviations (e.g. " | |
| "for cities), and write the digits in plain text unless specified otherwise. If " | |
| "you are asked for a comma separated list, apply the above rules depending of " | |
| "whether the element to be put in the list is a number or a string." | |
| ) | |
| ) | |
| # self.ctx = Context(self.agent) | |
| async def __call__(self, question: str) -> str: | |
| response = await self.agent.run(user_msg=question) # ctx=self.ctx) | |
| if self.langfuse: | |
| self.instrumentor.flush() | |
| return response.response.content.replace("FINAL ANSWER:", "").strip() | |