Spaces:
Sleeping
Sleeping
| from talentum_score.const import PROMPTS_PATH | |
| from talentum_score.domain import ( | |
| CompanyAnalysisToParse, | |
| CompanyEvaluation, | |
| PerplexitySource, | |
| ) | |
| from talentum_score.nodes.web_search import WebSearchNode | |
| def parse_company_evaluation(output: dict) -> CompanyAnalysisToParse: | |
| parsed_output = output["parsed"] | |
| raw_output = output["raw"] | |
| if output["parsing_error"] is not None: | |
| return CompanyEvaluation( | |
| analysis=raw_output.content, | |
| score=None, | |
| sources=None, | |
| citations=None, | |
| ) | |
| sources = [ | |
| PerplexitySource.model_validate(source) | |
| for source in raw_output.additional_kwargs.get("search_results", []) | |
| ] | |
| return CompanyEvaluation( | |
| analysis=parsed_output.analysis, | |
| score=parsed_output.score, | |
| sources=sources, | |
| citations=raw_output.additional_kwargs.get("citations", []), | |
| ) | |
| class CompanySearchNode(WebSearchNode): | |
| def __init__(self, output_key: str = "company_analysis"): | |
| super().__init__( | |
| system_prompt_path=PROMPTS_PATH / "search_company_info_system.txt", | |
| query="Company Name: {state.flow_input.company_name}", | |
| structured_output=CompanyAnalysisToParse, | |
| output_key=output_key, | |
| parse_output_fn=parse_company_evaluation, | |
| model="sonar", | |
| ) | |