Spaces:
Sleeping
Sleeping
| # app/agent_executor.py | |
| import json | |
| from langchain.agents import initialize_agent, AgentType | |
| from app.recommend_tool import RecommendTool | |
| from app.generateRecom import RecommendToolGeneration | |
| from app.llm_client import LLMClient | |
| class AgentExecutor: | |
| def __init__(self, api_key: str): | |
| llm_client = LLMClient(api_key) | |
| rec_tool = RecommendTool(llm_client).tool | |
| self.agent = initialize_agent( | |
| tools=[rec_tool], | |
| llm=llm_client.llm, | |
| agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, | |
| verbose=False | |
| ) | |
| gen_tool = RecommendToolGeneration(llm_client).tool | |
| self.agent = initialize_agent( | |
| tools=[rec_tool, gen_tool], | |
| llm=llm_client.llm, | |
| agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, | |
| verbose=False | |
| ) | |
| def run_completion(self, data: dict) -> dict: | |
| prompt = f"Tool: recommend_missing\nInput: {json.dumps(data, ensure_ascii=False)}" | |
| return self._call_agent(prompt) | |
| def run_generation(self, data: dict) -> dict: | |
| prompt = f"Tool: generate_recommendation_sections\nInput: {json.dumps(data, ensure_ascii=False)}" | |
| return self._call_agent(prompt) | |
| def _call_agent(self, prompt: str) -> dict: | |
| result_str = self.agent.run(input=prompt) | |
| result_str=result_str.replace("```json\n", "").replace("```", "").strip() | |
| try: | |
| result = json.loads(result_str) # convertir la string JSON en dict Python | |
| except json.JSONDecodeError: | |
| return { | |
| "error": "Invalid JSON returned", | |
| "raw": result_str | |
| } | |
| return result # ce dict sera validé contre response_model | |