recap.agents

Submodules

recap.agents.analytics

class recap.agents.analytics.AnalyticsAgent(model, artifact_dir='./artifacts')

Bases: BaseAgent

Agent responsible for performing analytical operations on retrieved review data.

This agent uses an LLM to generate an analytics plan from a natural language query, then executes that plan against records loaded from retrieval artifacts. Supported operations include grouping, aggregation, sorting, value counting, and selecting the top results.

args_schema

alias of AnalyticsAgentInput

description: str = 'The analytics agent is a tool used to derive analytics on a set of retrieval artifact_refs. It answers questionsrequiring grouping, aggregration (mean,max,min,sum), top_k or sorting of columns of the provided results. Note retrievalsshould be broad enough to allow analytic tool to analyze'

execute_plan(plan: AnalyticsPlan, df: DataFrame) → DataFrame

Given a pandas DataFrame and AnaltyicsPlan execute plan and return the resulting dataframe

Parameters:

• df – DataFrame to run analytics on

• plan – AnalyticsPlan

Returns:

Pandas DataFrame with executed analytics

generate_plan(query: str, df: DataFrame) → AnalyticsPlan

invoke(query: str, artifact_refs: list[str]) → str

Given a query and a set of artifact_refs run analysis on the provided artifacts (groupby, average, top_k,e tc.)

Parameters:

• summarize (query - List of string contents to)

• use (artifact_refs - a list of retrieval references to)

Returns:

Str of the JSON with analytics plan and results

name: str = 'analytics'

class recap.agents.analytics.AnalyticsAgentInput(*, query: str = 'This is a user query ', artifact_refs: list[str])

Bases: BaseModel

artifact_refs: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

query: str

class recap.agents.analytics.AnalyticsPlan(operation: Literal['group_agg', 'column_agg', 'sort', 'value_counts'], group_by: list[str] | None = None, agg_column: str | None = None, aggregation: Literal['sum', 'mean', 'count', 'min', 'max'] | None = None, sort_column: str | None = None, sort_desc: bool = True, top_k: int | None = None, column: str | None = None, normalize: bool = False)

Bases: object

agg_column: str | None = None

aggregation: Literal['sum', 'mean', 'count', 'min', 'max'] | None = None

column: str | None = None

group_by: list[str] | None = None

normalize: bool = False

operation: Literal['group_agg', 'column_agg', 'sort', 'value_counts']

sort_column: str | None = None

sort_desc: bool = True

top_k: int | None = None

recap.agents.base

class recap.agents.base.BaseAgent(model: BaseChatModel | None, tools: list[StructuredTool] | None = None, system_prompt: str | None = None)

Bases: object

Base class for all agents in the application.

This class provides the common infrastructure for agent implementations, including LangChain agent creation, prompt loading, tool conversion, JSON response parsing, and graph visualization. Subclasses should define the agent’s metadata (name, description, and optionally args_schema) and typically override invoke() to implement task-specific behavior.

The class also provides a method for easily converting agents into structured tools for use by other agents using the as_tool() function

args_schema: type[BaseModel] | None = None

as_tool() → StructuredTool

Wrap this agent’s .run() as a LangChain StructuredTool. The orchestrator calls this and gets back a plain tool — it has no knowledge of the class underneath.

Returns:

Structured Tool for the agents invoke function

description: str = ''

generate_agent_graph(output_file: str = 'output/agent_graph.png')

Save a mermaid .png file of the agent graph to specified output_file

Parameters:

output_file – File path for png output. Defaults to “output/agent_graph.png”

invoke(*args, **kwargs) → str

This is generic implementation of the invoke method for agent base class. It will call the underlying Langchain agent by default and return message history.

This should be overwritten by sub agents with their specific functionality

static load_prompt(prompt_name: str) → str

Load a prompt from the prompts directory.

Parameters:

prompt_name – Name of the prompt file (without .md extension)

Returns:

str - Prompt content

name: str = ''

parse_json_response(content: str) → Tuple[dict | None, str]

Helper function for agents to be able to parse inline json from response. This allows for more durable parsing of LLM responses in varied formats

Parameters:

content – str response from LLM

Returns:

Tuple with first element being parsed json if found second being json as str

recap.agents.orchestrator

class recap.agents.orchestrator.Orchestrator(model: BaseChatModel | None, tools: list[StructuredTool] | None = None, system_prompt: str | None = None)

Bases: BaseAgent

This is orchestrator agent. It utilizes the prompt in prompts/orchestrator.md and is used as user facing agent that invokes sub agents to answer user queries

name: str = 'orchestrator'

recap.agents.retrieval

class recap.agents.retrieval.ChromaRetrievalAgent(model, dir: str = './chroma', semantic_column='review', artifact_dir='./artifacts')

Bases: BaseAgent

Agent responsible for retrieving paper review records from a ChromaDB database.

This agent uses an LLM to translate natural language queries into semantic search queries, metadata filters, or a combination of both. Retrieved records are returned directly for small result sets, while larger result sets are written to an artifact file with a sample included in the response.

In order to use this class with a different dataset update the schema functonality

args_schema

alias of ChromaRetrievalAgentInput

description: str = 'Searches the MRED paper review dataset. It can do both semantic search on content as well as filterspecific paper(s) based on metadata criteria. The number of records is returned in record_count field.If more than 10 are retrieved agent will return reference to artifact holding all records and sample of 10 records.artifact_ref should be used to answer questions refering to whole set, top_10 would just be examples'

generate_plan(query) → ChromaRetrievalPlan

Given a natural language query use LLM to generate a retrieval plan. The plan will contain either a semantic query, chroma filter statement or both to be executed on the database

Parameters:

query – Natural language query to run on the database

Returns:

ChromaRetrievalPlan object

invoke(query: str) → str

Given a natural language query return relevant documents from the chroma database. This supports both metadata filtering and semantic search over column sepcified

Parameters:

with (query - natural language query to search database)

Returns:

Str of the JSON reports objects

name: str = 'retrieval'

class recap.agents.retrieval.ChromaRetrievalAgentInput(*, query: str)

Bases: BaseModel

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

query: str

class recap.agents.retrieval.ChromaRetrievalPlan(filter_statement: dict | None, semantic_query: str | None, query_name: str, explanation: str)

Bases: object

explanation: str

filter_statement: dict | None

query_name: str

semantic_query: str | None

class recap.agents.retrieval.RetrievedRecords(filter_statement: dict | None, semantic_query: str | None, record_count: int, top_10: list, artifact_ref: str | None)

Bases: object

artifact_ref: str | None

filter_statement: dict | None

record_count: int

semantic_query: str | None

top_10: list

recap.agents.summary

class recap.agents.retrieval.ChromaRetrievalAgent(model, dir: str = './chroma', semantic_column='review', artifact_dir='./artifacts')

Bases: BaseAgent

Agent responsible for retrieving paper review records from a ChromaDB database.

This agent uses an LLM to translate natural language queries into semantic search queries, metadata filters, or a combination of both. Retrieved records are returned directly for small result sets, while larger result sets are written to an artifact file with a sample included in the response.

In order to use this class with a different dataset update the schema functonality

args_schema

alias of ChromaRetrievalAgentInput

description: str = 'Searches the MRED paper review dataset. It can do both semantic search on content as well as filterspecific paper(s) based on metadata criteria. The number of records is returned in record_count field.If more than 10 are retrieved agent will return reference to artifact holding all records and sample of 10 records.artifact_ref should be used to answer questions refering to whole set, top_10 would just be examples'

generate_plan(query) → ChromaRetrievalPlan

Given a natural language query use LLM to generate a retrieval plan. The plan will contain either a semantic query, chroma filter statement or both to be executed on the database

Parameters:

query – Natural language query to run on the database

Returns:

ChromaRetrievalPlan object

invoke(query: str) → str

Given a natural language query return relevant documents from the chroma database. This supports both metadata filtering and semantic search over column sepcified

Parameters:

with (query - natural language query to search database)

Returns:

Str of the JSON reports objects

name: str = 'retrieval'

class recap.agents.retrieval.ChromaRetrievalAgentInput(*, query: str)

Bases: BaseModel

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

query: str

class recap.agents.retrieval.ChromaRetrievalPlan(filter_statement: dict | None, semantic_query: str | None, query_name: str, explanation: str)

Bases: object

explanation: str

filter_statement: dict | None

query_name: str

semantic_query: str | None

class recap.agents.retrieval.RetrievedRecords(filter_statement: dict | None, semantic_query: str | None, record_count: int, top_10: list, artifact_ref: str | None)

Bases: object

artifact_ref: str | None

filter_statement: dict | None

record_count: int

semantic_query: str | None

top_10: list

recap.agents.visualization

class recap.agents.visualization.VisualizationAgent(model, artifact_dir='./artifacts')

Bases: BaseAgent

Agent responsible for generating Chart.js specifications from structured data.

This agent accepts data, a supported chart type, and optional metadata such as a title and description. It uses an LLM to generate a Chart.js configuration that can be rendered by the frontend.

args_schema

alias of VisualizationAgentInput

description: str = 'The visualzation agent is a tool used to generate graphics. It takes in data, chart type, title and descriptionand returns a json chart specfication to be used by the frontend to render a chart. Commonly called after analyticsagent. Do NOT make up data to pass to tool. Note if you need to refer to the chart in response it is rendered below the message'

generate_chart_js_spec(data: dict[str, Any], chart_type: Literal['pie', 'bar'], title: str | None = None, description: str | None = None) → dict

invoke(data: dict[str, Any], chart_type: Literal['pie', 'bar'], title: str | None = None, description: str | None = None) → str

Given data, chart type, title and description genereate json string for chart js specification.

Parameters:

• chart (data - Dictionary of the data to use in)

• (Options (chart_type - The type of chart to generate) – pie, bar)

• generate (description - Optionally description for the chart to)

• generate

Returns:

Str of the JSON with chart specification

name: str = 'visualization'

class recap.agents.visualization.VisualizationAgentInput(*, data: dict[str, Any], chart_type: Literal['pie', 'bar'], title: str | None = None, description: str | None = None)

Bases: BaseModel

chart_type: Literal['pie', 'bar']

data: dict[str, Any]

description: str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

title: str | None

Module contents